297 lines
9.7 KiB
Python
297 lines
9.7 KiB
Python
#!/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()
|