#!/usr/bin/env python ''' TexMagPy - a change up of Paul's code for testing on my system. Watt Midnite Solar Forum if you have questions, holler. I am NOT responsible for mising or confused bits, bytes or electrons. Magic smoke is on you. ''' ''' MagPy - an application to interface with Magnum inverters The application decodes the MagNet data protocol to display data for all devices connected and active. MagPy, the other sort https://en.wikipedia.org/wiki/Australian_magpie This program was initially inspired by Chris (aka user cpfl) Midnightsolar forum. Thank you Chris for leading me into Python :) Created 13 Jun 2015 Modified 19 Jun 2105 @author: Paul Alting van Geusau ''' import serial import time import array import argparse class inverter_proto(): """definition of inverter packet """ # def __init__(self): status_descript = "NA" # status descriptive word: fault_descript = "NA" # fault descriptive word: status_code = '0x00' # 0 packet_buffer[0] fault_code = '0x00' # 1 packet_buffer[1] volts_dc = 0.0 # 2 packet_buffer[2] HIGH BYTE # 3 packet_buffer[3] LOW BYTE amps_dc = 0.0 # 4 packet_buffer[4] HIGH BYTE # 5 packet_buffer[5] LOW BYTE volts_ac_out = 0 # 6 packet_buffer[6] volts_ac_in = 0 # 7 packet_buffer[7] LED_inverter = 0 # 8 packet_buffer[8] LED_charger = 0 # 9 packet_buffer[9] revision = 0.0 # 10 packet_buffer[10] temp_battery = 0 # 11 packet_buffer[11] temp_transformer = 0 # 12 packet_buffer[12] temp_FET = 0 # 13 packet_buffer[13] model_id = 0 # 14 packet_buffer[14] stack_mode = "NA" # 15 packet_buffer[15] amps_ac_in = 0 # 16 packet_buffer[16] amps_ac_out = 0 # 17 packet_buffer[17] frequency_ac_out = 0.0 # 18 packet_buffer[18] HIGH BYTE # 19 packet_buffer[19] LOW BYTE # ALWAYS 0 # 20 packet_buffer[20] # ALWAYS 0xFF # 21 packet_buffer[21] # END # # inverter decode function: def decode(self, packet_buffer): # print("self.status_code= ", self.status_code) global system_bus_volts self.status_code = hex(packet_buffer[0]) if (self.status_code == '0x0') : self.status_descript = "Charger Standby" if (self.status_code == '0x1') : self.status_descript = "Equalise" if (self.status_code == '0x2') : self.status_descript = "Float" if (self.status_code == '0x4') : self.status_descript = "Absorb" if (self.status_code == '0x8') : self.status_descript = "Bulk" if (self.status_code == '0x9') : self.status_descript = "Battery Saver" if (self.status_code == '0x10') : self.status_descript = "Charge" if (self.status_code == '0x20') : self.status_descript = "Off" if (self.status_code == '0x40') : self.status_descript = "Invert" if (self.status_code == '0x50') : self.status_descript = "Standby" if (self.status_code == '0x60') : self.status_descript = "Search" self.fault_code = hex(packet_buffer[1]) if (self.fault_code == '0x0') : self.fault_descript = "No Faults" if (self.fault_code == '0x1') : self.fault_descript = "Stuck Relay" if (self.fault_code == '0x2') : self.fault_descript = "DC Overload" if (self.fault_code == '0x3') : self.fault_descript = "AC Overload" if (self.fault_code == '0x4') : self.fault_descript = "Dead Battery" if (self.fault_code == '0x5') : self.fault_descript = "AC Backfeed" if (self.fault_code == '0x8') : self.fault_descript = "Low Battery Cutout" if (self.fault_code == '0x9') : self.fault_descript = "High Battery Cutout" if (self.fault_code == '0xA') : self.fault_descript = "High AC Output Volts" if (self.fault_code == '0x10') : self.fault_descript = "Bad FET Bridge" if (self.fault_code == '0x12') : self.fault_descript = "FETs Over Temperature" if (self.fault_code == '0x13') : self.fault_descript = "FETs Over Temperature Quick" if (self.fault_code == '0x14') : self.fault_descript = "Internal Fault #4" if (self.fault_code == '0x16') : self.fault_descript = "Stacker Mode Fault" if (self.fault_code == '0x18') : self.fault_descript = "Stacker Sync Clock Out of Phase" if (self.fault_code == '0x17') : self.fault_descript = "Stacker Sync Clock Lost" if (self.fault_code == '0x19') : self.fault_descript = "Stacker AC Phase Fault" if (self.fault_code == '0x20') : self.fault_descript = "Over temperature Shutdown" if (self.fault_code == '0x21') : self.fault_descript = "Transfer Relay Fault" if (self.fault_code == '0x80') : self.fault_descript = "Charger Fault" if (self.fault_code == '0x81') : self.fault_descript = "Battery Temperature High" if (self.fault_code == '0x90') : self.fault_descript = "Transformer Temperature Cutout Open" if (self.fault_code == '0x91') : self.fault_descript = "AC Breaker CB3 Tripped" self.volts_dc = ((packet_buffer[2] * 256) + packet_buffer[3]) / 10.0 self.amps_dc = ((packet_buffer[4] * 256) + packet_buffer[5]) self.volts_ac_out = packet_buffer[6] self.volts_ac_in = packet_buffer[7] self.revision = packet_buffer[10] / 10.0 self.temp_battery = packet_buffer[11] self.temp_transformer = packet_buffer[12] self.temp_FET = packet_buffer[13] self.model_id = packet_buffer[14] if (self.model_id < 53): system_bus_volts = 12 if (self.model_id > 47): system_bus_volts = 24 if (self.model_id > 107): system_bus_volts = 48 self.stack_mode_code = hex(packet_buffer[15]) if (self.stack_mode_code == '0x0') : self.stack_mode = "Standalone Unit" if (self.stack_mode_code == '0x1') : self.stack_mode = "Master in Parallel Stack" if (self.stack_mode_code == '0x2') : self.stack_mode = "Slave in Parallel Stack" if (self.stack_mode_code == '0x4') : self.stack_mode = "Master in Series Stack" if (self.stack_mode_code == '0x8') : self.stack_mode = "Slave in Series Stack" self.amps_ac_in = packet_buffer[16] self.amps_ac_out = packet_buffer[17] self.frequency_ac_out = ((packet_buffer[18] * 256) + packet_buffer[19]) / 10.0 class remote_base_proto(): """definition of remote packet """ # def __init__(self): status_code = '0x00' # 00 packet_buffer[22] search_watts = 0.0 # 01 packet_buffer[23] battery_size = 0.0 # 02 packet_buffer[24] battery_type = 0.0 # 03 packet_buffer[25] charger_amps = 0.0 # 04 packet_buffer[26] shore_ac_amps = 0.0 # 05 packet_buffer[27] revision = 0.0 # 06 packet_buffer[28] parallel_threshold = 0 # 07 packet_buffer[29] LOW NIBBLE force_charge = '0x00' # 07 packet_buffer[29] HIGH NIBBLE genstart_auto = '0x00' # 08 packet_buffer[30] battery_low_trip = 0.0 # 09 packet_buffer[31] volts_ac_trip = 0 # 10 packet_buffer[32] float_volts = 0.0 # 11 packet_buffer[33] equalise_volts = 0.0 # 12 packet_buffer[34] absorb_time = 0.0 # 13 packet_buffer[35] absorb_volts = 14.4 status_descript = "NA" battery_type_descript = "NA" force_charge_descript = "NA" genstart_auto_descript = "NA" # END # # remote_base decode function: def decode(self, packet_buffer): global system_bus_volts if (system_bus_volts == 24): self.absorb_volts *= 2 if (system_bus_volts == 48): self.absorb_volts *= 4 self.status_code = hex(packet_buffer[22]) if (self.status_code == '0x0') : self.status_descript = "Remote Command Clear" self.search_watts = packet_buffer[23] self.battery_size = packet_buffer[24] * 10 self.battery_type = packet_buffer[25] if (self.battery_type == 2): self.battery_type_descript = "Gel" if (self.battery_type == 4): self.battery_type_descript = "Flooded" if (self.battery_type == 8): self.battery_type_descript = "AGM" if (self.battery_type == 10): self.battery_type_descript = "AGM2" if (self.battery_type > 100): self.battery_type_descript = "Custom" if (system_bus_volts == 12): self.absorb_volts = self.battery_type / 10.0 if (system_bus_volts == 24): self.absorb_volts = self.battery_type * 2 / 10.0 if (system_bus_volts == 48): self.absorb_volts = self.battery_type * 4 / 10.0 self.charger_amps = packet_buffer[26] # % of charger capacity ? self.shore_ac_amps = packet_buffer[27] self.revision = packet_buffer[28] / 10.0 self.parallel_threshold = (packet_buffer[29] & 0x0f) * 10 self.force_charge = hex(packet_buffer[29] & 0xf0) if (self.force_charge == '0x10'): self.force_charge_descript = "Disable Refloat" if (self.force_charge == '0x20'): self.force_charge_descript = "Force Silent" if (self.force_charge == '0x40'): self.force_charge_descript = "Force Float" if (self.force_charge == '0x80'): self.force_charge_descript = "Force Bulk" self.genstart_auto = packet_buffer[30] if (self.genstart_auto == 0): self.genstart_auto_descript = "Off" if (self.genstart_auto == 1): self.genstart_auto_descript = "Enable" if (self.genstart_auto == 2): self.genstart_auto_descript = "Test" if (self.genstart_auto == 4): self.genstart_auto_descript = "Enable with Quiet Time" if (self.genstart_auto == 5): self.genstart_auto_descript = "On" if (system_bus_volts == 12): self.battery_low_trip = packet_buffer[31] / 10.0 if (system_bus_volts == 24): self.battery_low_trip = packet_buffer[31] / 10.0 if (system_bus_volts == 48): self.battery_low_trip = packet_buffer[31] * 2 / 10.0 self.volts_ac_trip = packet_buffer[32] if (self.volts_ac_trip == 110): self.volts_ac_trip = 60 if (self.volts_ac_trip == 122): self.volts_ac_trip = 65 if (self.volts_ac_trip == 135): self.volts_ac_trip = 70 if (self.volts_ac_trip == 145): self.volts_ac_trip = 75 if (self.volts_ac_trip == 155): self.volts_ac_trip = 80 if (self.volts_ac_trip == 165): self.volts_ac_trip = 85 if (self.volts_ac_trip == 175): self.volts_ac_trip = 90 if (self.volts_ac_trip == 182): self.volts_ac_trip = 95 if (self.volts_ac_trip == 190): self.volts_ac_trip = 90 if (system_bus_volts == 12): self.float_volts = packet_buffer[33] / 10.0 if (system_bus_volts == 24): self.float_volts = packet_buffer[33] * 2 / 10.0 if (system_bus_volts == 48): self.float_volts = packet_buffer[33] * 4 / 10.0 if (system_bus_volts == 12): # value added to absorbtion volts to give equalise volts range 0 - 2 volts self.equalise_volts = self.absorb_volts + (packet_buffer[34] / 10.0) if (system_bus_volts == 24): # value added to absorbtion volts to give equalise volts range 0 - 2 volts self.equalise_volts = self.absorb_volts + (packet_buffer[34] * 2 / 10.0) if (system_bus_volts == 48): # value added to absorbtion volts to give equalise volts range 0 - 2 volts self.equalise_volts = self.absorb_volts + (packet_buffer[34] * 4 / 10.0) self.absorb_time = packet_buffer[35] / 10.0 # as decimal hours, 2.5 is 2 hours 30 minutes class remote_A0_agsA1_proto(): """definition of remote A0 packet """ # def __init__(self): remote_hours = 0.0 # 14 packet_buffer[36] remote_min = 0.0 # 15 packet_buffer[37] ags_run_time = 0.0 # 16 packet_buffer[38] ags_start_temp = 0.0 # 17 packet_buffer[39] ags_start_volts_dc = 0.0 # 18 packet_buffer[40] ags_quite_hours = 0 # 19 packet_buffer[41] # FOOTER 0xA0 # 20 packet_buffer[42] # HEADER 0xA1 # 21 packet_buffer[43] ags_status = 0 # 22 packet_buffer[44] ags_revision = 0.0 # 23 packet_buffer[45] ags_temperature = 0.0 # 24 packet_buffer[46] ags_gen_runtime = 0.0 # 25 packet_buffer[47] ags_volts_dc = 0.0 # 26 packet_buffer[48] ags_quiet_hours_descrip = "NA" ags_status_descript = "NA" # END # # remoteA0_agsA1 decode function: def decode(self, packet_buffer): global system_bus_volts self.remote_hours = packet_buffer[36] # duplicate from BMK class self.remote_min = packet_buffer[37] # duplicate from BMK class self.ags_run_time = packet_buffer[38] / 10.0 self.ags_start_temp = packet_buffer[39] if (system_bus_volts == 12): self.ags_start_volts_dc = packet_buffer[40] / 10.0 if (system_bus_volts == 24): self.ags_start_volts_dc = packet_buffer[40] * 2 / 10.0 if (system_bus_volts == 48): self.ags_start_volts_dc = packet_buffer[40] * 4 / 10.0 self.ags_quite_hours = packet_buffer[41] if (self.ags_quite_hours == 0): self.ags_quiet_hours_descrip = "Off" if (self.ags_quite_hours == 1): self.ags_quiet_hours_descrip = "21h - 07h" if (self.ags_quite_hours == 2): self.ags_quiet_hours_descrip = "21h - 09h" if (self.ags_quite_hours == 3): self.ags_quiet_hours_descrip = "21h - 09h" if (self.ags_quite_hours == 4): self.ags_quiet_hours_descrip = "22h - 08h" if (self.ags_quite_hours == 5): self.ags_quiet_hours_descrip = "23h - 08h" self.ags_status = packet_buffer[44] if (self.ags_status == 0): self.ags_status_descript = "Non Valid" if (self.ags_status == 1): self.ags_status_descript = "Off" if (self.ags_status == 2): self.ags_status_descript = "Ready" if (self.ags_status == 3): self.ags_status_descript = "Manual Run" if (self.ags_status == 4): self.ags_status_descript = "Inverter in Charge Mode" if (self.ags_status == 5): self.ags_status_descript = "In Quiet Time" if (self.ags_status == 6): self.ags_status_descript = "Start in Test Mode" if (self.ags_status == 7): self.ags_status_descript = "Start on Temperature" if (self.ags_status == 8): self.ags_status_descript = "Start on Voltage" if (self.ags_status == 9): self.ags_status_descript = "Fault Start on Test" if (self.ags_status == 10): self.ags_status_descript = "Fault Start on Temperature" if (self.ags_status == 11): self.ags_status_descript = "Fault Start on Voltage" if (self.ags_status == 12): self.ags_status_descript = "Start Time of Day" if (self.ags_status == 13): self.ags_status_descript = "Start State of Charge" if (self.ags_status == 14): self.ags_status_descript = "Start Exercise" if (self.ags_status == 15): self.ags_status_descript = "Fault Start Time of Day" if (self.ags_status == 16): self.ags_status_descript = "Fault Start State of Charge" if (self.ags_status == 17): self.ags_status_descript = "Fault Start Exercise" if (self.ags_status == 18): self.ags_status_descript = "Start on Amps" if (self.ags_status == 19): self.ags_status_descript = "Start on Topoff" if (self.ags_status == 20): self.ags_status_descript = "Non Valid" if (self.ags_status == 21): self.ags_status_descript = "Fault Start on Amps" if (self.ags_status == 22): self.ags_status_descript = "Fault Start on Topoff" if (self.ags_status == 23): self.ags_status_descript = "Non Valid" if (self.ags_status == 24): self.ags_status_descript = "Fault Maximum Run" if (self.ags_status == 25): self.ags_status_descript = "Gen Run Fault" if (self.ags_status == 26): self.ags_status_descript = "Generator in Warm Up" if (self.ags_status == 27): self.ags_status_descript = "Generator in Cool Down" self.ags_revision = packet_buffer[45] / 10.0 self.ags_temperature = packet_buffer[46] self.ags_run_time = packet_buffer[47] / 10.0 if (system_bus_volts == 12): self.ags_volts_dc = packet_buffer[48] / 10.0 if (system_bus_volts == 24): self.ags_volts_dc = packet_buffer[48] * 2 / 10.0 if (system_bus_volts == 48): self.ags_volts_dc = packet_buffer[48] * 4 / 10.0 class remote_A1_agsA2_proto(): """definition of remote A1 packet """ # def __init__(self): ags_start_time = 0.0 # 14 packet_buffer[36] ags_stop_time = 0.0 # 15 packet_buffer[37] ags_volts_dc_stop = 0.0 # 16 packet_buffer[38] ags_start_delay = 0.0 # 17 packet_buffer[39] ags_stop_delay = 0.0 # 18 packet_buffer[40] ags_max_run_time = 0.0 # 19 packet_buffer[41] # FOOTER 0xA1 # 20 packet_buffer[42] # HEADER 0xA2 # 21 packet_buffer[43] ags_days_last_gen_run = 0 # 22 packet_buffer[44] # ALWAYS 0 # 23 packet_buffer[45] # ALWAYS 0 # 24 packet_buffer[46] # ALWAYS 0 # 25 packet_buffer[47] # ALWAYS 0 # 26 packet_buffer[48] ags_start_stop_enable = "NA" # END # # remoteA1_agsA2 decode function: def decode(self, packet_buffer): global system_bus_volts self.ags_start_time = packet_buffer[36] * 0.25 self.ags_stop_time = packet_buffer[37] * 0.25 if (packet_buffer[36] == packet_buffer[37]): self.ags_start_stop_enable = "Disabled" else: self.ags_start_stop_enable = "Enabled" if (system_bus_volts == 12): self.ags_volts_dc_stop = packet_buffer[38] / 10.0 if (system_bus_volts == 24): self.ags_volts_dc_stop = packet_buffer[38] * 2 / 10.0 if (system_bus_volts == 48): self.ags_volts_dc_stop = packet_buffer[38] * 4 / 10.0 if (packet_buffer[39] & 0x80): # check for minutes selection as MSB: self.ags_start_delay = (packet_buffer[39] & 0x7f) * 60 # strip MSB and store as seconds else: self.ags_start_delay = packet_buffer[39] # store seconds if (packet_buffer[40] & 0x80): # check for minutes selection as MSB: self.ags_stop_delay = (packet_buffer[40] & 0x7f) * 60 # strip MSB and store as seconds else: self.ags_stop_delay = packet_buffer[40] # store seconds self.ags_max_run_time = packet_buffer[41] / 10.0 self.ags_days_last_gen_run = packet_buffer[44] class RTR_proto(): rtr_rev = 0 def decode(self, packet_buffer): self.rtr_rev = packet_buffer[1] / 10 class remote_A2_RTR_proto(): """definition of remote A2 packet """ # def __init__(self): ags_soc_start = 0 # 14 packet_buffer[36] ags_soc_stop = 0 # 15 packet_buffer[37] ags_amps_start = 0 # 16 packet_buffer[38] ags_amps_start_delay = 0 # 17 packet_buffer[39] ags_amps_stop = 0 # 18 packet_buffer[40] ags_amps_stop_delay = 0 # 19 packet_buffer[41] # FOOTER 0xA2 # 20 packet_buffer[42] # HEADER 0x91 # 21 packet_buffer[43] rtr_revision = 0 # 22 packet_buffer[44] # END # # remoteA2_RTR decode function: def decode(self, packet_buffer): global system_bus_volts self.ags_soc_start = packet_buffer[36] self.ags_soc_stop = packet_buffer[37] self.ags_amps_start = packet_buffer[38] if (packet_buffer[39] & 0x80): # check for minutes selection as MSB: self.ags_amps_start_delay = (packet_buffer[39] & 0x7f) * 60 # strip MSB and store as seconds else: self.ags_amps_start_delay = packet_buffer[39] # store seconds self.ags_amps_stop = packet_buffer[40] if (packet_buffer[41] & 0x80): # check for minutes selection as MSB: self.ags_amps_stop_delay = (packet_buffer[41] & 0x7f) * 60 # strip MSB and store as seconds else: self.ags_amps_stop_delay = packet_buffer[41] # store seconds self.rtr_revision = packet_buffer[44] / 10.0 class remote_A3_proto(): """definition of remote A3 packet """ # def __init__(self): ags_quite_time_start = 0.0 # 14 packet_buffer[36] ags_quite_time_stop = 0.0 # 15 packet_buffer[37] ags_exercise_days = 0 # 16 packet_buffer[38] ags_exercise_start_time = 0.0 # 17 packet_buffer[39] ags_exercise_run_time = 0 # 18 packet_buffer[40] ags_top_off = 0 # 19 packet_buffer[41] # FOOTER 0xA3 # 20 packet_buffer[42] # END # # remoteA3 decode function: def decode(self, packet_buffer): global system_bus_volts self.ags_quite_time_start = packet_buffer[36] * 0.25 self.ags_quite_time_stop = packet_buffer[37] * 0.25 self.ags_exercise_days = packet_buffer[38] self.ags_exercise_start_time = packet_buffer[39] * 0.25 self.ags_exercise_run_time = packet_buffer[40] / 10 self.ags_top_off = packet_buffer[41] class remote_A4_proto(): """definition of remote A4 packet """ # def __init__(self): ags_warm_up_time = 0 # 14 packet_buffer[36] ags_cool_down_time = 0 # 15 packet_buffer[37] # ALWAYS 0 # 16 packet_buffer[38] # ALWAYS 0 # 17 packet_buffer[39] # ALWAYS 0 # 18 packet_buffer[40] # ALWAYS 0 # 19 packet_buffer[41] # FOOTER 0xA4 # 20 packet_buffer[42] # END # # remoteA4 decode function: def decode(self, packet_buffer): global system_bus_volts if (packet_buffer[36] & 0x80): # check for minutes selection as MSB: self.ags_warm_up_time = (packet_buffer[36] & 0x7f) * 60 # strip MSB and store as seconds else: self.ags_warm_up_time = packet_buffer[36] # store seconds if (packet_buffer[37] & 0x80): # check for minutes selection as MSB: self.ags_cool_down_time = (packet_buffer[37] & 0x7f) * 60 # strip MSB and store as seconds else: self.ags_cool_down_time = packet_buffer[37] # store seconds class remote_Z0_proto(): """definition of remote Z0 packet """ # def __init__(self): remote_hours = 0 # 14 packet_buffer[36] remote_min = 0 # 15 packet_buffer[37] # ALWAYS 0 # 16 packet_buffer[38] # ALWAYS 0 # 17 packet_buffer[39] # ALWAYS 0 # 18 packet_buffer[40] # ALWAYS 0 # 19 packet_buffer[41] # ALWAYS 0 # 20 packet_buffer[42] # END # # remote_Z0 decode function: def decode(self, packet_buffer): global system_bus_volts # nothing to do here, move along: class remote_BMK_proto(): """definition of remote BMK packet """ # def __init__(self): remote_hours = "" # 14 packet_buffer[36] remote_min = "" # 15 packet_buffer[37] bmk_battery_efficiency = 0 # 16 packet_buffer[38] bmk_resets = 0 # 17 packet_buffer[39] bmk_battery_size = 0 # 18 packet_buffer[40] # ALWAYS 0 # 19 packet_buffer[41] # FOOTER 0x80 # 20 packet_buffer[42] # HEADER 0x81 # 21 packet_buffer[43] bmk_soc = 0 # 22 packet_buffer[44] bmk_volts_dc = 0.0 # 23 packet_buffer[45] HIGH BYTE # 24 packet_buffer[46] LOW BYTE bmk_amps_dc = 0 # 25 packet_buffer[47] HIGH BYTE # 26 packet_buffer[48] LOW BYTE bmk_volts_min = 0.0 # 27 packet_buffer[49] HIGH BYTE # 28 packet_buffer[50] LOW BYTE bmk_volts_max = 0.0 # 29 packet_buffer[51] HIGH BYTE # 30 packet_buffer[52] LOW BYTE bmk_amp_hour_net = 0.0 # 31 packet_buffer[53] HIGH BYTE # 32 packet_buffer[54] LOW BYTE bmk_amp_hour_trip = 0 # 33 packet_buffer[55] HIGH BYTE # 34 packet_buffer[56] LOW BYTE bmk_amp_hour_total = 0 # 35 packet_buffer[57] HIGH BYTE # 36 packet_buffer[58] LOW BYTE bmk_revision = 0.0 # 37 packet_buffer[59] bmk_fault = 0 # 38 packet_buffer[60] bmk_fault_descrip = "NA" # END # # remote_BMK decode function: def decode(self, packet_buffer): global system_bus_volts self.remote_hours = str(packet_buffer[36]) self.remote_min = str(packet_buffer[37]) self.bmk_battery_efficiency = packet_buffer[38] self.bmk_resets = packet_buffer[39] # XXX needs decoding XXX self.bmk_battery_size = packet_buffer[40] * 10 self.bmk_battery_soc = packet_buffer[44] self.bmk_volts_dc = ((packet_buffer[45] * 256) + packet_buffer[46]) / 100.0 # self.bmk_amps_dc = (65535 - ((packet_buffer[47] * 256) + packet_buffer[48])) / 10.0 self.bmk_amps_dc = (packet_buffer[47] * 256) + packet_buffer[48] if (self.bmk_amps_dc > 32768): self.bmk_amps_dc = (65535 - self.bmk_amps_dc) / 10.0 self.bmk_volts_min = ((packet_buffer[49] * 256) + packet_buffer[50]) / 100.0 self.bmk_volts_max = ((packet_buffer[51] * 256) + packet_buffer[52]) / 100.0 self.bmk_amp_hour_net = (packet_buffer[53] * 256) + packet_buffer[54] if (self.bmk_amp_hour_net > 32768): self.bmk_amp_hour_net = 65535 - self.bmk_amp_hour_net self.bmk_amp_hour_trip = ((packet_buffer[55] * 256) + packet_buffer[56]) / 10 self.bmk_amp_hour_total = ((packet_buffer[57] * 256) + packet_buffer[58]) * 100 self.bmk_revision = packet_buffer[59] / 10.0 self.bmk_fault = packet_buffer[60] if (self.bmk_fault == 0): self.bmk_fault_descrip = "Reserved" if (self.bmk_fault == 1): self.bmk_fault_descrip = "No Faults" # documentation says 'Normal', what ever that is: if (self.bmk_fault == 2): self.bmk_fault_descrip = "Fault Start" # main application function: def main(): inverter = inverter_proto() remote_base = remote_base_proto() remote_A0_agsA1 = remote_A0_agsA1_proto() remote_A1_agsA2 = remote_A1_agsA2_proto() remote_A2_RTR = remote_A2_RTR_proto() remote_A3 = remote_A3_proto() remote_A4 = remote_A4_proto() remote_Z0 = remote_Z0_proto() remote_BMK = remote_BMK_proto() RTR = RTR_proto() elapsed_time_base = time.time() global serial_port # access to the global serial port argument: # open commuication port: def openPort(serial_port): try: # mag_port = serial.Serial(port = serial_port, mag_port = serial.Serial(port = '/dev/ttyUSB0', # mag_port = serial.Serial(port = '/dev/tty.usbmodem621', baudrate = 19200, bytesize = 8, timeout = None, interCharTimeout = 0) return(mag_port) except: print('Error: Failed to open commuications port, exiting') exit() # main application loop: def mainLoop(): cflag_1 = 0 cflag_2 = 0 cflag_3 = 0 cflag_4 = 0 cflag_5 = 0 cflag_6 = 0 cflag_7 = 0 serial_byte = 0 datacount = 0 loop_count = 0 # elapsed_time = 0 all_checked = 1 sync_locked = 0 serial_buffer = array.array('i') # call to open the communications port and assign handle on success: mag_port = openPort(serial_port) while (all_checked != 0 ): sync_check_start = time.time() try: serial_byte = ord(mag_port.read(1)) # print(hex(serial_byte)) except: print("Error: Failed to read communications Port, exiting") mag_port.close() exit() sync_check_stop = time.time() sync_locked = sync_check_stop - sync_check_start # print("sync-locked = ", sync_locked) # if (sync_locked > 0.03): ####### # print("sync") ####### while (sync_locked > 0.015): serial_buffer.append(serial_byte) datacount = datacount+1 print(datacount, hex(serial_byte)) # RTR if (datacount == 2): if (serial_buffer[0] == 145) and (serial_buffer[1] == 32): RTR.decode(serial_buffer) print("Decoding Packet RTR") serial_buffer = [] datacount = 0 # Inverter + Remote_B+A0+A1 if (datacount == 49): if (serial_buffer[42] == 160) and (serial_buffer[43] == 161): # 0xa0 and 0xa1 print("Decoding Packets Remote_Base, A0, A1") remote_A0_agsA1.decode(serial_buffer) serial_buffer = [] datacount = 0 cflag_1 = 1 # Inverter + Remote_B+A1+A2 if (datacount == 49): if (serial_buffer[42] == 161) and (serial_buffer[43] == 162): # 0xa1 and 0xa2 print("Decoding Packets Remote_Base, A1, A2") remote_A1_agsA2.decode(serial_buffer) serial_buffer = [] datacount = 0 cflag_2 = 1 # Need to decode yet if (datacount == 49): if (serial_buffer[42] == 162) and (serial_buffer[43] == 161): print("Need to decode yet") print("serial_buffer ", str(serial_buffer)) remote_A1_agsA2.decode(serial_buffer) serial_buffer = [] datacount = 0 cflag_3 = 1 # Inverter + Remote_B+A2+RTR if (datacount == 50): # print("datacount ", datacount, "serial_buffer[48] ", serial_buffer[48], "serial_buffer[49] ", serial_buffer[49]) if (serial_buffer[48] == 0) and (serial_buffer[49] == 124): # 0x0 and 0x7c print("Decoding Packets Remote_Base, A2, RTR") remote_A2_RTR.decode(serial_buffer) serial_buffer = [] datacount = 0 cflag_3 = 1 # Inverter + Remote_B+A3 if (datacount == 43): if (serial_buffer[42] == 163): # 0xa3 print("Decoding Packets Remote_Base, A3") remote_A3.decode(serial_buffer) serial_buffer = [] datacount = 0 cflag_4 = 1 # Inverter + Remote_B+A4 if (datacount == 43): if (serial_buffer[42] == 164): # 0xa4 print("Decoding Packets Remote_Base, A4") remote_A4.decode(serial_buffer) serial_buffer = [] datacount = 0 cflag_5 = 1 # Inverter + Remote_B+Z0 if (datacount == 43): ## print('datacount inverter remotebase z0 ', datacount , serial_buffer[41], serial_buffer[42]) if (serial_buffer[40] == 0) and (serial_buffer[41] == 0): # 0x0 and 0x0 print("Decoding Packets Remote_Base, Z0") remote_Z0.decode(serial_buffer) serial_buffer = [] loop_count += 1 if (loop_count > 5): mag_port.close() report_results() datacount = 0 cflag_6 = 1 # Inverter + Remote_B+80+81 if (datacount == 61): if (serial_buffer[42] == 128) and (serial_buffer[43] == 129): # 0xa0 and 0xa1 print("Decoding Packets Remote_Base, BMK80, BMK81") ## sb = str(serial_buffer) ## print("serial buffer- ", serial_buffer, sb) inverter.decode(serial_buffer) remote_base.decode(serial_buffer) remote_BMK.decode(serial_buffer) serial_buffer = [] datacount = 0 cflag_7 = 1 if (datacount > 127): # escape hatch: # print("datacount = ", datacount) print("Error: Failed to decode, please try again. Exiting") print("buffer- ", serial_buffer) # datacount = 0 # serial_buffer = [] mag_port.close() report_results() # main() # datacount = 0 if (cflag_1 and cflag_2 and cflag_3 and cflag_4 and cflag_5 and cflag_6 and cflag_7): all_checked = 0 sync_locked = 0 # if ( loop_count > 2000 ): # print('loop count =', loop_count) # loop_count = 0 # mag_port.close() # report_results() else: try: serial_byte = ord(mag_port.read(1)) # print(hex(serial_byte)) except: print("Error: Failed to read communications Port, exiting") mag_port.close() exit() report_results() # better print out the results: mag_port.close() # close the com port and finish: # application report function: def report_results(): print("\nLive Data") print("\tClock {:02}:{:02}".format(remote_A0_agsA1.remote_hours, remote_A0_agsA1.remote_min)) print("\tInverter Mode: " + inverter.status_descript) print("\tFault: " + inverter.fault_descript) print("") print("\tInverter Input Volts: {0} Vdc".format(inverter.volts_dc)) print("\tInverter Input Amps: {0} Amps DC".format(inverter.amps_dc)) print("") print("\tInverter Output AC Volts: {0} Vac".format(inverter.volts_ac_out)) print("\tInverter Output AC Amps: {0} Amps".format(inverter.amps_ac_out)) print("\tExternal Input AC Volts: {0} Vac".format(inverter.volts_ac_in)) print("\tExternal Input AC Amps: {0} Amps".format(inverter.amps_ac_in)) print("\tInverter Line Frequency {0} Hz".format(inverter.frequency_ac_out)) print("") print("\tTemperature Battery {0} C".format(inverter.temp_battery)) print("\tTemperature Transformer {0} C".format(inverter.temp_transformer)) print("\tTemperature FETs {0} C".format(inverter.temp_FET)) # print("\tBattery State of Charge: {0} %".format(remote_BMK.bmk_battery_soc)) print("") print("\tGenerator Run Time {0} Hrs".format(remote_A0_agsA1.ags_run_time)) print("\tGenerator Last Run {0} Days".format(remote_A1_agsA2.ags_days_last_gen_run)) print("") print("Remote - ver({0})".format(remote_base.revision)) print("\tStatus: " + remote_base.status_descript) print("") print("Router - ver({0})".format(RTR.rtr_rev)) print("") print("BMK - ver({0})".format(remote_BMK.bmk_revision)) print("\tFault Status: " + remote_BMK.bmk_fault_descrip) print("\tBattery Volts DC: {0} Vdc".format(remote_BMK.bmk_volts_dc)) print("\tAmps DC: {0} Amps".format(remote_BMK.bmk_amps_dc)) print("\tBattery Max: {0} Vdc".format(remote_BMK.bmk_volts_max)) print("\tBattery Min: {0} Vdc".format(remote_BMK.bmk_volts_min)) print("\tBattery Amp Hour Net {0} AmpHr".format(remote_BMK.bmk_amp_hour_net)) print("\tBattery Amp Hour Trip {0} AmpHr".format(remote_BMK.bmk_amp_hour_trip)) print("\tBattery Amp Hour Total {0} AmpHr".format(remote_BMK.bmk_amp_hour_total)) print("") print("AGS - ver({0})".format(remote_A0_agsA1.ags_revision)) print("\tStatus: " + remote_A0_agsA1.ags_status_descript) print("\tGen Start Mode: " + remote_base.genstart_auto_descript) print("\tQuiet Hours: " + remote_A0_agsA1.ags_quiet_hours_descrip) print("\t\tQuiet Time Start {0} Hrs".format(remote_A3.ags_quite_time_start)) print("\t\tQuiet Time Stop {0} Hrs".format(remote_A3.ags_quite_time_stop)) print("") print("\tAuto Start " + remote_A1_agsA2.ags_start_stop_enable) print("\t\tStart Time {0} Hrs".format(remote_A1_agsA2.ags_start_time)) print("\t\tStop Time {0} Hrs".format(remote_A1_agsA2.ags_stop_time)) print("") print("\tStart Delay {0} Sec".format(remote_A1_agsA2.ags_start_delay)) print("\tStop Delay {0} Sec".format(remote_A1_agsA2.ags_stop_delay)) print("\tWarm Up Time {0} Sec".format(remote_A4.ags_warm_up_time)) print("\tCool Down Time {0} Sec".format(remote_A4.ags_cool_down_time)) print("\tSOC Start {0} %".format(remote_A2_RTR.ags_soc_start)) print("\tSOC Stop {0} %".format(remote_A2_RTR.ags_soc_stop)) print("\tAmps Start {0} Amps".format(remote_A2_RTR.ags_amps_start)) print("\tAmps Stop {0} Amps".format(remote_A2_RTR.ags_amps_stop)) print("\t\tAmps Start Delay {0} Sec".format(remote_A2_RTR.ags_amps_start_delay)) print("\t\tAmps Stop Delay {0} Sec".format(remote_A2_RTR.ags_amps_stop_delay)) print("") print("\tMaximum Run Time {0} Hrs".format(remote_A1_agsA2.ags_max_run_time)) print("\tTop Off Time {0} Min".format(remote_A3.ags_top_off)) print("\tExercise Day Period {0} Days".format(remote_A3.ags_exercise_days)) print("\tExercise Start Time {0} Hrs".format(remote_A3.ags_exercise_start_time)) print("\tExercise Run Time {0} Hrs".format(remote_A3.ags_exercise_run_time)) print("\tStart Temperature {0} F".format(remote_A0_agsA1.ags_start_temp)) print("\tStart Volts {0} Vdc".format(remote_A0_agsA1.ags_start_volts_dc)) print("\tStop Volts {0} Vdc".format(remote_A1_agsA2.ags_volts_dc_stop)) print("") print("Inverter - ver({0})".format(inverter.revision)) print("\tInverter Model ID: 0x{0:x}".format(inverter.model_id)) print("\tSystem Buss Voltage: {0} Vdc".format(system_bus_volts)) print("\tInverter Stack Mode: " + inverter.stack_mode) print("\tSearch Watts: {0} Watts".format(remote_base.search_watts)) print("\tCharger Amps: {0} %".format(remote_base.charger_amps)) print("\tShore AC Amps: {0} Amps".format(remote_base.shore_ac_amps)) print("\tParallel Threshold: {0} %".format(remote_base.parallel_threshold)) # print("\tForce Charge: {0}".format(remote_base.force_charge)) # print("\tForce Charge: " + remote_base.force_charge_descript) print("\tAC Volts Trip: {0} Vac".format(remote_base.volts_ac_trip)) print("") print("Battery Settings") print("\tBattery Size, remote: {0} AmpHr".format(remote_base.battery_size)) print("\tBattery Size, BMK: {0} AmpHr".format(remote_BMK.bmk_battery_size)) print("\tBattery Type: " + remote_base.battery_type_descript) print("\tBattery Efficiency: {0} %".format(remote_BMK.bmk_battery_efficiency)) print("\tFloat Volts: {0} Vdc".format(remote_base.float_volts)) print("\tAbsorb Volts: {0} Vdc".format(remote_base.absorb_volts)) print("\tEqualise Volts: {0} Vdc".format(remote_base.equalise_volts)) print("\tBattery Low Trip: {0} Vdc".format(remote_base.battery_low_trip)) mainLoop() #application entry point: if __name__ == "__main__": parser = argparse.ArgumentParser() parser.add_argument("-p", "--port", default = "/dev/ttyUSB0", help="commuications port descriptor, e.g /dev/ttyUSB0 or COM1") parser.add_argument("-d", "--debug", default = 0, type=int, choices=[0, 1, 2], help="debug data") args = parser.parse_args() serial_port = args.port debug_level = args.debug print("MagPy Magnum Energy MagnaSine Data Protocol Decoder\n") print("Debug level : {0}".format(debug_level)) print("serial port : " + serial_port + "\n") system_bus_volts = 0 # set as global variable: main() # call the main function: