257 lines
7.8 KiB
Python
257 lines
7.8 KiB
Python
#!/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")
|
|
|
|
|
|
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))
|
|
|
|
@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)
|
|
|
|
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)
|
|
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)
|
|
|
|
pub.subscribe (self.update_battery_module_grid, "bms.module.update")
|
|
|
|
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 = self.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]))
|
|
|
|
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()
|