#!/usr/bin/env python # -*- coding: UTF-8 -*- import datetime import csv import win32api import wx import wx.propgrid import gettext from pubsub import pub from enum import Enum import can import cantools from VoltageView import BmsMonitorFrame startTime = datetime.datetime.now() winUptime_ms = win32api.GetTickCount() winUptime_dt = datetime.timedelta(milliseconds=winUptime_ms) bootTime = startTime-winUptime_dt db = cantools.database.load_file('res/EBUS_Host.dbc') nModule = 3 nCell = 12 completeKey = "isModuleDataComplete" timestampKey = "MessageTimestamp" lastUpdateKey = timestampKey + ".old" lastMsgKey = "LastMessageId" 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 <= 0x0C: 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.start_write, "view.start") pub.subscribe(self.stop_write, "view.stop") pub.subscribe (self.write_battery_module, "bms.module.update") pub.subscribe (self.model.setter, "view.id_selector") def start_write(self): filename = self.view.text_ctrl_1.GetValue() self.output_files = [open(filename+"-M{}.csv".format(i+1), 'w') for i in range(nModule)] self.csv_writers = [csv.writer(output_file, delimiter=',') for output_file in self.output_files] for csv_writer in self.csv_writers: csv_writer.writerow(["time"] + ["CV{:02d}".format(i) for i in range(nCell)]) def stop_write(self): for output_file in self.output_files: output_file.close() def write_battery_module(self, state, i_module): module_dict = state[i_module] if module_dict[lastMsgKey] & 0xff == 0x0c: keyString = 'BatS_Cell{}_V' voltages = [module_dict.get(keyString.format(x), -1) for x in range(nCell)] dt = str(bootTime+datetime.timedelta(seconds=module_dict.get(timestampKey, 0.0))) self.csv_writers[i_module].writerow([dt]+voltages) 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 = 0x09 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, msg): i_module = dmsg['BatS_Cell_ModuleNo'] self.battery_module_state[i_module][lastUpdateKey] = self.battery_module_state[i_module].get(timestampKey,0.0) self.battery_module_state[i_module][timestampKey] = msg.timestamp self.battery_module_state[i_module][lastMsgKey] = msg.arbitration_id if self.battery_module_state[i_module][timestampKey] - self.battery_module_state[i_module][lastUpdateKey] > 0.1: self.battery_module_state[i_module][completeKey] = 4 else: self.battery_module_state[i_module][completeKey] = self.battery_module_state[i_module][completeKey] - 1 for k in dmsg: self.battery_module_state[i_module][k] = dmsg[k] return i_module 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: i_module = self.__update_bm_state(decoded, msg) pub.sendMessage("bms.module.update", state=self.battery_module_state, i_module=i_module) 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) 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): # grids = [self.grid_1, self.grid_2, self.grid_3] ts_ctrls = [self.timestamp_1, self.timestamp_2, self.timestamp_3, ] grid = self.grid_1 ts_ctrl = ts_ctrls[i_module] module_dict = state[i_module] if module_dict[lastMsgKey] & 0xff == 0x0c: voltages = self.GetModuleVoltage(module_dict) bal_states = self.GetModuleBalState(module_dict) dt = datetime.datetime.fromtimestamp(module_dict.get(timestampKey, 0.0)) msg_id = module_dict.get(lastMsgKey, 0) ts_ctrl.SetValue(str(bootTime+datetime.timedelta(seconds=dt.timestamp()))) for i in range(nCell): grid.SetCellValue(i, i_module, str(voltages[i])) print("refreshing module ", i_module, "latest update: ", bootTime+datetime.timedelta(seconds=dt.timestamp()), dt.timestamp(), "{:x}".format(msg_id)) grid.ForceRefresh() def OnIdChoice(self, event): print ("sending update request") # event.Get # wx.Event.GetEventObject() 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(add_printer=True) 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()