#!/usr/bin/env python # -*- coding: UTF-8 -*- import wx import wx.propgrid import gettext from pubsub import pub from enum import Enum import can import cantools db = cantools.database.load_file('res/EBUS_Host.dbc') nModule = 22 nCell = 12 from BmsView import BmsMonitorFrame def parse_can_id (arbitration_id): mtag = (arbitration_id & 0xfffff000) >> 12 mid = (arbitration_id & 0xf00) >> 8 mtype = arbitration_id & 0xff return mtag, mid, mtype BMS_MSG_CAT = Enum( "BMS_MSG_CAT", "BMS_PROFILE " "PACK_PROFILE " "PACK_ALARM " "PACK_DATA " "MODULE_DATA " "UNDEFINED " "ERROR " ) def findCategory (imsg, isignal): tag, id, type = parse_can_id(imsg) if tag != 0x14180: return BMS_MSG_CAT.ERROR if 0x00 <= type <= 0x06: return BMS_MSG_CAT.BMS_PROFILE elif type == 0x11: return BMS_MSG_CAT.PACK_PROFILE elif type == 0x10: if isignal < 4: return BMS_MSG_CAT.PACK_PROFILE else: return BMS_MSG_CAT.PACK_ALARM elif type == 0x15: if isignal < 3: return BMS_MSG_CAT.PACK_DATA else: return BMS_MSG_CAT.PACK_ALARM elif type == 0x14: return BMS_MSG_CAT.PACK_DATA elif 0x16 <= type <= 0x20: return BMS_MSG_CAT.PACK_DATA elif 0x07 <= type <= 0x0B: return BMS_MSG_CAT.MODULE_DATA else: return BMS_MSG_CAT.UNDEFINED cat_propgrid = [ BMS_MSG_CAT.BMS_PROFILE, BMS_MSG_CAT.PACK_PROFILE, BMS_MSG_CAT.PACK_ALARM, BMS_MSG_CAT.PACK_DATA ] tohex = "{:#04X}-".format class BmsMonitorView(BmsMonitorFrame): def __init__(self, *args, **kwds): BmsMonitorFrame.__init__(self, *args, **kwds) self.battery_module_state = [{} for i in range(nModule)] self.timer = wx.Timer(self) self.Bind(wx.EVT_TIMER, self.OnTimer, self.timer) self.property_grid_page_info = self.property_grid_1.AddPage("Info") for c in cat_propgrid: cat_name = c.name cat = wx.propgrid.PropertyCategory(label=cat_name) self.property_grid_page_info.Append(cat) # bms_model = wx.propgrid.StringProperty(label="Model", name="bms_model", value="B3DWER32EW") # self.property_grid_page_info.AppendIn("packdata", bms_model) for msg in db.messages: tag, id, type = parse_can_id(msg.frame_id) if id == 0x02: if findCategory(msg.frame_id, 0) == BMS_MSG_CAT.BMS_PROFILE: prop = wx.propgrid.StringProperty(label=msg.name) self.property_grid_page_info.AppendIn(BMS_MSG_CAT.BMS_PROFILE.name, prop) for isig, signal in enumerate(msg.signals): plabel = tohex(type)+signal.name prop = wx.propgrid.StringProperty(label=plabel) prop.SetHelpString(signal.comment) cat = findCategory(msg.frame_id, isig) if cat in cat_propgrid: self.property_grid_page_info.AppendIn(cat.name, prop) ''' if tag == 0x14180 and id == 0x2 and 0x00 <= type <= 0x06: # BMS Profile self.property_grid_page_info.AppendIn(BMS_MSG_CAT.BMS_PROFILE, prop) if tag == 0x14180 and id == 0x2 and 0x10 <= type <= 0x20: # Pack Profile and Data self.property_grid_page_info.AppendIn("packdata", prop) ''' def OnToggle(self, event): btnLabel = self.button_1.GetLabel() if self.button_1.Value: print ("starting timer...") self.bus = can.Bus() self.timer.Start(10) self.button_1.SetLabel("Stop") else: print ( "timer stopped!" ) self.bus.shutdown() self.bus = None self.timer.Stop() self.button_1.SetLabel("Start") event.Skip() def __update_bm_state(self, dmsg): imodule = dmsg['BatS_Cell_ModuleNo'] for k in dmsg: self.battery_module_state[imodule][k] = dmsg[k] def GetModuleVoltage(self, module_dict): keyString='BatS_Cell{}_V' return [module_dict.get(keyString.format(x), -1) for x in range(nCell)] def GetModuleBalState(self, module_dict): keyString='BatS_Cell{}_BalState' return [module_dict.get(keyString.format(x), -1) for x in range(nCell)] def OnTimer(self, event): # print("Read CAN Message!") msg = self.bus.recv(timeout=1.) fullid = msg.arbitration_id mtag, mid, mtype = parse_can_id(msg.arbitration_id) dbmsg = db.get_message_by_frame_id(msg.arbitration_id) master_id = int(self.choice_1.GetStringSelection(), 16) if mid == master_id: if findCategory(fullid, 0) == BMS_MSG_CAT.MODULE_DATA: try: dmsg = db.decode_message(msg.arbitration_id, msg.data) except KeyError as key: print(key, " does not exist") else: self.__update_bm_state(dmsg) elif findCategory(fullid, 0) == BMS_MSG_CAT.BMS_PROFILE: try: print ("bmsprofile message raw", (dbmsg.name), msg.data, msg.data.decode()) except UnicodeDecodeError: print ("bmsprofile message raw", (dbmsg.name), msg.data, int.from_bytes(msg.data, byteorder="big")) if self.property_grid_page_info.GetProperty(dbmsg.name): self.property_grid_page_info.SetPropertyValue(dbmsg.name, msg.data.decode()) else: dmsg = db.decode_message(msg.arbitration_id, msg.data) for isig, signal in enumerate(dbmsg.signals): plabel = tohex(mtype) + signal.name cat = findCategory(fullid, isig) if cat in cat_propgrid: if self.property_grid_page_info.GetProperty(plabel): # print (plabel, dmsg[signal.name]) value = float(dmsg[signal.name]) if signal.is_float else dmsg[signal.name] self.property_grid_page_info.SetPropertyValue(plabel, value) else: pass # print (plabel, " is Not Available") ''' if tag == 0x14180 and id == 0x2 and 0x00 <= type <= 0x06: # BMS Profile self.property_grid_page_info.AppendIn(BMS_MSG_CAT.BMS_PROFILE, prop) if tag == 0x14180 and id == 0x2 and 0x10 <= type <= 0x20: # Pack Profile and Data self.property_grid_page_info.AppendIn("packdata", prop) ''' i_module = self.spin_ctrl_1.GetValue() module_dict = self.battery_module_state[i_module] voltages = self.GetModuleVoltage(module_dict) bal_states = self.GetModuleBalState(module_dict) for i in range(nCell): self.grid_1.SetCellValue(i, 0, str(voltages[i])) self.grid_1.SetCellValue(i, 2, str(bal_states[i])) event.Skip() # end of class BmsMonitorFrame class BmsApp(wx.App): def OnInit(self): self.frame = BmsMonitorView(None, wx.ID_ANY, "") self.SetTopWindow(self.frame) self.frame.Show() return True # end of class BmsApp if __name__ == "__main__": gettext.install("app") # replace with the appropriate catalog name app = BmsApp(0) app.MainLoop()