#!/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 from BmsView import BmsMonitorFrame db = cantools.database.load_file('res/EBUS_Host.dbc') nModule = 22 nCell = 12 def parse_can_id(arbitration_id): mprefix = (arbitration_id & 0xfffff000) >> 12 mid = (arbitration_id & 0xf00) >> 8 mtype = arbitration_id & 0xff return mprefix, mid, mtype BMS_MSG_CAT = Enum( "BMS_MSG_CAT", "BMS_PROFILE " "PACK_PROFILE " "PACK_ALARM " "PACK_DATA " "MODULE_DATA " "UNDEFINED " "ERROR " ) def find_category(imsg, isignal): msg_prefix, master_id, msg_type = parse_can_id(imsg) if msg_prefix != 0x14180: return BMS_MSG_CAT.ERROR if 0x00 <= msg_type <= 0x06: return BMS_MSG_CAT.BMS_PROFILE elif 0x07 <= msg_type <= 0x0B: return BMS_MSG_CAT.MODULE_DATA elif msg_type == 0x10: if isignal < 4: return BMS_MSG_CAT.PACK_PROFILE else: return BMS_MSG_CAT.PACK_ALARM elif msg_type == 0x11: return BMS_MSG_CAT.PACK_PROFILE elif msg_type == 0x14: return BMS_MSG_CAT.PACK_DATA elif msg_type == 0x15: if isignal < 3: return BMS_MSG_CAT.PACK_DATA else: return BMS_MSG_CAT.PACK_ALARM elif 0x16 <= msg_type <= 0x20: return BMS_MSG_CAT.PACK_DATA else: return BMS_MSG_CAT.UNDEFINED pgrid_category = [ BMS_MSG_CAT.BMS_PROFILE, BMS_MSG_CAT.PACK_PROFILE, BMS_MSG_CAT.PACK_ALARM, BMS_MSG_CAT.PACK_DATA ] str_hex = "{:#04X}-".format class BmsListener (can.Listener): def __init__(self, parent): self.parent = parent def on_message_received(self, msg): print("{:#X}".format(msg.arbitration_id)) class Controller: def __init__(self, m, v): self.view = v self.model = m pub.subscribe(self.model.start_can, "model.start") pub.subscribe(self.model.stop_can, "model.stop") class Model: def __init__(self): self.battery_module_state = [{} for i in range(nModule)] self.bus = None self.notifier = None self.listen_bms = can.Printer() #self.listen_bms = BmsListener(self) def start_can(self): self.bus = can.Bus() self.notifier = can.Notifier(self.bus, [self.listen_bms]) def stop_can(self): self.notifier.stop() self.bus.shutdown() self.notifier = None self.bus = None def process_message(self, msg): pass 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 pgrid_category: 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: prefix, id, type = parse_can_id(msg.frame_id) if id == 0x02: if find_category(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 = str_hex(type) + signal.name prop = wx.propgrid.StringProperty(label=plabel) prop.SetHelpString(signal.comment) cat = find_category(msg.frame_id, isig) if cat in pgrid_category: self.property_grid_page_info.AppendIn(cat.name, prop) def OnToggle(self, event): btnLabel = self.button_1.GetLabel() if self.button_1.Value: # self.bus = can.Bus() # self.timer.Start(10) pub.sendMessage ("model.start") self.button_1.SetLabel("Stop") else: # self.bus.shutdown() # self.bus = None # self.timer.Stop() pub.sendMessage ("model.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.) full_id = msg.arbitration_id msg_prefix, master_id, msg_type = parse_can_id(msg.arbitration_id) db_msg = db.get_message_by_frame_id(msg.arbitration_id) id_chosen = int(self.choice_1.GetStringSelection(), 16) if master_id == id_chosen: if find_category(full_id, 0) == BMS_MSG_CAT.MODULE_DATA: try: decoded = db.decode_message(msg.arbitration_id, msg.data) except KeyError as key: print(key, " does not exist") else: self.__update_bm_state(decoded) elif find_category(full_id, 0) == BMS_MSG_CAT.BMS_PROFILE: try: print("bmsprofile message raw", db_msg.name, msg.data, msg.data.decode()) except UnicodeDecodeError: print("bmsprofile message raw", db_msg.name, msg.data, int.from_bytes(msg.data, byteorder="big")) if self.property_grid_page_info.GetProperty(db_msg.name): self.property_grid_page_info.SetPropertyValue(db_msg.name, msg.data.decode()) else: decoded = db.decode_message(msg.arbitration_id, msg.data) for isig, signal in enumerate(db_msg.signals): plabel = str_hex(msg_type) + signal.name cat = find_category(full_id, isig) if cat in pgrid_category: if self.property_grid_page_info.GetProperty(plabel): # print (plabel, decoded[signal.name]) value = float(decoded[signal.name]) if signal.is_float else decoded[signal.name] self.property_grid_page_info.SetPropertyValue(plabel, value) else: pass # print (plabel, " is Not Available") 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.bms = Model() self.frame = BmsMonitorView(None, wx.ID_ANY, "") self.bms_controller = Controller(self.bms, self.frame) 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()