Innovenergy_trunk/firmware/opt/innovenergy/scripts/dump-bms-memory

265 lines
7.0 KiB
Plaintext
Raw Normal View History

2023-02-16 12:57:06 +00:00
#!/usr/bin/python2 -u
# coding=utf-8
import os
import re
import struct
from sys import argv, exit
from datetime import datetime
from pymodbus.pdu import ModbusRequest, ModbusResponse, ExceptionResponse
from pymodbus.other_message import ReportSlaveIdRequest
from pymodbus.exceptions import ModbusException
from pymodbus.pdu import ExceptionResponse
from pymodbus.factory import ClientDecoder
from pymodbus.client.sync import ModbusSerialClient as Modbus
# trick the pycharm type-checker into thinking Callable is in scope, not used at runtime
# noinspection PyUnreachableCode
if False:
from typing import List, Optional, NoReturn
RESET_REGISTER = 0x2087
FIRMWARE_VERSION_REGISTER = 1054
SERIAL_STARTER_DIR = '/opt/victronenergy/serial-starter/'
INSTALLATION_NAME_FILE = '/data/innovenergy/openvpn/installation-name'
OUTPUT_DIR = '/data/innovenergy'
class ReadLogRequest(ModbusRequest):
function_code = 0x42
_rtu_frame_size = 5 # not used
def __init__(self, address = None, **kwargs):
ModbusRequest.__init__(self, **kwargs)
self.sub_function = 0 if address is None else 1
self.address = address
# FUGLY as hell, but necessary bcs PyModbus cannot deal
# with responses that have lengths depending on the sub_function.
# it goes without saying that this isn't thread-safe
ReadLogResponse._rtu_frame_size = 9 if self.sub_function == 0 else 9+128
def encode(self):
if self.sub_function == 0:
return struct.pack('>B', self.sub_function)
else:
return struct.pack('>BI', self.sub_function, self.address)
def decode(self, data):
self.sub_function = struct.unpack('>B', data)
def execute(self, context):
print("EXECUTE1")
def get_response_pdu_size(self):
return ReadLogResponse._rtu_frame_size - 3
def __str__(self):
return "ReadLogAddressRequest"
class ReadLogResponse(ModbusResponse):
function_code = 0x42
_rtu_frame_size = 9 # the WHOLE frame incl crc
def __init__(self, sub_function=0, address=b'\x00', data=None, **kwargs):
ModbusResponse.__init__(self, **kwargs)
self.sub_function = sub_function
self.address = address
self.data = data
def encode(self):
pass
def decode(self, data):
self.address, self.address = struct.unpack_from(">BI", data)
self.data = data[5:]
def __str__(self):
arguments = (self.function_code, self.address)
return "ReadLogAddressResponse(%s, %s)" % arguments
# unfortunately we have to monkey-patch this global table because
# the current (victron) version of PyModbus does not have a
# way to "register" new function-codes yet
ClientDecoder._ClientDecoder__function_table.append(ReadLogResponse)
class LockTTY(object):
def __init__(self, tty):
# type: (str) -> None
self.tty = tty
def __enter__(self):
os.system(SERIAL_STARTER_DIR + 'stop-tty.sh ' + self.tty)
return self
def __exit__(self, exc_type, exc_val, exc_tb):
os.system(SERIAL_STARTER_DIR + 'start-tty.sh ' + self.tty)
def wrap_try_except(error_msg):
def decorate(f):
def applicator(*args, **kwargs):
try:
return f(*args, **kwargs)
except:
print(error_msg)
exit(1)
return applicator
return decorate
def init_modbus(tty):
# type: (str) -> Modbus
return Modbus(
port='/dev/' + tty,
method='rtu',
baudrate=115200,
stopbits=1,
bytesize=8,
timeout=1, # seconds
parity='O')
def print_line(address, data):
address = "{:06X}".format(address)
hex = ' '.join('{:02X}'.format(ord(b)) for b in data)
asc = ''.join(c if ' ' <= c <= '~' else '.' for c in data)
print((address + " : " + hex + " " + asc))
@ wrap_try_except("Failed to dump BMS memory!")
def download_log(modbus, node_id, battery_id):
# type: (Modbus, int, str) -> NoReturn
# Get address of latest log entry
# request = ReadLogRequest(unit=slave_id)
print(('dumping BMS memory from node ' + str(node_id) + ' ...'))
log_file = battery_id + "-node" + str(node_id) + "-" + datetime.now().strftime('%d-%m-%Y') + ".mem"
with open(log_file, 'wb') as f:
start = 0x300000
line = 0x20
for address in range(start, 0x400000, 4*line):
request = ReadLogRequest(address, unit=node_id)
result = modbus.execute(request) # type: ReadLogResponse
data = result.data
for i in range(0, 4):
print_line(address + line*i, data[i*line:(i+1)*line])
f.write(data)
print("done")
print(("wrote memory to " + log_file))
return True
@wrap_try_except("Failed to contact battery!")
def identify_battery(modbus, node_id):
# type: (Modbus, int) -> str
target = 'battery #' + str(node_id) + ' at ' + modbus.port
print(('contacting ' + target + ' ...'))
request = ReportSlaveIdRequest(unit=node_id)
response = modbus.execute(request)
sid = re.sub(r'[^\x20-\x7E]', '', response.identifier)
response = modbus.read_input_registers(address=FIRMWARE_VERSION_REGISTER, count=1, unit=node_id)
fw = '{0:0>4X}'.format(response.registers[0])
return re.sub(" +", "-", sid + " " + fw)
def is_int(value):
# type: (str) -> bool
try:
_ = int(value)
return True
except ValueError:
return False
def print_usage():
print(('Usage: ' + __file__ + ' <slave id> <serial device>'))
print(('Example: ' + __file__ + ' 2 ttyUSB0'))
print ('')
print ('You can omit the "ttyUSB" prefix of the serial device:')
print((' ' + __file__ + ' 2 0'))
print ('')
print ('You can omit the serial device entirely when the "com.victronenergy.battery.<serial device>" service is running:')
print((' ' + __file__ + ' 2'))
print ('')
def get_tty_from_battery_service_name():
# type: () -> Optional[str]
import dbus
bus = dbus.SystemBus()
tty = (
name.split('.')[-1]
for name in bus.list_names()
if name.startswith('com.victronenergy.battery.')
)
return next(tty, None)
def parse_tty(tty):
# type: (Optional[str]) -> str
if tty is None:
return get_tty_from_battery_service_name()
if is_int(tty):
return 'ttyUSB' + argv[1]
else:
return tty
def parse_cmdline_args(argv):
# type: (List[str]) -> (str, int)
slave_id = element_at_or_none(argv, 0)
tty = parse_tty(element_at_or_none(argv, 1))
if slave_id is None or tty is None:
print_usage()
exit(2)
return tty, int(slave_id)
def element_at_or_none(lst, index):
return next(iter(lst[index:]), None)
def main(argv):
# type: (List[str]) -> ()
tty, node_id = parse_cmdline_args(argv)
with LockTTY(tty), init_modbus(tty) as modbus:
battery_id = identify_battery(modbus, node_id)
download_log(modbus, node_id, battery_id)
exit(0)
main(argv[1:])