252 lines
7.7 KiB
Python
252 lines
7.7 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
|
|
|
|
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 0x07 <= type <= 0x0B:
|
|
return BMS_MSG_CAT.MODULE_DATA
|
|
elif type == 0x10:
|
|
if isignal < 4:
|
|
return BMS_MSG_CAT.PACK_PROFILE
|
|
else:
|
|
return BMS_MSG_CAT.PACK_ALARM
|
|
elif type == 0x11:
|
|
return BMS_MSG_CAT.PACK_PROFILE
|
|
elif type == 0x14:
|
|
return BMS_MSG_CAT.PACK_DATA
|
|
elif type == 0x15:
|
|
if isignal < 3:
|
|
return BMS_MSG_CAT.PACK_DATA
|
|
else:
|
|
return BMS_MSG_CAT.PACK_ALARM
|
|
elif 0x16 <= type <= 0x20:
|
|
return BMS_MSG_CAT.PACK_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 Controller:
|
|
def __init__(self, m, v):
|
|
self.view = v
|
|
self.model = m
|
|
|
|
pub.subscribe(lambda: "dummy function", "dummy message")
|
|
|
|
|
|
class Model:
|
|
def __init__(self):
|
|
self.battery_module_state = [{} for i in range(nModule)]
|
|
|
|
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 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.)
|
|
full_id = msg.arbitration_id
|
|
mtag, mid, mtype = parse_can_id(msg.arbitration_id)
|
|
db_msg = db.get_message_by_frame_id(msg.arbitration_id)
|
|
|
|
master_id = int(self.choice_1.GetStringSelection(), 16)
|
|
|
|
if mid == master_id:
|
|
if findCategory(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 findCategory(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 = tohex(mtype) + signal.name
|
|
cat = findCategory(full_id, isig)
|
|
|
|
if cat in cat_propgrid:
|
|
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")
|
|
|
|
'''
|
|
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.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()
|