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

#!/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:])