#!/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 = 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)) self.parent.process_message(msg) class Controller: def __init__(self, m, v): self.view = v self.model = m pub.subscribe(self.model.start_can, "view.start") pub.subscribe(self.model.stop_can, "view.stop") pub.subscribe (self.view.update_battery_module_grid, "bms.module.update") pub.subscribe (self.model.setter, "view.id_selector") class PmgrowBms: def __init__(self, add_printer=True): self.battery_module_state = [{} for i in range(nModule)] self.bus = None self.notifier = None self.listeners = [can.Printer()] if add_printer else [] self.listeners.append(BmsListener(self)) self.master_id = 0x02 def setter (self, id): print ("updating master id") self.master_id = id @property def bms_id (self): return self._bms_id def start_can(self): self.bus = can.Bus() self.notifier = can.Notifier(self.bus, self.listeners) def stop_can(self): self.notifier.stop() self.bus.shutdown() self.notifier = None self.bus = None def __update_bm_state(self, dmsg): imodule = dmsg['BatS_Cell_ModuleNo'] for k in dmsg: self.battery_module_state[imodule][k] = dmsg[k] def process_message(self, msg): msg_prefix, master_id, msg_type = parse_can_id(msg.arbitration_id) full_id = (msg_prefix, master_id, msg_type) db_msg = db.get_message_by_frame_id(msg.arbitration_id) print(master_id, self.master_id) if master_id == self.master_id: 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) pub.sendMessage("bms.module.update", state=self.battery_module_state) elif find_category(full_id, 0) == BMS_MSG_CAT.BMS_PROFILE: pass ''' 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: pass ''' 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") ''' class BmsMonitorView(BmsMonitorFrame): def __init__(self, *args, **kwds): BmsMonitorFrame.__init__(self, *args, **kwds) 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) id_tuple = prefix, id, type if id == 0x02: if find_category(id_tuple, 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(id_tuple, isig) if cat in pgrid_category: self.property_grid_page_info.AppendIn(cat.name, prop) 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 update_battery_module_grid(self, state): i_module = self.spin_ctrl_1.GetValue() module_dict = 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])) self.grid_1.ForceRefresh() def OnIdChoice(self, event): print ("sending update request") pub.sendMessage("view.id_selector", id=int(self.choice_1.GetStringSelection(), 16)) event.Skip() def OnToggle(self, event): if self.button_1.Value: pub.sendMessage("view.start") self.button_1.SetLabel("Stop") else: pub.sendMessage("view.stop") self.button_1.SetLabel("Start") event.Skip() # end of class BmsMonitorFrame class BmsApp(wx.App): def OnInit(self): self.bms = PmgrowBms() 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()