Merge remote-tracking branch 'origin/main'

This commit is contained in:
kostas 2024-06-13 15:18:19 +02:00
commit be325dd90c
30 changed files with 46283 additions and 307 deletions

Binary file not shown.

View File

@ -0,0 +1,51 @@
import serial
import logging
# dbus configuration
FIRMWARE_VERSION = 1 # value returned by getValue (getText returns string value reported by battery)
HARDWARE_VERSION = 1 # value returned by getValue (getText returns string value reported by battery)
CONNECTION = 'Modbus RTU'
PRODUCT_NAME = 'FZS 48TL200'
PRODUCT_ID = 0xB012 # assigned by victron
DEVICE_INSTANCE = 1
SERVICE_NAME_PREFIX = 'com.victronenergy.battery.'
# driver configuration
SOFTWARE_VERSION = '3.0.3'
UPDATE_INTERVAL = 2000 # milliseconds
#LOG_LEVEL = logging.INFO
LOG_LEVEL = logging.DEBUG
# modbus configuration
BASE_ADDRESS = 999
#NO_OF_REGISTERS = 63
NO_OF_REGISTERS = 64
MAX_SLAVE_ADDRESS = 10
# RS 485 configuration
PARITY = serial.PARITY_ODD
TIMEOUT = 0.1 # seconds
BAUD_RATE = 115200
BYTE_SIZE = 8
STOP_BITS = 1
MODE = 'rtu'
# battery configuration
MAX_CHARGE_VOLTAGE = 58
I_MAX_PER_STRING = 15
NUM_OF_STRING_PER_BATTERY = 5
AH_PER_STRING = 40
V_MAX = 54.2
R_STRING_MIN = 0.125
R_STRING_MAX = 0.250

View File

@ -0,0 +1,119 @@
from collections import Iterable
from decimal import *
import config as cfg
from data import LedState, BatteryStatus
# trick the pycharm type-checker into thinking Callable is in scope, not used at runtime
# noinspection PyUnreachableCode
if False:
from typing import Callable
def read_bool(register, bit):
# type: (int, int) -> Callable[[BatteryStatus], bool]
def get_value(status):
# type: (BatteryStatus) -> bool
value = status.modbus_data[register - cfg.BASE_ADDRESS]
return value & (1 << bit) > 0
return get_value
def read_float(register, scale_factor=1.0, offset=0.0, places=2):
# type: (int, float, float) -> Callable[[BatteryStatus], float]
def get_value(status):
# type: (BatteryStatus) -> float
value = status.modbus_data[register - cfg.BASE_ADDRESS]
if value >= 0x8000: # convert to signed int16
value -= 0x10000 # fiamm stores their integers signed AND with sign-offset @#%^&!
result = (value+offset)*scale_factor
return round(result,places)
return get_value
def read_hex_string(register, count):
# type: (int, int) -> Callable[[BatteryStatus], str]
"""
reads count consecutive modbus registers from start_address,
and returns a hex representation of it:
e.g. for count=4: DEAD BEEF DEAD BEEF.
"""
start = register - cfg.BASE_ADDRESS
end = start + count
def get_value(status):
# type: (BatteryStatus) -> str
return ' '.join(['{0:0>4X}'.format(x) for x in status.modbus_data[start:end]])
return get_value
def read_led_state(register, led):
# type: (int, int) -> Callable[[BatteryStatus], int]
read_lo = read_bool(register, led * 2)
read_hi = read_bool(register, led * 2 + 1)
def get_value(status):
# type: (BatteryStatus) -> int
lo = read_lo(status)
hi = read_hi(status)
if hi:
if lo:
return LedState.blinking_fast
else:
return LedState.blinking_slow
else:
if lo:
return LedState.on
else:
return LedState.off
return get_value
def read_bitmap(register):
# type: (int) -> Callable[[BatteryStatus], bitmap]
def get_value(status):
# type: (BatteryStatus) -> bitmap
value = status.modbus_data[register - cfg.BASE_ADDRESS]
return value
return get_value
def append_unit(unit):
# type: (unicode) -> Callable[[unicode], unicode]
def get_text(v):
# type: (unicode) -> unicode
return "{0}{1}".format(str(v), unit)
return get_text
def mean(numbers):
# type: (Iterable[float] | Iterable[int]) -> float
return float("{:.2f}".format(float(sum(numbers)) / len(numbers)))
def ssum(numbers):
# type: (Iterable[float] | Iterable[int]) -> float
return float("{:.2f}".format(float(sum(numbers))))
def first(ts):
return next(t for t in ts)
def return_in_list(ts):
return ts

View File

@ -0,0 +1,63 @@
import config as cfg
from collections import Iterable
# trick the pycharm type-checker into thinking Callable is in scope, not used at runtime
# noinspection PyUnreachableCode
if False:
from typing import Callable
class LedState(object):
"""
from page 6 of the '48TLxxx ModBus Protocol doc'
"""
off = 0
on = 1
blinking_slow = 2
blinking_fast = 3
class LedColor(object):
green = 0
amber = 1
blue = 2
red = 3
class CsvSignal(object):
def __init__(self, name, get_value, get_text = None):
self.name = name
self.get_value = get_value if callable(get_value) else lambda _: get_value
self.get_text = get_text
if get_text is None:
self.get_text = ""
class Battery(object):
""" Data record to hold hardware and firmware specs of the battery """
def __init__(self, slave_address, hardware_version, firmware_version, bms_version, ampere_hours):
# type: (int, str, str, str, int) -> None
self.slave_address = slave_address
self.hardware_version = hardware_version
self.firmware_version = firmware_version
self.bms_version = bms_version
self.ampere_hours = ampere_hours
def __str__(self):
return 'slave address = {0}\nhardware version = {1}\nfirmware version = {2}\nbms version = {3}\nampere hours = {4}'.format(
self.slave_address, self.hardware_version, self.firmware_version, self.bms_version, str(self.ampere_hours))
class BatteryStatus(object):
"""
record holding the current status of a battery
"""
def __init__(self, battery, modbus_data):
# type: (Battery, list[int]) -> None
self.battery = battery
self.modbus_data = modbus_data

View File

@ -0,0 +1,731 @@
#! /usr/bin/python3 -u
import re
import sys
import logging
from gi.repository import GLib
import config as cfg
import convert as c
from pymodbus.register_read_message import ReadInputRegistersResponse
from pymodbus.client.sync import ModbusSerialClient as Modbus
from pymodbus.other_message import ReportSlaveIdRequest
from pymodbus.exceptions import ModbusException
from pymodbus.pdu import ExceptionResponse
from dbus.mainloop.glib import DBusGMainLoop
from data import BatteryStatus, Battery, LedColor, CsvSignal, LedState
from collections import Iterable
from os import path
app_dir = path.dirname(path.realpath(__file__))
sys.path.insert(1, path.join(app_dir, 'ext', 'velib_python'))
#from vedbus import VeDbusService as DBus
import time
import os
import csv
import requests
import hmac
import hashlib
import base64
from datetime import datetime
import io
class S3config:
def __init__(self):
self.bucket = "1-c0436b6a-d276-4cd8-9c44-1eae86cf5d0e"
self.region = "sos-ch-dk-2"
self.provider = "exo.io"
self.key = "EXOcc0e47a4c4d492888ff5a7f2"
self.secret = "79QG4unMh7MeVacMnXr5xGxEyAlWZDIdM-dg_nXFFr4"
self.content_type = "text/plain; charset=utf-8"
@property
def host(self):
return f"{self.bucket}.{self.region}.{self.provider}"
@property
def url(self):
return f"https://{self.host}"
def create_put_request(self, s3_path, data):
headers = self._create_request("PUT", s3_path)
url = f"{self.url}/{s3_path}"
response = requests.put(url, headers=headers, data=data)
return response
def _create_request(self, method, s3_path):
date = datetime.utcnow().strftime('%a, %d %b %Y %H:%M:%S GMT')
auth = self._create_authorization(method, self.bucket, s3_path, date, self.key, self.secret, self.content_type)
headers = {
"Host": self.host,
"Date": date,
"Authorization": auth,
"Content-Type": self.content_type
}
return headers
@staticmethod
def _create_authorization(method, bucket, s3_path, date, s3_key, s3_secret, content_type="", md5_hash=""):
payload = f"{method}\n{md5_hash}\n{content_type}\n{date}\n/{bucket.strip('/')}/{s3_path.strip('/')}"
signature = base64.b64encode(
hmac.new(s3_secret.encode(), payload.encode(), hashlib.sha1).digest()
).decode()
return f"AWS {s3_key}:{signature}"
def read_csv_as_string(file_path):
"""
Reads a CSV file from the given path and returns its content as a single string.
"""
try:
with open(file_path, 'r', encoding='utf-8') as file:
return file.read()
except FileNotFoundError:
print(f"Error: The file {file_path} does not exist.")
return None
except IOError as e:
print(f"IO error occurred: {str(e)}")
return None
CSV_DIR = "/data/csv_files_service/"
#CSV_DIR = "csv_files/"
# Define the path to the file containing the installation name
INSTALLATION_NAME_FILE = '/data/innovenergy/openvpn/installation-name'
# trick the pycharm type-checker into thinking Callable is in scope, not used at runtime
# noinspection PyUnreachableCode
if False:
from typing import Callable
def interpret_limb_bitmap(bitmap_value):
# The bit for string 1 also monitors all 5 strings: 0000 0000 means All 5 strings activated. 0000 0001 means string 1 disabled.
string1_disabled = int((bitmap_value & 0b00001) != 0)
string2_disabled = int((bitmap_value & 0b00010) != 0)
string3_disabled = int((bitmap_value & 0b00100) != 0)
string4_disabled = int((bitmap_value & 0b01000) != 0)
string5_disabled = int((bitmap_value & 0b10000) != 0)
n_limb_strings = string1_disabled+string2_disabled+string3_disabled+string4_disabled+string5_disabled
return n_limb_strings
def create_csv_signals(firmware_version):
def read_power(status):
return int(read_current(status) * read_voltage(status))
read_voltage = c.read_float(register=999, scale_factor=0.01, offset=0, places=2)
read_current = c.read_float(register=1000, scale_factor=0.01, offset=-10000, places=2)
read_limb_bitmap = c.read_bitmap(1059)
def string1_disabled(status):
bitmap_value = read_limb_bitmap(status)
return int((bitmap_value & 0b00001) != 0)
def string2_disabled(status):
bitmap_value = read_limb_bitmap(status)
return int((bitmap_value & 0b00010) != 0)
def string3_disabled(status):
bitmap_value = read_limb_bitmap(status)
return int((bitmap_value & 0b00100) != 0)
def string4_disabled(status):
bitmap_value = read_limb_bitmap(status)
return int((bitmap_value & 0b01000) != 0)
def string5_disabled(status):
bitmap_value = read_limb_bitmap(status)
return int((bitmap_value & 0b10000) != 0)
def limp_strings_value(status):
return interpret_limb_bitmap(read_limb_bitmap(status))
def calc_power_limit_imposed_by_voltage_limit(v, i, v_limit, r_int):
# type: (float, float, float, float) -> float
dv = v_limit - v
di = dv / r_int
p_limit = v_limit * (i + di)
return p_limit
def calc_power_limit_imposed_by_current_limit(v, i, i_limit, r_int):
# type: (float, float, float, float) -> float
di = i_limit - i
dv = di * r_int
p_limit = i_limit * (v + dv)
return p_limit
def calc_max_charge_power(status):
# type: (BatteryStatus) -> int
n_strings = cfg.NUM_OF_STRING_PER_BATTERY-limp_strings_value(status)
i_max = n_strings * cfg.I_MAX_PER_STRING
v_max = cfg.V_MAX
r_int_min = cfg.R_STRING_MIN / n_strings
r_int_max = cfg.R_STRING_MAX / n_strings
v = read_voltage(status)
i = read_current(status)
p_limits = [
calc_power_limit_imposed_by_voltage_limit(v, i, v_max,r_int_min),
calc_power_limit_imposed_by_voltage_limit(v, i, v_max, r_int_max),
calc_power_limit_imposed_by_current_limit(v, i, i_max, r_int_min),
calc_power_limit_imposed_by_current_limit(v, i, i_max, r_int_max),
]
p_limit = min(p_limits) # p_limit is normally positive here (signed)
p_limit = max(p_limit, 0) # charge power must not become negative
return int(p_limit)
def calc_max_discharge_power(status):
n_strings = cfg.NUM_OF_STRING_PER_BATTERY-limp_strings_value(status)
max_discharge_current = n_strings*cfg.I_MAX_PER_STRING
return int(max_discharge_current*read_voltage(status))
def return_led_state_blue(status):
led_state = c.read_led_state(register=1004, led=LedColor.blue)(status)
if led_state == LedState.blinking_fast or led_state == LedState.blinking_slow:
return "Blinking"
elif led_state == LedState.on:
return "On"
elif led_state == LedState.off:
return "Off"
return "Unknown"
def return_led_state_red(status):
led_state = c.read_led_state(register=1004, led=LedColor.red)(status)
if led_state == LedState.blinking_fast or led_state == LedState.blinking_slow:
return "Blinking"
elif led_state == LedState.on:
return "On"
elif led_state == LedState.off:
return "Off"
return "Unknown"
def return_led_state_green(status):
led_state = c.read_led_state(register=1004, led=LedColor.green)(status)
if led_state == LedState.blinking_fast or led_state == LedState.blinking_slow:
return "Blinking"
elif led_state == LedState.on:
return "On"
elif led_state == LedState.off:
return "Off"
return "Unknown"
def return_led_state_amber(status):
led_state = c.read_led_state(register=1004, led=LedColor.amber)(status)
if led_state == LedState.blinking_fast or led_state == LedState.blinking_slow:
return "Blinking"
elif led_state == LedState.on:
return "On"
elif led_state == LedState.off:
return "Off"
return "Unknown"
total_current = c.read_float(register=1062, scale_factor=0.01, offset=-10000, places=1)
def read_total_current(status):
return total_current(status)
def read_heating_current(status):
return total_current(status) - read_current(status)
def read_heating_power(status):
return read_voltage(status) * read_heating_current(status)
soc_ah = c.read_float(register=1002, scale_factor=0.1, offset=-10000, places=1)
def read_soc_ah(status):
return soc_ah(status)
def hex_string_to_ascii(hex_string):
# Ensure the hex_string is correctly formatted without spaces
hex_string = hex_string.replace(" ", "")
# Convert every two characters (a byte) in the hex string to ASCII
ascii_string = ''.join([chr(int(hex_string[i:i+2], 16)) for i in range(0, len(hex_string), 2)])
return ascii_string
battery_status_reader = c.read_hex_string(1060,2)
def read_eoc_reached(status):
battery_status_string = battery_status_reader(status)
#if hex_string_to_ascii(battery_status_string) == "EOC_":
#return True
#return False
return hex_string_to_ascii(battery_status_string) == "EOC_"
def read_serial_number(status):
serial_regs = [1055, 1056, 1057, 1058]
serial_parts = []
for reg in serial_regs:
# reading each register as a single hex value
hex_value_fun = c.read_hex_string(reg, 1)
hex_value = hex_value_fun(status)
# append without spaces and leading zeros stripped if any
serial_parts.append(hex_value.replace(' ', ''))
# concatenate all parts to form the full serial number
serial_number = ''.join(serial_parts).rstrip('0')
return serial_number
return [
CsvSignal('/Battery/Devices/FwVersion', firmware_version),
CsvSignal('/Battery/Devices/Dc/Power', read_power, 'W'),
CsvSignal('/Battery/Devices/Dc/Voltage', read_voltage, 'V'),
CsvSignal('/Battery/Devices/Soc', c.read_float(register=1053, scale_factor=0.1, offset=0, places=1), '%'),
CsvSignal('/Battery/Devices/Temperatures/Cells/Average', c.read_float(register=1003, scale_factor=0.1, offset=-400, places=1), 'C'),
CsvSignal('/Battery/Devices/Dc/Current', read_current, 'A'),
CsvSignal('/Battery/Devices/BusCurrent', read_total_current, 'A'),
CsvSignal('/Battery/Devices/CellsCurrent', read_current, 'A'),
CsvSignal('/Battery/Devices/HeatingCurrent', read_heating_current, 'A'),
CsvSignal('/Battery/Devices/HeatingPower', read_heating_power, 'W'),
CsvSignal('/Battery/Devices/SOCAh', read_soc_ah),
CsvSignal('/Battery/Devices/Leds/Blue', return_led_state_blue),
CsvSignal('/Battery/Devices/Leds/Red', return_led_state_red),
CsvSignal('/Battery/Devices/Leds/Green', return_led_state_green),
CsvSignal('/Battery/Devices/Leds/Amber', return_led_state_amber),
CsvSignal('/Battery/Devices/BatteryStrings/String1Active', string1_disabled),
CsvSignal('/Battery/Devices/BatteryStrings/String2Active', string2_disabled),
CsvSignal('/Battery/Devices/BatteryStrings/String3Active', string3_disabled),
CsvSignal('/Battery/Devices/BatteryStrings/String4Active', string4_disabled),
CsvSignal('/Battery/Devices/BatteryStrings/String5Active', string5_disabled),
CsvSignal('/Battery/Devices/IoStatus/ConnectedToDcBus', c.read_bool(register=1013, bit=0)),
CsvSignal('/Battery/Devices/IoStatus/AlarmOutActive', c.read_bool(register=1013, bit=1)),
CsvSignal('/Battery/Devices/IoStatus/InternalFanActive', c.read_bool(register=1013, bit=2)),
CsvSignal('/Battery/Devices/IoStatus/VoltMeasurementAllowed', c.read_bool(register=1013, bit=3)),
CsvSignal('/Battery/Devices/IoStatus/AuxRelayBus', c.read_bool(register=1013, bit=4)),
CsvSignal('/Battery/Devices/IoStatus/RemoteStateActive', c.read_bool(register=1013, bit=5)),
CsvSignal('/Battery/Devices/IoStatus/RiscActive', c.read_bool(register=1013, bit=6)),
CsvSignal('/Battery/Devices/Eoc', read_eoc_reached),
CsvSignal('/Battery/Devices/SerialNumber', read_serial_number),
CsvSignal('/Battery/Devices/TimeSinceTOC', c.read_float(register=1052)),
CsvSignal('/Battery/Devices/MaxChargePower', calc_max_charge_power),
CsvSignal('/Battery/Devices/MaxDischargePower', calc_max_discharge_power),
# Warnings
CsvSignal('/Battery/Devices/WarningFlags/TaM1', c.read_bool(register=1005, bit=1)),
CsvSignal('/Battery/Devices/WarningFlags/TbM1', c.read_bool(register=1005, bit=4)),
CsvSignal('/Battery/Devices/WarningFlags/VBm1', c.read_bool(register=1005, bit=6)),
CsvSignal('/Battery/Devices/WarningFlags/VBM1', c.read_bool(register=1005, bit=8)),
CsvSignal('/Battery/Devices/WarningFlags/IDM1', c.read_bool(register=1005, bit=10)),
CsvSignal('/Battery/Devices/WarningFlags/vsm1', c.read_bool(register=1005, bit=22)),
CsvSignal('/Battery/Devices/WarningFlags/vsM1', c.read_bool(register=1005, bit=24)),
CsvSignal('/Battery/Devices/WarningFlags/iCM1', c.read_bool(register=1005, bit=26)),
CsvSignal('/Battery/Devices/WarningFlags/iDM1', c.read_bool(register=1005, bit=28)),
CsvSignal('/Battery/Devices/WarningFlags/MID1', c.read_bool(register=1005, bit=30)),
CsvSignal('/Battery/Devices/WarningFlags/BLPW', c.read_bool(register=1005, bit=32)),
CsvSignal('/Battery/Devices/WarningFlags/CCBF', c.read_bool(register=1005, bit=33)),
CsvSignal('/Battery/Devices/WarningFlags/Ah_W', c.read_bool(register=1005, bit=35)),
CsvSignal('/Battery/Devices/WarningFlags/MPMM', c.read_bool(register=1005, bit=38)),
CsvSignal('/Battery/Devices/WarningFlags/TCdi', c.read_bool(register=1005, bit=40)),
CsvSignal('/Battery/Devices/WarningFlags/LMPW', c.read_bool(register=1005, bit=44)),
CsvSignal('/Battery/Devices/WarningFlags/TOCW', c.read_bool(register=1005, bit=47)),
CsvSignal('/Battery/Devices/WarningFlags/BUSL', c.read_bool(register=1005, bit=49)),
# Alarms
CsvSignal('/Battery/Devices/AlarmFlags/Tam', c.read_bool(register=1005, bit=0)),
CsvSignal('/Battery/Devices/AlarmFlags/TaM2', c.read_bool(register=1005, bit=2)),
CsvSignal('/Battery/Devices/AlarmFlags/Tbm', c.read_bool(register=1005, bit=3)),
CsvSignal('/Battery/Devices/AlarmFlags/TbM2', c.read_bool(register=1005, bit=5)),
CsvSignal('/Battery/Devices/AlarmFlags/VBm2', c.read_bool(register=1005, bit=7)),
CsvSignal('/Battery/Devices/AlarmFlags/VBM2', c.read_bool(register=1005, bit=9)),
CsvSignal('/Battery/Devices/AlarmFlags/IDM2', c.read_bool(register=1005, bit=11)),
CsvSignal('/Battery/Devices/AlarmFlags/ISOB', c.read_bool(register=1005, bit=12)),
CsvSignal('/Battery/Devices/AlarmFlags/MSWE', c.read_bool(register=1005, bit=13)),
CsvSignal('/Battery/Devices/AlarmFlags/FUSE', c.read_bool(register=1005, bit=14)),
CsvSignal('/Battery/Devices/AlarmFlags/HTRE', c.read_bool(register=1005, bit=15)),
CsvSignal('/Battery/Devices/AlarmFlags/TCPE', c.read_bool(register=1005, bit=16)),
CsvSignal('/Battery/Devices/AlarmFlags/STRE', c.read_bool(register=1005, bit=17)),
CsvSignal('/Battery/Devices/AlarmFlags/CME', c.read_bool(register=1005, bit=18)),
CsvSignal('/Battery/Devices/AlarmFlags/HWFL', c.read_bool(register=1005, bit=19)),
CsvSignal('/Battery/Devices/AlarmFlags/HWEM', c.read_bool(register=1005, bit=20)),
CsvSignal('/Battery/Devices/AlarmFlags/ThM', c.read_bool(register=1005, bit=21)),
CsvSignal('/Battery/Devices/AlarmFlags/vsm2', c.read_bool(register=1005, bit=23)),
CsvSignal('/Battery/Devices/AlarmFlags/vsM2', c.read_bool(register=1005, bit=25)),
CsvSignal('/Battery/Devices/AlarmFlags/iCM2', c.read_bool(register=1005, bit=27)),
CsvSignal('/Battery/Devices/AlarmFlags/iDM2', c.read_bool(register=1005, bit=29)),
CsvSignal('/Battery/Devices/AlarmFlags/MID2', c.read_bool(register=1005, bit=31)),
CsvSignal('/Battery/Devices/AlarmFlags/HTFS', c.read_bool(register=1005, bit=42)),
CsvSignal('/Battery/Devices/AlarmFlags/DATA', c.read_bool(register=1005, bit=43)),
CsvSignal('/Battery/Devices/AlarmFlags/LMPA', c.read_bool(register=1005, bit=45)),
CsvSignal('/Battery/Devices/AlarmFlags/HEBT', c.read_bool(register=1005, bit=46)),
CsvSignal('/Battery/Devices/AlarmFlags/CURM', c.read_bool(register=1005, bit=48)),
]
def init_modbus(tty):
# type: (str) -> Modbus
logging.debug('initializing Modbus')
return Modbus(
port='/dev/' + tty,
method=cfg.MODE,
baudrate=cfg.BAUD_RATE,
stopbits=cfg.STOP_BITS,
bytesize=cfg.BYTE_SIZE,
timeout=cfg.TIMEOUT,
parity=cfg.PARITY)
def read_modbus_registers(modbus, slave_address, base_address=cfg.BASE_ADDRESS, count=cfg.NO_OF_REGISTERS):
# type: (Modbus, int) -> ReadInputRegistersResponse
logging.debug('requesting modbus registers {0}-{1}'.format(base_address, base_address + count))
return modbus.read_input_registers(
address=base_address,
count=count,
unit=slave_address)
def read_firmware_version(modbus, slave_address):
# type: (Modbus, int) -> str
logging.debug('reading firmware version')
try:
modbus.connect()
response = read_modbus_registers(modbus, slave_address, base_address=1054, count=1)
register = response.registers[0]
return '{0:0>4X}'.format(register)
finally:
modbus.close() # close in any case
def init_main_loop():
# type: () -> DBusGMainLoop
logging.debug('initializing DBusGMainLoop Loop')
DBusGMainLoop(set_as_default=True)
return GLib.MainLoop()
def report_slave_id(modbus, slave_address):
# type: (Modbus, int) -> str
slave = str(slave_address)
logging.debug('requesting slave id from node ' + slave)
try:
modbus.connect()
request = ReportSlaveIdRequest(unit=slave_address)
response = modbus.execute(request)
if response is ExceptionResponse or issubclass(type(response), ModbusException):
raise Exception('failed to get slave id from ' + slave + ' : ' + str(response))
return response.identifier
finally:
modbus.close()
def parse_slave_id(modbus, slave_address):
# type: (Modbus, int) -> (str, str, int)
slave_id = report_slave_id(modbus, slave_address)
sid = re.sub(b'[^\x20-\x7E]', b'', slave_id) # remove weird special chars
match = re.match('(?P<hw>48TL(?P<ah>\d+)) *(?P<bms>.*)', sid.decode('ascii'))
if match is None:
raise Exception('no known battery found')
return match.group('hw'), match.group('bms'), int(match.group('ah'))
def identify_battery(modbus, slave_address):
# type: (Modbus, int) -> Battery
logging.info('identifying battery...')
hardware_version, bms_version, ampere_hours = parse_slave_id(modbus, slave_address)
firmware_version = read_firmware_version(modbus, slave_address)
specs = Battery(
slave_address=slave_address,
hardware_version=hardware_version,
firmware_version=firmware_version,
bms_version=bms_version,
ampere_hours=ampere_hours)
logging.info('battery identified:\n{0}'.format(str(specs)))
return specs
def identify_batteries(modbus):
# type: (Modbus) -> list[Battery]
def _identify_batteries():
address_range = range(1, cfg.MAX_SLAVE_ADDRESS + 1)
for slave_address in address_range:
try:
yield identify_battery(modbus, slave_address)
except Exception as e:
logging.info('failed to identify battery at {0} : {1}'.format(str(slave_address), str(e)))
return list(_identify_batteries()) # force that lazy iterable!
def read_modbus_registers(modbus, slave_address, base_address=cfg.BASE_ADDRESS, count=cfg.NO_OF_REGISTERS):
# type: (Modbus, int) -> ReadInputRegistersResponse
logging.debug('requesting modbus registers {0}-{1}'.format(base_address, base_address + count))
return modbus.read_input_registers(
address=base_address,
count=count,
unit=slave_address)
def read_battery_status(modbus, battery):
# type: (Modbus, Battery) -> BatteryStatus
"""
Read the modbus registers containing the battery's status info.
"""
logging.debug('reading battery status')
try:
modbus.connect()
data = read_modbus_registers(modbus, battery.slave_address)
return BatteryStatus(battery, data.registers)
finally:
modbus.close() # close in any case
def get_installation_name(file_path):
with open(file_path, 'r') as file:
return file.read().strip()
def manage_csv_files(directory_path, max_files=20):
csv_files = [f for f in os.listdir(directory_path)]
csv_files.sort(key=lambda x: os.path.getctime(os.path.join(directory_path, x)))
# Remove oldest files if exceeds maximum
while len(csv_files) > max_files:
file_to_delete = os.path.join(directory_path, csv_files.pop(0))
os.remove(file_to_delete)
def serialize_for_csv(value):
if isinstance(value, (dict, list, tuple)):
return json.dumps(value, ensure_ascii=False)
return str(value)
def insert_id(path, id_number):
parts = path.split("/")
insert_position = parts.index("Devices") + 1
parts.insert(insert_position, str(id_number))
return "/".join(parts)
def create_csv_files(signals, statuses, node_numbers):
timestamp = int(time.time())
if timestamp % 2 != 0:
timestamp -= 1
# Create CSV directory if it doesn't exist
if not os.path.exists(CSV_DIR):
os.makedirs(CSV_DIR)
#installation_name = get_installation_name(INSTALLATION_NAME_FILE)
csv_filename = f"{timestamp}.csv"
csv_path = os.path.join(CSV_DIR, csv_filename)
# Append values to the CSV file
with open(csv_path, 'a', newline='') as csvfile:
csv_writer = csv.writer(csvfile, delimiter=';')
# Add a special row for the nodes configuration
nodes_config_path = "/Config/Devices/BatteryNodes"
nodes_list = ",".join(str(node) for node in node_numbers)
config_row = [nodes_config_path, nodes_list, ""]
csv_writer.writerow(config_row)
# Iterate over each node and signal to create rows in the new format
for i, node in enumerate(node_numbers):
for s in signals:
signal_name = insert_id(s.name, i+1)
#value = serialize_for_csv(s.get_value(statuses[i]))
value = s.get_value(statuses[i])
row_values = [signal_name, value, s.get_text]
csv_writer.writerow(row_values)
# Manage CSV files, keep a limited number of files
# Create the CSV as a string
csv_data = read_csv_as_string(csv_path)
# Create an S3config instance
s3_config = S3config()
response = s3_config.create_put_request(csv_filename, csv_data)
if response.status_code == 200:
os.remove(csv_path)
print("Success")
else:
failed_dir = os.path.join(CSV_DIR, "failed")
if not os.path.exists(failed_dir):
os.makedirs(failed_dir)
failed_path = os.path.join(failed_dir, csv_filename)
os.rename(csv_path, failed_path)
print("Uploading failed")
manage_csv_files(failed_dir, 10)
manage_csv_files(CSV_DIR)
def update(modbus, batteries, csv_signals):
# type: (Modbus, Iterable[Battery], DBus, Iterable[Signal]) -> bool
"""
Main update function
1. requests status record each battery via modbus,
2. parses the data using Signal.get_value
3. aggregates the data from all batteries into one datum using Signal.aggregate
4. publishes the data on the dbus
"""
logging.debug('starting update cycle')
statuses = [read_battery_status(modbus, battery) for battery in batteries]
node_numbers = [battery.slave_address for battery in batteries]
create_csv_files(csv_signals, statuses, node_numbers)
logging.debug('finished update cycle\n')
return True
def print_usage():
print ('Usage: ' + __file__ + ' <serial device>')
print ('Example: ' + __file__ + ' ttyUSB0')
def parse_cmdline_args(argv):
# type: (list[str]) -> str
if len(argv) == 0:
logging.info('missing command line argument for tty device')
print_usage()
sys.exit(1)
return argv[0]
alive = True # global alive flag, watchdog_task clears it, update_task sets it
def create_update_task(modbus, batteries, csv_signals, main_loop):
# type: (Modbus, DBus, Iterable[Battery], Iterable[Signal], DBusGMainLoop) -> Callable[[],bool]
"""
Creates an update task which runs the main update function
and resets the alive flag
"""
def update_task():
# type: () -> bool
global alive
alive = update(modbus, batteries, csv_signals)
if not alive:
logging.info('update_task: quitting main loop because of error')
main_loop.quit()
return alive
return update_task
def create_watchdog_task(main_loop):
# type: (DBusGMainLoop) -> Callable[[],bool]
"""
Creates a Watchdog task that monitors the alive flag.
The watchdog kills the main loop if the alive flag is not periodically reset by the update task.
Who watches the watchdog?
"""
def watchdog_task():
# type: () -> bool
global alive
if alive:
logging.debug('watchdog_task: update_task is alive')
alive = False
return True
else:
logging.info('watchdog_task: killing main loop because update_task is no longer alive')
main_loop.quit()
return False
return watchdog_task
def main(argv):
# type: (list[str]) -> ()
print("PAME")
logging.basicConfig(level=cfg.LOG_LEVEL)
logging.info('starting ' + __file__)
tty = parse_cmdline_args(argv)
modbus = init_modbus(tty)
batteries = identify_batteries(modbus)
n = len(batteries)
logging.info('found ' + str(n) + (' battery' if n == 1 else ' batteries'))
if n <= 0:
sys.exit(2)
bat = c.first(batteries) # report hw and fw version of first battery found
csv_signals = create_csv_signals(bat.firmware_version)
main_loop = init_main_loop() # must run before init_dbus because gobject does some global magic
# we do not use dbus this time. we only want modbus
update_task = create_update_task(modbus, batteries, csv_signals, main_loop)
watchdog_task = create_watchdog_task(main_loop)
GLib.timeout_add(cfg.UPDATE_INTERVAL * 2, watchdog_task) # add watchdog first
GLib.timeout_add(cfg.UPDATE_INTERVAL, update_task) # call update once every update_interval
logging.info('starting GLib.MainLoop')
main_loop.run()
logging.info('GLib.MainLoop was shut down')
sys.exit(0xFF) # reaches this only on error
if __name__ == "__main__":
main(sys.argv[1:])

View File

@ -0,0 +1,7 @@
#!/bin/bash
. /opt/victronenergy/serial-starter/run-service.sh
app=/opt/victronenergy/dbus-csv-files/dbus-csv-files.py
args="$tty"
start $args

View File

@ -0,0 +1,51 @@
import serial
import logging
# dbus configuration
FIRMWARE_VERSION = 1 # value returned by getValue (getText returns string value reported by battery)
HARDWARE_VERSION = 1 # value returned by getValue (getText returns string value reported by battery)
CONNECTION = 'Modbus RTU'
PRODUCT_NAME = 'FZS 48TL200'
PRODUCT_ID = 0xB012 # assigned by victron
DEVICE_INSTANCE = 1
SERVICE_NAME_PREFIX = 'com.victronenergy.battery.'
# driver configuration
SOFTWARE_VERSION = '3.0.3'
UPDATE_INTERVAL = 2000 # milliseconds
#LOG_LEVEL = logging.INFO
LOG_LEVEL = logging.DEBUG
# modbus configuration
BASE_ADDRESS = 999
#NO_OF_REGISTERS = 63
NO_OF_REGISTERS = 64
MAX_SLAVE_ADDRESS = 10
# RS 485 configuration
PARITY = serial.PARITY_ODD
TIMEOUT = 0.1 # seconds
BAUD_RATE = 115200
BYTE_SIZE = 8
STOP_BITS = 1
MODE = 'rtu'
# battery configuration
MAX_CHARGE_VOLTAGE = 58
I_MAX_PER_STRING = 15
NUM_OF_STRING_PER_BATTERY = 5
AH_PER_STRING = 40
V_MAX = 54.2
R_STRING_MIN = 0.125
R_STRING_MAX = 0.250

View File

@ -0,0 +1,119 @@
from collections import Iterable
from decimal import *
import config as cfg
from data import LedState, BatteryStatus
# trick the pycharm type-checker into thinking Callable is in scope, not used at runtime
# noinspection PyUnreachableCode
if False:
from typing import Callable
def read_bool(register, bit):
# type: (int, int) -> Callable[[BatteryStatus], bool]
def get_value(status):
# type: (BatteryStatus) -> bool
value = status.modbus_data[register - cfg.BASE_ADDRESS]
return value & (1 << bit) > 0
return get_value
def read_float(register, scale_factor=1.0, offset=0.0, places=2):
# type: (int, float, float) -> Callable[[BatteryStatus], float]
def get_value(status):
# type: (BatteryStatus) -> float
value = status.modbus_data[register - cfg.BASE_ADDRESS]
if value >= 0x8000: # convert to signed int16
value -= 0x10000 # fiamm stores their integers signed AND with sign-offset @#%^&!
result = (value+offset)*scale_factor
return round(result,places)
return get_value
def read_hex_string(register, count):
# type: (int, int) -> Callable[[BatteryStatus], str]
"""
reads count consecutive modbus registers from start_address,
and returns a hex representation of it:
e.g. for count=4: DEAD BEEF DEAD BEEF.
"""
start = register - cfg.BASE_ADDRESS
end = start + count
def get_value(status):
# type: (BatteryStatus) -> str
return ' '.join(['{0:0>4X}'.format(x) for x in status.modbus_data[start:end]])
return get_value
def read_led_state(register, led):
# type: (int, int) -> Callable[[BatteryStatus], int]
read_lo = read_bool(register, led * 2)
read_hi = read_bool(register, led * 2 + 1)
def get_value(status):
# type: (BatteryStatus) -> int
lo = read_lo(status)
hi = read_hi(status)
if hi:
if lo:
return LedState.blinking_fast
else:
return LedState.blinking_slow
else:
if lo:
return LedState.on
else:
return LedState.off
return get_value
def read_bitmap(register):
# type: (int) -> Callable[[BatteryStatus], bitmap]
def get_value(status):
# type: (BatteryStatus) -> bitmap
value = status.modbus_data[register - cfg.BASE_ADDRESS]
return value
return get_value
def append_unit(unit):
# type: (unicode) -> Callable[[unicode], unicode]
def get_text(v):
# type: (unicode) -> unicode
return "{0}{1}".format(str(v), unit)
return get_text
def mean(numbers):
# type: (Iterable[float] | Iterable[int]) -> float
return float("{:.2f}".format(float(sum(numbers)) / len(numbers)))
def ssum(numbers):
# type: (Iterable[float] | Iterable[int]) -> float
return float("{:.2f}".format(float(sum(numbers))))
def first(ts):
return next(t for t in ts)
def return_in_list(ts):
return ts

View File

@ -0,0 +1,97 @@
import config as cfg
from collections import Iterable
# trick the pycharm type-checker into thinking Callable is in scope, not used at runtime
# noinspection PyUnreachableCode
if False:
from typing import Callable
class LedState(object):
"""
from page 6 of the '48TLxxx ModBus Protocol doc'
"""
off = 0
on = 1
blinking_slow = 2
blinking_fast = 3
class LedColor(object):
green = 0
amber = 1
blue = 2
red = 3
class CsvSignal(object):
def __init__(self, name, get_value, get_text = None):
self.name = name
self.get_value = get_value if callable(get_value) else lambda _: get_value
self.get_text = get_text
if get_text is None:
self.get_text = ""
class Signal(object):
def __init__(self, dbus_path, aggregate, get_value, get_text=None):
# type: (str, Callable[[Iterable[object]],object], Callable[[BatteryStatus],object] | object, Callable[[object],unicode] | object)->None
"""
A Signal holds all information necessary for the handling of a
certain datum (e.g. voltage) published by the battery.
:param dbus_path: str
object_path on DBus where the datum needs to be published
:param aggregate: Iterable[object] -> object
function that combines the values of multiple batteries into one.
e.g. sum for currents, or mean for voltages
:param get_value: (BatteryStatus) -> object
function to extract the datum from the modbus record,
alternatively: a constant
:param get_text: (object) -> unicode [optional]
function to render datum to text, needed by DBus
alternatively: a constant
"""
self.dbus_path = dbus_path
self.aggregate = aggregate
self.get_value = get_value if callable(get_value) else lambda _: get_value
self.get_text = get_text if callable(get_text) else lambda _: str(get_text)
# if no 'get_text' provided use 'default_text' if available, otherwise str()
if get_text is None:
self.get_text = str
class Battery(object):
""" Data record to hold hardware and firmware specs of the battery """
def __init__(self, slave_address, hardware_version, firmware_version, bms_version, ampere_hours):
# type: (int, str, str, str, int) -> None
self.slave_address = slave_address
self.hardware_version = hardware_version
self.firmware_version = firmware_version
self.bms_version = bms_version
self.ampere_hours = ampere_hours
def __str__(self):
return 'slave address = {0}\nhardware version = {1}\nfirmware version = {2}\nbms version = {3}\nampere hours = {4}'.format(
self.slave_address, self.hardware_version, self.firmware_version, self.bms_version, str(self.ampere_hours))
class BatteryStatus(object):
"""
record holding the current status of a battery
"""
def __init__(self, battery, modbus_data):
# type: (Battery, list[int]) -> None
self.battery = battery
self.modbus_data = modbus_data

View File

@ -0,0 +1,980 @@
#!/usr/bin/python3 -u
# coding=utf-8
import re
import sys
import logging
from gi.repository import GLib
import config as cfg
import convert as c
from pymodbus.register_read_message import ReadInputRegistersResponse
from pymodbus.client.sync import ModbusSerialClient as Modbus
from pymodbus.other_message import ReportSlaveIdRequest
from pymodbus.exceptions import ModbusException
from pymodbus.pdu import ExceptionResponse
from dbus.mainloop.glib import DBusGMainLoop
from data import BatteryStatus, Signal, Battery, LedColor, CsvSignal, LedState
from collections import Iterable
from os import path
app_dir = path.dirname(path.realpath(__file__))
sys.path.insert(1, path.join(app_dir, 'ext', 'velib_python'))
from vedbus import VeDbusService as DBus
import time
import os
import csv
import requests
import hmac
import hashlib
import base64
from datetime import datetime
import io
class S3config:
def __init__(self):
self.bucket = "1-c0436b6a-d276-4cd8-9c44-1eae86cf5d0e"
self.region = "sos-ch-dk-2"
self.provider = "exo.io"
self.key = "EXOcc0e47a4c4d492888ff5a7f2"
self.secret = "79QG4unMh7MeVacMnXr5xGxEyAlWZDIdM-dg_nXFFr4"
self.content_type = "text/plain; charset=utf-8"
@property
def host(self):
return f"{self.bucket}.{self.region}.{self.provider}"
@property
def url(self):
return f"https://{self.host}"
def create_put_request(self, s3_path, data):
headers = self._create_request("PUT", s3_path)
url = f"{self.url}/{s3_path}"
response = requests.put(url, headers=headers, data=data)
return response
def _create_request(self, method, s3_path):
date = datetime.utcnow().strftime('%a, %d %b %Y %H:%M:%S GMT')
auth = self._create_authorization(method, self.bucket, s3_path, date, self.key, self.secret, self.content_type)
headers = {
"Host": self.host,
"Date": date,
"Authorization": auth,
"Content-Type": self.content_type
}
return headers
@staticmethod
def _create_authorization(method, bucket, s3_path, date, s3_key, s3_secret, content_type="", md5_hash=""):
payload = f"{method}\n{md5_hash}\n{content_type}\n{date}\n/{bucket.strip('/')}/{s3_path.strip('/')}"
signature = base64.b64encode(
hmac.new(s3_secret.encode(), payload.encode(), hashlib.sha1).digest()
).decode()
return f"AWS {s3_key}:{signature}"
def read_csv_as_string(file_path):
"""
Reads a CSV file from the given path and returns its content as a single string.
"""
try:
with open(file_path, 'r', encoding='utf-8') as file:
return file.read()
except FileNotFoundError:
print(f"Error: The file {file_path} does not exist.")
return None
except IOError as e:
print(f"IO error occurred: {str(e)}")
return None
CSV_DIR = "/data/csv_files/"
#CSV_DIR = "csv_files/"
# Define the path to the file containing the installation name
INSTALLATION_NAME_FILE = '/data/innovenergy/openvpn/installation-name'
# trick the pycharm type-checker into thinking Callable is in scope, not used at runtime
# noinspection PyUnreachableCode
if False:
from typing import Callable
def interpret_limb_bitmap(bitmap_value):
# The bit for string 1 also monitors all 5 strings: 0000 0000 means All 5 strings activated. 0000 0001 means string 1 disabled.
string1_disabled = int((bitmap_value & 0b00001) != 0)
string2_disabled = int((bitmap_value & 0b00010) != 0)
string3_disabled = int((bitmap_value & 0b00100) != 0)
string4_disabled = int((bitmap_value & 0b01000) != 0)
string5_disabled = int((bitmap_value & 0b10000) != 0)
n_limb_strings = string1_disabled+string2_disabled+string3_disabled+string4_disabled+string5_disabled
return n_limb_strings
def create_csv_signals(firmware_version):
def read_power(status):
return int(read_current(status) * read_voltage(status))
read_voltage = c.read_float(register=999, scale_factor=0.01, offset=0, places=2)
read_current = c.read_float(register=1000, scale_factor=0.01, offset=-10000, places=2)
read_limb_bitmap = c.read_bitmap(1059)
def string1_disabled(status):
bitmap_value = read_limb_bitmap(status)
return int((bitmap_value & 0b00001) != 0)
def string2_disabled(status):
bitmap_value = read_limb_bitmap(status)
return int((bitmap_value & 0b00010) != 0)
def string3_disabled(status):
bitmap_value = read_limb_bitmap(status)
return int((bitmap_value & 0b00100) != 0)
def string4_disabled(status):
bitmap_value = read_limb_bitmap(status)
return int((bitmap_value & 0b01000) != 0)
def string5_disabled(status):
bitmap_value = read_limb_bitmap(status)
return int((bitmap_value & 0b10000) != 0)
def limp_strings_value(status):
return interpret_limb_bitmap(read_limb_bitmap(status))
def calc_power_limit_imposed_by_voltage_limit(v, i, v_limit, r_int):
# type: (float, float, float, float) -> float
dv = v_limit - v
di = dv / r_int
p_limit = v_limit * (i + di)
return p_limit
def calc_power_limit_imposed_by_current_limit(v, i, i_limit, r_int):
# type: (float, float, float, float) -> float
di = i_limit - i
dv = di * r_int
p_limit = i_limit * (v + dv)
return p_limit
def calc_max_charge_power(status):
# type: (BatteryStatus) -> int
n_strings = cfg.NUM_OF_STRING_PER_BATTERY-limp_strings_value(status)
i_max = n_strings * cfg.I_MAX_PER_STRING
v_max = cfg.V_MAX
r_int_min = cfg.R_STRING_MIN / n_strings
r_int_max = cfg.R_STRING_MAX / n_strings
v = read_voltage(status)
i = read_current(status)
p_limits = [
calc_power_limit_imposed_by_voltage_limit(v, i, v_max,r_int_min),
calc_power_limit_imposed_by_voltage_limit(v, i, v_max, r_int_max),
calc_power_limit_imposed_by_current_limit(v, i, i_max, r_int_min),
calc_power_limit_imposed_by_current_limit(v, i, i_max, r_int_max),
]
p_limit = min(p_limits) # p_limit is normally positive here (signed)
p_limit = max(p_limit, 0) # charge power must not become negative
return int(p_limit)
def calc_max_discharge_power(status):
n_strings = cfg.NUM_OF_STRING_PER_BATTERY-limp_strings_value(status)
max_discharge_current = n_strings*cfg.I_MAX_PER_STRING
return int(max_discharge_current*read_voltage(status))
def return_led_state_blue(status):
led_state = c.read_led_state(register=1004, led=LedColor.blue)(status)
if led_state == LedState.blinking_fast or led_state == LedState.blinking_slow:
return "Blinking"
elif led_state == LedState.on:
return "On"
elif led_state == LedState.off:
return "Off"
return "Unknown"
def return_led_state_red(status):
led_state = c.read_led_state(register=1004, led=LedColor.red)(status)
if led_state == LedState.blinking_fast or led_state == LedState.blinking_slow:
return "Blinking"
elif led_state == LedState.on:
return "On"
elif led_state == LedState.off:
return "Off"
return "Unknown"
def return_led_state_green(status):
led_state = c.read_led_state(register=1004, led=LedColor.green)(status)
if led_state == LedState.blinking_fast or led_state == LedState.blinking_slow:
return "Blinking"
elif led_state == LedState.on:
return "On"
elif led_state == LedState.off:
return "Off"
return "Unknown"
def return_led_state_amber(status):
led_state = c.read_led_state(register=1004, led=LedColor.amber)(status)
if led_state == LedState.blinking_fast or led_state == LedState.blinking_slow:
return "Blinking"
elif led_state == LedState.on:
return "On"
elif led_state == LedState.off:
return "Off"
return "Unknown"
total_current = c.read_float(register=1062, scale_factor=0.01, offset=-10000, places=1)
def read_total_current(status):
return total_current(status)
def read_heating_current(status):
return total_current(status) - read_current(status)
def read_heating_power(status):
return read_voltage(status) * read_heating_current(status)
soc_ah = c.read_float(register=1002, scale_factor=0.1, offset=-10000, places=1)
def read_soc_ah(status):
return soc_ah(status)
def hex_string_to_ascii(hex_string):
# Ensure the hex_string is correctly formatted without spaces
hex_string = hex_string.replace(" ", "")
# Convert every two characters (a byte) in the hex string to ASCII
ascii_string = ''.join([chr(int(hex_string[i:i+2], 16)) for i in range(0, len(hex_string), 2)])
return ascii_string
battery_status_reader = c.read_hex_string(1060,2)
def read_eoc_reached(status):
battery_status_string = battery_status_reader(status)
#if hex_string_to_ascii(battery_status_string) == "EOC_":
#return True
#return False
return hex_string_to_ascii(battery_status_string) == "EOC_"
def read_serial_number(status):
serial_regs = [1055, 1056, 1057, 1058]
serial_parts = []
for reg in serial_regs:
# reading each register as a single hex value
hex_value_fun = c.read_hex_string(reg, 1)
hex_value = hex_value_fun(status)
# append without spaces and leading zeros stripped if any
serial_parts.append(hex_value.replace(' ', ''))
# concatenate all parts to form the full serial number
serial_number = ''.join(serial_parts).rstrip('0')
return serial_number
return [
CsvSignal('/Battery/Devices/FwVersion', firmware_version),
CsvSignal('/Battery/Devices/Dc/Power', read_power, 'W'),
CsvSignal('/Battery/Devices/Dc/Voltage', read_voltage, 'V'),
CsvSignal('/Battery/Devices/Soc', c.read_float(register=1053, scale_factor=0.1, offset=0, places=1), '%'),
CsvSignal('/Battery/Devices/Temperatures/Cells/Average', c.read_float(register=1003, scale_factor=0.1, offset=-400, places=1), 'C'),
CsvSignal('/Battery/Devices/Dc/Current', read_current, 'A'),
CsvSignal('/Battery/Devices/BusCurrent', read_total_current, 'A'),
CsvSignal('/Battery/Devices/CellsCurrent', read_current, 'A'),
CsvSignal('/Battery/Devices/HeatingCurrent', read_heating_current, 'A'),
CsvSignal('/Battery/Devices/HeatingPower', read_heating_power, 'W'),
CsvSignal('/Battery/Devices/SOCAh', read_soc_ah),
CsvSignal('/Battery/Devices/Leds/Blue', return_led_state_blue),
CsvSignal('/Battery/Devices/Leds/Red', return_led_state_red),
CsvSignal('/Battery/Devices/Leds/Green', return_led_state_green),
CsvSignal('/Battery/Devices/Leds/Amber', return_led_state_amber),
CsvSignal('/Battery/Devices/BatteryStrings/String1Active', string1_disabled),
CsvSignal('/Battery/Devices/BatteryStrings/String2Active', string2_disabled),
CsvSignal('/Battery/Devices/BatteryStrings/String3Active', string3_disabled),
CsvSignal('/Battery/Devices/BatteryStrings/String4Active', string4_disabled),
CsvSignal('/Battery/Devices/BatteryStrings/String5Active', string5_disabled),
CsvSignal('/Battery/Devices/IoStatus/ConnectedToDcBus', c.read_bool(register=1013, bit=0)),
CsvSignal('/Battery/Devices/IoStatus/AlarmOutActive', c.read_bool(register=1013, bit=1)),
CsvSignal('/Battery/Devices/IoStatus/InternalFanActive', c.read_bool(register=1013, bit=2)),
CsvSignal('/Battery/Devices/IoStatus/VoltMeasurementAllowed', c.read_bool(register=1013, bit=3)),
CsvSignal('/Battery/Devices/IoStatus/AuxRelayBus', c.read_bool(register=1013, bit=4)),
CsvSignal('/Battery/Devices/IoStatus/RemoteStateActive', c.read_bool(register=1013, bit=5)),
CsvSignal('/Battery/Devices/IoStatus/RiscActive', c.read_bool(register=1013, bit=6)),
CsvSignal('/Battery/Devices/Eoc', read_eoc_reached),
CsvSignal('/Battery/Devices/SerialNumber', read_serial_number),
CsvSignal('/Battery/Devices/TimeSinceTOC', c.read_float(register=1052)),
CsvSignal('/Battery/Devices/MaxChargePower', calc_max_charge_power),
CsvSignal('/Battery/Devices/MaxDischargePower', calc_max_discharge_power),
# Warnings
CsvSignal('/Battery/Devices/WarningFlags/TaM1', c.read_bool(register=1005, bit=1)),
CsvSignal('/Battery/Devices/WarningFlags/TbM1', c.read_bool(register=1005, bit=4)),
CsvSignal('/Battery/Devices/WarningFlags/VBm1', c.read_bool(register=1005, bit=6)),
CsvSignal('/Battery/Devices/WarningFlags/VBM1', c.read_bool(register=1005, bit=8)),
CsvSignal('/Battery/Devices/WarningFlags/IDM1', c.read_bool(register=1005, bit=10)),
CsvSignal('/Battery/Devices/WarningFlags/vsm1', c.read_bool(register=1005, bit=22)),
CsvSignal('/Battery/Devices/WarningFlags/vsM1', c.read_bool(register=1005, bit=24)),
CsvSignal('/Battery/Devices/WarningFlags/iCM1', c.read_bool(register=1005, bit=26)),
CsvSignal('/Battery/Devices/WarningFlags/iDM1', c.read_bool(register=1005, bit=28)),
CsvSignal('/Battery/Devices/WarningFlags/MID1', c.read_bool(register=1005, bit=30)),
CsvSignal('/Battery/Devices/WarningFlags/BLPW', c.read_bool(register=1005, bit=32)),
CsvSignal('/Battery/Devices/WarningFlags/CCBF', c.read_bool(register=1005, bit=33)),
CsvSignal('/Battery/Devices/WarningFlags/Ah_W', c.read_bool(register=1005, bit=35)),
CsvSignal('/Battery/Devices/WarningFlags/MPMM', c.read_bool(register=1005, bit=38)),
CsvSignal('/Battery/Devices/WarningFlags/TCdi', c.read_bool(register=1005, bit=40)),
CsvSignal('/Battery/Devices/WarningFlags/LMPW', c.read_bool(register=1005, bit=44)),
CsvSignal('/Battery/Devices/WarningFlags/TOCW', c.read_bool(register=1005, bit=47)),
CsvSignal('/Battery/Devices/WarningFlags/BUSL', c.read_bool(register=1005, bit=49)),
# Alarms
CsvSignal('/Battery/Devices/AlarmFlags/Tam', c.read_bool(register=1005, bit=0)),
CsvSignal('/Battery/Devices/AlarmFlags/TaM2', c.read_bool(register=1005, bit=2)),
CsvSignal('/Battery/Devices/AlarmFlags/Tbm', c.read_bool(register=1005, bit=3)),
CsvSignal('/Battery/Devices/AlarmFlags/TbM2', c.read_bool(register=1005, bit=5)),
CsvSignal('/Battery/Devices/AlarmFlags/VBm2', c.read_bool(register=1005, bit=7)),
CsvSignal('/Battery/Devices/AlarmFlags/VBM2', c.read_bool(register=1005, bit=9)),
CsvSignal('/Battery/Devices/AlarmFlags/IDM2', c.read_bool(register=1005, bit=11)),
CsvSignal('/Battery/Devices/AlarmFlags/ISOB', c.read_bool(register=1005, bit=12)),
CsvSignal('/Battery/Devices/AlarmFlags/MSWE', c.read_bool(register=1005, bit=13)),
CsvSignal('/Battery/Devices/AlarmFlags/FUSE', c.read_bool(register=1005, bit=14)),
CsvSignal('/Battery/Devices/AlarmFlags/HTRE', c.read_bool(register=1005, bit=15)),
CsvSignal('/Battery/Devices/AlarmFlags/TCPE', c.read_bool(register=1005, bit=16)),
CsvSignal('/Battery/Devices/AlarmFlags/STRE', c.read_bool(register=1005, bit=17)),
CsvSignal('/Battery/Devices/AlarmFlags/CME', c.read_bool(register=1005, bit=18)),
CsvSignal('/Battery/Devices/AlarmFlags/HWFL', c.read_bool(register=1005, bit=19)),
CsvSignal('/Battery/Devices/AlarmFlags/HWEM', c.read_bool(register=1005, bit=20)),
CsvSignal('/Battery/Devices/AlarmFlags/ThM', c.read_bool(register=1005, bit=21)),
CsvSignal('/Battery/Devices/AlarmFlags/vsm2', c.read_bool(register=1005, bit=23)),
CsvSignal('/Battery/Devices/AlarmFlags/vsM2', c.read_bool(register=1005, bit=25)),
CsvSignal('/Battery/Devices/AlarmFlags/iCM2', c.read_bool(register=1005, bit=27)),
CsvSignal('/Battery/Devices/AlarmFlags/iDM2', c.read_bool(register=1005, bit=29)),
CsvSignal('/Battery/Devices/AlarmFlags/MID2', c.read_bool(register=1005, bit=31)),
CsvSignal('/Battery/Devices/AlarmFlags/HTFS', c.read_bool(register=1005, bit=42)),
CsvSignal('/Battery/Devices/AlarmFlags/DATA', c.read_bool(register=1005, bit=43)),
CsvSignal('/Battery/Devices/AlarmFlags/LMPA', c.read_bool(register=1005, bit=45)),
CsvSignal('/Battery/Devices/AlarmFlags/HEBT', c.read_bool(register=1005, bit=46)),
CsvSignal('/Battery/Devices/AlarmFlags/CURM', c.read_bool(register=1005, bit=48)),
]
def init_signals(hardware_version, firmware_version, n_batteries):
# type: (str,str,int) -> Iterable[Signal]
"""
A Signal holds all information necessary for the handling of a
certain datum (e.g. voltage) published by the battery.
Signal(dbus_path, aggregate, get_value, get_text = str)
dbus_path: str
object_path on DBus where the datum needs to be published
aggregate: Iterable[object] -> object
function that combines the values of multiple batteries into one.
e.g. sum for currents, or mean for voltages
get_value: (BatteryStatus) -> object [optional]
function to extract the datum from the modbus record,
alternatively: a constant
get_text: (object) -> unicode [optional]
function to render datum to text, needed by DBus
alternatively: a constant
The conversion functions use the same parameters (e.g scale_factor, offset)
as described in the document 'T48TLxxx ModBus Protocol Rev.7.1' which can
be found in the /doc folder
"""
product_id_hex = '0x{0:04x}'.format(cfg.PRODUCT_ID)
read_voltage = c.read_float(register=999, scale_factor=0.01, offset=0, places=2)
read_current = c.read_float(register=1000, scale_factor=0.01, offset=-10000, places=2)
def read_power(status):
return int(read_current(status) * read_voltage(status))
read_limb_bitmap = c.read_bitmap(1059)
def limp_strings_value(status):
return interpret_limb_bitmap(read_limb_bitmap(status))
def max_discharge_current(status):
return (cfg.NUM_OF_STRING_PER_BATTERY-limp_strings_value(status))*cfg.I_MAX_PER_STRING
def max_charge_current(status):
return status.battery.ampere_hours/2
def calc_power_limit_imposed_by_voltage_limit(v, i, v_limit, r_int):
# type: (float, float, float, float) -> float
dv = v_limit - v
di = dv / r_int
p_limit = v_limit * (i + di)
return p_limit
def calc_power_limit_imposed_by_current_limit(v, i, i_limit, r_int):
# type: (float, float, float, float) -> float
di = i_limit - i
dv = di * r_int
p_limit = i_limit * (v + dv)
return p_limit
def calc_max_charge_power(status):
# type: (BatteryStatus) -> int
n_strings = cfg.NUM_OF_STRING_PER_BATTERY-limp_strings_value(status)
i_max = n_strings * cfg.I_MAX_PER_STRING
v_max = cfg.V_MAX
r_int_min = cfg.R_STRING_MIN / n_strings
r_int_max = cfg.R_STRING_MAX / n_strings
v = read_voltage(status)
i = read_current(status)
p_limits = [
calc_power_limit_imposed_by_voltage_limit(v, i, v_max,r_int_min),
calc_power_limit_imposed_by_voltage_limit(v, i, v_max, r_int_max),
calc_power_limit_imposed_by_current_limit(v, i, i_max, r_int_min),
calc_power_limit_imposed_by_current_limit(v, i, i_max, r_int_max),
]
p_limit = min(p_limits) # p_limit is normally positive here (signed)
p_limit = max(p_limit, 0) # charge power must not become negative
return int(p_limit)
product_name = cfg.PRODUCT_NAME
if n_batteries > 1:
product_name = cfg.PRODUCT_NAME + ' x' + str(n_batteries)
return [
# Node Red related dbus paths
Signal('/TimeToTOCRequest', min, c.read_float(register=1052)),
Signal('/NumOfLimbStrings', c.return_in_list, get_value=limp_strings_value),
Signal('/NumOfBatteries', max, get_value=n_batteries),
Signal('/Dc/0/Voltage', c.mean, get_value=read_voltage, get_text=c.append_unit('V')),
Signal('/Dc/0/Current', c.ssum, get_value=read_current, get_text=c.append_unit('A')),
Signal('/Dc/0/Power', c.ssum, get_value=read_power, get_text=c.append_unit('W')),
Signal('/BussVoltage', c.mean, c.read_float(register=1001, scale_factor=0.01, offset=0, places=2), c.append_unit('V')),
Signal('/Soc', c.mean, c.read_float(register=1053, scale_factor=0.1, offset=0, places=1), c.append_unit('%')),
Signal('/LowestSoc', min, c.read_float(register=1053, scale_factor=0.1, offset=0, places=1), c.append_unit('%')),
Signal('/Dc/0/Temperature', c.mean, c.read_float(register=1003, scale_factor=0.1, offset=-400, places=1), c.append_unit(u'°C')),
Signal('/Dc/0/LowestTemperature', min, c.read_float(register=1003, scale_factor=0.1, offset=-400, places=1), c.append_unit(u'°C')),
# Charge/Discharge current, voltage and power
Signal('/Info/MaxDischargeCurrent', c.ssum, max_discharge_current,c.append_unit('A')),
Signal('/Info/MaxChargeCurrent', c.ssum, max_charge_current, c.append_unit('A')),
Signal('/Info/MaxChargeVoltage', min, cfg.MAX_CHARGE_VOLTAGE, c.append_unit('V')),
Signal('/Info/MaxChargePower', c.ssum, calc_max_charge_power),
# Victron mandatory dbus paths
Signal('/Mgmt/ProcessName', c.first, __file__),
Signal('/Mgmt/ProcessVersion', c.first, cfg.SOFTWARE_VERSION),
Signal('/Mgmt/Connection', c.first, cfg.CONNECTION),
Signal('/DeviceInstance', c.first, cfg.DEVICE_INSTANCE),
Signal('/ProductName', c.first, product_name),
Signal('/ProductId', c.first, cfg.PRODUCT_ID, product_id_hex),
Signal('/Connected', c.first, 1),
#Signal('/FirmwareVersion', c.first, cfg.FIRMWARE_VERSION, firmware_version),
Signal('/FirmwareVersion', c.return_in_list, firmware_version),
Signal('/HardwareVersion', c.first, cfg.HARDWARE_VERSION, hardware_version),
## Diagnostics
Signal('/Diagnostics/BmsVersion', c.first, lambda s: s.battery.bms_version),
# Warnings
#Signal('/Diagnostics/WarningFlags', c.first, c.read_hex_string(register=1005, count=4)),
Signal('/WarningFlags/TaM1', c.return_in_list, c.read_bool(register=1005, bit=1)),
Signal('/WarningFlags/TbM1', c.return_in_list, c.read_bool(register=1005, bit=4)),
Signal('/WarningFlags/VBm1', c.return_in_list, c.read_bool(register=1005, bit=6)),
Signal('/WarningFlags/VBM1', c.return_in_list, c.read_bool(register=1005, bit=8)),
Signal('/WarningFlags/IDM1', c.return_in_list, c.read_bool(register=1005, bit=10)),
Signal('/WarningFlags/vsm1', c.return_in_list, c.read_bool(register=1005, bit=22)),
Signal('/WarningFlags/vsM1', c.return_in_list, c.read_bool(register=1005, bit=24)),
Signal('/WarningFlags/iCM1', c.return_in_list, c.read_bool(register=1005, bit=26)),
Signal('/WarningFlags/iDM1', c.return_in_list, c.read_bool(register=1005, bit=28)),
Signal('/WarningFlags/MID1', c.return_in_list, c.read_bool(register=1005, bit=30)),
Signal('/WarningFlags/BLPW', c.return_in_list, c.read_bool(register=1005, bit=32)),
Signal('/WarningFlags/CCBF', c.return_in_list, c.read_bool(register=1005, bit=33)),
Signal('/WarningFlags/Ah_W', c.return_in_list, c.read_bool(register=1005, bit=35)),
Signal('/WarningFlags/MPMM', c.return_in_list, c.read_bool(register=1005, bit=38)),
#Signal('/WarningFlags/TCMM', c.return_in_list, c.read_bool(register=1005, bit=39)),
Signal('/WarningFlags/TCdi', c.return_in_list, c.read_bool(register=1005, bit=40)),
Signal('/WarningFlags/LMPW', c.return_in_list, c.read_bool(register=1005, bit=44)),
Signal('/WarningFlags/TOCW', c.return_in_list, c.read_bool(register=1005, bit=47)),
Signal('/WarningFlags/BUSL', c.return_in_list, c.read_bool(register=1005, bit=49)),
# Alarms
#Signal('/Diagnostics/AlarmFlags', c.first, c.read_hex_string(register=1009, count=4)),
Signal('/AlarmFlags/Tam', c.return_in_list, c.read_bool(register=1005, bit=0)),
Signal('/AlarmFlags/TaM2', c.return_in_list, c.read_bool(register=1005, bit=2)),
Signal('/AlarmFlags/Tbm', c.return_in_list, c.read_bool(register=1005, bit=3)),
Signal('/AlarmFlags/TbM2', c.return_in_list, c.read_bool(register=1005, bit=5)),
Signal('/AlarmFlags/VBm2', c.return_in_list, c.read_bool(register=1005, bit=7)),
Signal('/AlarmFlags/VBM2', c.return_in_list, c.read_bool(register=1005, bit=9)),
Signal('/AlarmFlags/IDM2', c.return_in_list, c.read_bool(register=1005, bit=11)),
Signal('/AlarmFlags/ISOB', c.return_in_list, c.read_bool(register=1005, bit=12)),
Signal('/AlarmFlags/MSWE', c.return_in_list, c.read_bool(register=1005, bit=13)),
Signal('/AlarmFlags/FUSE', c.return_in_list, c.read_bool(register=1005, bit=14)),
Signal('/AlarmFlags/HTRE', c.return_in_list, c.read_bool(register=1005, bit=15)),
Signal('/AlarmFlags/TCPE', c.return_in_list, c.read_bool(register=1005, bit=16)),
Signal('/AlarmFlags/STRE', c.return_in_list, c.read_bool(register=1005, bit=17)),
Signal('/AlarmFlags/CME', c.return_in_list, c.read_bool(register=1005, bit=18)),
Signal('/AlarmFlags/HWFL', c.return_in_list, c.read_bool(register=1005, bit=19)),
Signal('/AlarmFlags/HWEM', c.return_in_list, c.read_bool(register=1005, bit=20)),
Signal('/AlarmFlags/ThM', c.return_in_list, c.read_bool(register=1005, bit=21)),
Signal('/AlarmFlags/vsm2', c.return_in_list, c.read_bool(register=1005, bit=23)),
Signal('/AlarmFlags/vsM2', c.return_in_list, c.read_bool(register=1005, bit=25)),
Signal('/AlarmFlags/iCM2', c.return_in_list, c.read_bool(register=1005, bit=27)),
Signal('/AlarmFlags/iDM2', c.return_in_list, c.read_bool(register=1005, bit=29)),
Signal('/AlarmFlags/MID2', c.return_in_list, c.read_bool(register=1005, bit=31)),
#Signal('/AlarmFlags/TcBM', c.return_in_list, c.read_bool(register=1005, bit=36)),
#Signal('/AlarmFlags/BRNF', c.return_in_list, c.read_bool(register=1005, bit=37)),
Signal('/AlarmFlags/HTFS', c.return_in_list, c.read_bool(register=1005, bit=42)),
Signal('/AlarmFlags/DATA', c.return_in_list, c.read_bool(register=1005, bit=43)),
Signal('/AlarmFlags/LMPA', c.return_in_list, c.read_bool(register=1005, bit=45)),
Signal('/AlarmFlags/HEBT', c.return_in_list, c.read_bool(register=1005, bit=46)),
Signal('/AlarmFlags/CURM', c.return_in_list, c.read_bool(register=1005, bit=48)),
# LedStatus
Signal('/Diagnostics/LedStatus/Red', c.first, c.read_led_state(register=1004, led=LedColor.red)),
Signal('/Diagnostics/LedStatus/Blue', c.first, c.read_led_state(register=1004, led=LedColor.blue)),
Signal('/Diagnostics/LedStatus/Green', c.first, c.read_led_state(register=1004, led=LedColor.green)),
Signal('/Diagnostics/LedStatus/Amber', c.first, c.read_led_state(register=1004, led=LedColor.amber)),
# IO Status
Signal('/Diagnostics/IoStatus/MainSwitchClosed', c.return_in_list, c.read_bool(register=1013, bit=0)),
Signal('/Diagnostics/IoStatus/AlarmOutActive', c.return_in_list, c.read_bool(register=1013, bit=1)),
Signal('/Diagnostics/IoStatus/InternalFanActive', c.return_in_list, c.read_bool(register=1013, bit=2)),
Signal('/Diagnostics/IoStatus/VoltMeasurementAllowed', c.return_in_list, c.read_bool(register=1013, bit=3)),
Signal('/Diagnostics/IoStatus/AuxRelay', c.return_in_list, c.read_bool(register=1013, bit=4)),
Signal('/Diagnostics/IoStatus/RemoteState', c.return_in_list, c.read_bool(register=1013, bit=5)),
Signal('/Diagnostics/IoStatus/RiscOn', c.return_in_list, c.read_bool(register=1013, bit=6)),
]
def init_modbus(tty):
# type: (str) -> Modbus
logging.debug('initializing Modbus')
return Modbus(
port='/dev/' + tty,
method=cfg.MODE,
baudrate=cfg.BAUD_RATE,
stopbits=cfg.STOP_BITS,
bytesize=cfg.BYTE_SIZE,
timeout=cfg.TIMEOUT,
parity=cfg.PARITY)
def init_dbus(tty, signals):
# type: (str, Iterable[Signal]) -> DBus
logging.debug('initializing DBus service')
dbus = DBus(servicename=cfg.SERVICE_NAME_PREFIX + tty)
logging.debug('initializing DBus paths')
for signal in signals:
init_dbus_path(dbus, signal)
return dbus
# noinspection PyBroadException
def try_get_value(sig):
# type: (Signal) -> object
try:
return sig.get_value(None)
except:
return None
def init_dbus_path(dbus, sig):
# type: (DBus, Signal) -> ()
dbus.add_path(
sig.dbus_path,
try_get_value(sig),
gettextcallback=lambda _, v: sig.get_text(v))
def init_main_loop():
# type: () -> DBusGMainLoop
logging.debug('initializing DBusGMainLoop Loop')
DBusGMainLoop(set_as_default=True)
return GLib.MainLoop()
def report_slave_id(modbus, slave_address):
# type: (Modbus, int) -> str
slave = str(slave_address)
logging.debug('requesting slave id from node ' + slave)
try:
modbus.connect()
request = ReportSlaveIdRequest(unit=slave_address)
response = modbus.execute(request)
if response is ExceptionResponse or issubclass(type(response), ModbusException):
raise Exception('failed to get slave id from ' + slave + ' : ' + str(response))
return response.identifier
finally:
modbus.close()
def identify_battery(modbus, slave_address):
# type: (Modbus, int) -> Battery
logging.info('identifying battery...')
hardware_version, bms_version, ampere_hours = parse_slave_id(modbus, slave_address)
firmware_version = read_firmware_version(modbus, slave_address)
specs = Battery(
slave_address=slave_address,
hardware_version=hardware_version,
firmware_version=firmware_version,
bms_version=bms_version,
ampere_hours=ampere_hours)
logging.info('battery identified:\n{0}'.format(str(specs)))
return specs
def identify_batteries(modbus):
# type: (Modbus) -> list[Battery]
def _identify_batteries():
address_range = range(1, cfg.MAX_SLAVE_ADDRESS + 1)
for slave_address in address_range:
try:
yield identify_battery(modbus, slave_address)
except Exception as e:
logging.info('failed to identify battery at {0} : {1}'.format(str(slave_address), str(e)))
return list(_identify_batteries()) # force that lazy iterable!
def parse_slave_id(modbus, slave_address):
# type: (Modbus, int) -> (str, str, int)
slave_id = report_slave_id(modbus, slave_address)
sid = re.sub(b'[^\x20-\x7E]', b'', slave_id) # remove weird special chars
match = re.match('(?P<hw>48TL(?P<ah>\d+)) *(?P<bms>.*)', sid.decode('ascii'))
if match is None:
raise Exception('no known battery found')
return match.group('hw'), match.group('bms'), int(match.group('ah'))
def read_firmware_version(modbus, slave_address):
# type: (Modbus, int) -> str
logging.debug('reading firmware version')
try:
modbus.connect()
response = read_modbus_registers(modbus, slave_address, base_address=1054, count=1)
register = response.registers[0]
return '{0:0>4X}'.format(register)
finally:
modbus.close() # close in any case
def read_modbus_registers(modbus, slave_address, base_address=cfg.BASE_ADDRESS, count=cfg.NO_OF_REGISTERS):
# type: (Modbus, int) -> ReadInputRegistersResponse
logging.debug('requesting modbus registers {0}-{1}'.format(base_address, base_address + count))
return modbus.read_input_registers(
address=base_address,
count=count,
unit=slave_address)
def read_battery_status(modbus, battery):
# type: (Modbus, Battery) -> BatteryStatus
"""
Read the modbus registers containing the battery's status info.
"""
logging.debug('reading battery status')
try:
modbus.connect()
data = read_modbus_registers(modbus, battery.slave_address)
return BatteryStatus(battery, data.registers)
finally:
modbus.close() # close in any case
def publish_values(dbus, signals, statuses):
# type: (DBus, Iterable[Signal], Iterable[BatteryStatus]) -> ()
for s in signals:
values = [s.get_value(status) for status in statuses]
with dbus as srv:
srv[s.dbus_path] = s.aggregate(values)
def update(modbus, batteries, dbus, signals, csv_signals):
# type: (Modbus, Iterable[Battery], DBus, Iterable[Signal]) -> bool
"""
Main update function
1. requests status record each battery via modbus,
2. parses the data using Signal.get_value
3. aggregates the data from all batteries into one datum using Signal.aggregate
4. publishes the data on the dbus
"""
logging.debug('starting update cycle')
statuses = [read_battery_status(modbus, battery) for battery in batteries]
node_numbers = [battery.slave_address for battery in batteries]
publish_values(dbus, signals, statuses)
create_csv_files(csv_signals, statuses, node_numbers)
logging.debug('finished update cycle\n')
return True
def print_usage():
print ('Usage: ' + __file__ + ' <serial device>')
print ('Example: ' + __file__ + ' ttyUSB0')
def parse_cmdline_args(argv):
# type: (list[str]) -> str
if len(argv) == 0:
logging.info('missing command line argument for tty device')
print_usage()
sys.exit(1)
return argv[0]
alive = True # global alive flag, watchdog_task clears it, update_task sets it
def create_update_task(modbus, dbus, batteries, signals, csv_signals, main_loop):
# type: (Modbus, DBus, Iterable[Battery], Iterable[Signal], DBusGMainLoop) -> Callable[[],bool]
"""
Creates an update task which runs the main update function
and resets the alive flag
"""
def update_task():
# type: () -> bool
global alive
alive = update(modbus, batteries, dbus, signals, csv_signals)
if not alive:
logging.info('update_task: quitting main loop because of error')
main_loop.quit()
return alive
return update_task
def create_watchdog_task(main_loop):
# type: (DBusGMainLoop) -> Callable[[],bool]
"""
Creates a Watchdog task that monitors the alive flag.
The watchdog kills the main loop if the alive flag is not periodically reset by the update task.
Who watches the watchdog?
"""
def watchdog_task():
# type: () -> bool
global alive
if alive:
logging.debug('watchdog_task: update_task is alive')
alive = False
return True
else:
logging.info('watchdog_task: killing main loop because update_task is no longer alive')
main_loop.quit()
return False
return watchdog_task
def get_installation_name(file_path):
with open(file_path, 'r') as file:
return file.read().strip()
def manage_csv_files(directory_path, max_files=20):
csv_files = [f for f in os.listdir(directory_path)]
csv_files.sort(key=lambda x: os.path.getctime(os.path.join(directory_path, x)))
# Remove oldest files if exceeds maximum
while len(csv_files) > max_files:
file_to_delete = os.path.join(directory_path, csv_files.pop(0))
os.remove(file_to_delete)
def serialize_for_csv(value):
if isinstance(value, (dict, list, tuple)):
return json.dumps(value, ensure_ascii=False)
return str(value)
def insert_id(path, id_number):
parts = path.split("/")
insert_position = parts.index("Devices") + 1
parts.insert(insert_position, str(id_number))
return "/".join(parts)
def create_csv_files(signals, statuses, node_numbers):
timestamp = int(time.time())
if timestamp % 2 != 0:
timestamp -= 1
# Create CSV directory if it doesn't exist
if not os.path.exists(CSV_DIR):
os.makedirs(CSV_DIR)
#installation_name = get_installation_name(INSTALLATION_NAME_FILE)
csv_filename = f"{timestamp}.csv"
csv_path = os.path.join(CSV_DIR, csv_filename)
# Append values to the CSV file
with open(csv_path, 'a', newline='') as csvfile:
csv_writer = csv.writer(csvfile, delimiter=';')
# Add a special row for the nodes configuration
nodes_config_path = "/Config/Devices/BatteryNodes"
nodes_list = ",".join(str(node) for node in node_numbers)
config_row = [nodes_config_path, nodes_list, ""]
csv_writer.writerow(config_row)
# Iterate over each node and signal to create rows in the new format
for i, node in enumerate(node_numbers):
for s in signals:
signal_name = insert_id(s.name, i+1)
#value = serialize_for_csv(s.get_value(statuses[i]))
value = s.get_value(statuses[i])
row_values = [signal_name, value, s.get_text]
csv_writer.writerow(row_values)
# Manage CSV files, keep a limited number of files
# Create the CSV as a string
csv_data = read_csv_as_string(csv_path)
# Create an S3config instance
s3_config = S3config()
response = s3_config.create_put_request(csv_filename, csv_data)
if response.status_code == 200:
os.remove(csv_path)
print("Success")
else:
failed_dir = os.path.join(CSV_DIR, "failed")
if not os.path.exists(failed_dir):
os.makedirs(failed_dir)
failed_path = os.path.join(failed_dir, csv_filename)
os.rename(csv_path, failed_path)
print("Uploading failed")
manage_csv_files(failed_dir, 10)
manage_csv_files(CSV_DIR)
def main(argv):
# type: (list[str]) -> ()
logging.basicConfig(level=cfg.LOG_LEVEL)
logging.info('starting ' + __file__)
tty = parse_cmdline_args(argv)
modbus = init_modbus(tty)
batteries = identify_batteries(modbus)
n = len(batteries)
logging.info('found ' + str(n) + (' battery' if n == 1 else ' batteries'))
if n <= 0:
sys.exit(2)
bat = c.first(batteries) # report hw and fw version of first battery found
signals = init_signals(bat.hardware_version, bat.firmware_version, n)
csv_signals = create_csv_signals(bat.firmware_version)
main_loop = init_main_loop() # must run before init_dbus because gobject does some global magic
dbus = init_dbus(tty, signals)
update_task = create_update_task(modbus, dbus, batteries, signals, csv_signals, main_loop)
watchdog_task = create_watchdog_task(main_loop)
GLib.timeout_add(cfg.UPDATE_INTERVAL * 2, watchdog_task) # add watchdog first
GLib.timeout_add(cfg.UPDATE_INTERVAL, update_task) # call update once every update_interval
logging.info('starting GLib.MainLoop')
main_loop.run()
logging.info('GLib.MainLoop was shut down')
sys.exit(0xFF) # reaches this only on error
if __name__ == "__main__":
main(sys.argv[1:])

View File

@ -0,0 +1,276 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import sys
from traceback import print_exc
from os import _exit as os_exit
from os import statvfs
from subprocess import check_output, CalledProcessError
import logging
import dbus
logger = logging.getLogger(__name__)
VEDBUS_INVALID = dbus.Array([], signature=dbus.Signature('i'), variant_level=1)
class NoVrmPortalIdError(Exception):
pass
# Use this function to make sure the code quits on an unexpected exception. Make sure to use it
# when using GLib.idle_add and also GLib.timeout_add.
# Without this, the code will just keep running, since GLib does not stop the mainloop on an
# exception.
# Example: GLib.idle_add(exit_on_error, myfunc, arg1, arg2)
def exit_on_error(func, *args, **kwargs):
try:
return func(*args, **kwargs)
except:
try:
print ('exit_on_error: there was an exception. Printing stacktrace will be tried and then exit')
print_exc()
except:
pass
# sys.exit() is not used, since that throws an exception, which does not lead to a program
# halt when used in a dbus callback, see connection.py in the Python/Dbus libraries, line 230.
os_exit(1)
__vrm_portal_id = None
def get_vrm_portal_id():
# The original definition of the VRM Portal ID is that it is the mac
# address of the onboard- ethernet port (eth0), stripped from its colons
# (:) and lower case. This may however differ between platforms. On Venus
# the task is therefore deferred to /sbin/get-unique-id so that a
# platform specific method can be easily defined.
#
# If /sbin/get-unique-id does not exist, then use the ethernet address
# of eth0. This also handles the case where velib_python is used as a
# package install on a Raspberry Pi.
#
# On a Linux host where the network interface may not be eth0, you can set
# the VRM_IFACE environment variable to the correct name.
global __vrm_portal_id
if __vrm_portal_id:
return __vrm_portal_id
portal_id = None
# First try the method that works if we don't have a data partition. This
# will fail when the current user is not root.
try:
portal_id = check_output("/sbin/get-unique-id").decode("utf-8", "ignore").strip()
if not portal_id:
raise NoVrmPortalIdError("get-unique-id returned blank")
__vrm_portal_id = portal_id
return portal_id
except CalledProcessError:
# get-unique-id returned non-zero
raise NoVrmPortalIdError("get-unique-id returned non-zero")
except OSError:
# File doesn't exist, use fallback
pass
# Fall back to getting our id using a syscall. Assume we are on linux.
# Allow the user to override what interface is used using an environment
# variable.
import fcntl, socket, struct, os
iface = os.environ.get('VRM_IFACE', 'eth0').encode('ascii')
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
try:
info = fcntl.ioctl(s.fileno(), 0x8927, struct.pack('256s', iface[:15]))
except IOError:
raise NoVrmPortalIdError("ioctl failed for eth0")
__vrm_portal_id = info[18:24].hex()
return __vrm_portal_id
# See VE.Can registers - public.docx for definition of this conversion
def convert_vreg_version_to_readable(version):
def str_to_arr(x, length):
a = []
for i in range(0, len(x), length):
a.append(x[i:i+length])
return a
x = "%x" % version
x = x.upper()
if len(x) == 5 or len(x) == 3 or len(x) == 1:
x = '0' + x
a = str_to_arr(x, 2);
# remove the first 00 if there are three bytes and it is 00
if len(a) == 3 and a[0] == '00':
a.remove(0);
# if we have two or three bytes now, and the first character is a 0, remove it
if len(a) >= 2 and a[0][0:1] == '0':
a[0] = a[0][1];
result = ''
for item in a:
result += ('.' if result != '' else '') + item
result = 'v' + result
return result
def get_free_space(path):
result = -1
try:
s = statvfs(path)
result = s.f_frsize * s.f_bavail # Number of free bytes that ordinary users
except Exception as ex:
logger.info("Error while retrieving free space for path %s: %s" % (path, ex))
return result
def _get_sysfs_machine_name():
try:
with open('/sys/firmware/devicetree/base/model', 'r') as f:
return f.read().rstrip('\x00')
except IOError:
pass
return None
# Returns None if it cannot find a machine name. Otherwise returns the string
# containing the name
def get_machine_name():
# First try calling the venus utility script
try:
return check_output("/usr/bin/product-name").strip().decode('UTF-8')
except (CalledProcessError, OSError):
pass
# Fall back to sysfs
name = _get_sysfs_machine_name()
if name is not None:
return name
# Fall back to venus build machine name
try:
with open('/etc/venus/machine', 'r', encoding='UTF-8') as f:
return f.read().strip()
except IOError:
pass
return None
def get_product_id():
""" Find the machine ID and return it. """
# First try calling the venus utility script
try:
return check_output("/usr/bin/product-id").strip().decode('UTF-8')
except (CalledProcessError, OSError):
pass
# Fall back machine name mechanism
name = _get_sysfs_machine_name()
return {
'Color Control GX': 'C001',
'Venus GX': 'C002',
'Octo GX': 'C006',
'EasySolar-II': 'C007',
'MultiPlus-II': 'C008',
'Maxi GX': 'C009',
'Cerbo GX': 'C00A'
}.get(name, 'C003') # C003 is Generic
# Returns False if it cannot open the file. Otherwise returns its rstripped contents
def read_file(path):
content = False
try:
with open(path, 'r') as f:
content = f.read().rstrip()
except Exception as ex:
logger.debug("Error while reading %s: %s" % (path, ex))
return content
def wrap_dbus_value(value):
if value is None:
return VEDBUS_INVALID
if isinstance(value, float):
return dbus.Double(value, variant_level=1)
if isinstance(value, bool):
return dbus.Boolean(value, variant_level=1)
if isinstance(value, int):
try:
return dbus.Int32(value, variant_level=1)
except OverflowError:
return dbus.Int64(value, variant_level=1)
if isinstance(value, str):
return dbus.String(value, variant_level=1)
if isinstance(value, list):
if len(value) == 0:
# If the list is empty we cannot infer the type of the contents. So assume unsigned integer.
# A (signed) integer is dangerous, because an empty list of signed integers is used to encode
# an invalid value.
return dbus.Array([], signature=dbus.Signature('u'), variant_level=1)
return dbus.Array([wrap_dbus_value(x) for x in value], variant_level=1)
if isinstance(value, dict):
# Wrapping the keys of the dictionary causes D-Bus errors like:
# 'arguments to dbus_message_iter_open_container() were incorrect,
# assertion "(type == DBUS_TYPE_ARRAY && contained_signature &&
# *contained_signature == DBUS_DICT_ENTRY_BEGIN_CHAR) || (contained_signature == NULL ||
# _dbus_check_is_valid_signature (contained_signature))" failed in file ...'
return dbus.Dictionary({(k, wrap_dbus_value(v)) for k, v in value.items()}, variant_level=1)
return value
dbus_int_types = (dbus.Int32, dbus.UInt32, dbus.Byte, dbus.Int16, dbus.UInt16, dbus.UInt32, dbus.Int64, dbus.UInt64)
def unwrap_dbus_value(val):
"""Converts D-Bus values back to the original type. For example if val is of type DBus.Double,
a float will be returned."""
if isinstance(val, dbus_int_types):
return int(val)
if isinstance(val, dbus.Double):
return float(val)
if isinstance(val, dbus.Array):
v = [unwrap_dbus_value(x) for x in val]
return None if len(v) == 0 else v
if isinstance(val, (dbus.Signature, dbus.String)):
return str(val)
# Python has no byte type, so we convert to an integer.
if isinstance(val, dbus.Byte):
return int(val)
if isinstance(val, dbus.ByteArray):
return "".join([bytes(x) for x in val])
if isinstance(val, (list, tuple)):
return [unwrap_dbus_value(x) for x in val]
if isinstance(val, (dbus.Dictionary, dict)):
# Do not unwrap the keys, see comment in wrap_dbus_value
return dict([(x, unwrap_dbus_value(y)) for x, y in val.items()])
if isinstance(val, dbus.Boolean):
return bool(val)
return val
# When supported, only name owner changes for the the given namespace are reported. This
# prevents spending cpu time at irrelevant changes, like scripts accessing the bus temporarily.
def add_name_owner_changed_receiver(dbus, name_owner_changed, namespace="com.victronenergy"):
# support for arg0namespace is submitted upstream, but not included at the time of
# writing, Venus OS does support it, so try if it works.
if namespace is None:
dbus.add_signal_receiver(name_owner_changed, signal_name='NameOwnerChanged')
else:
try:
dbus.add_signal_receiver(name_owner_changed,
signal_name='NameOwnerChanged', arg0namespace=namespace)
except TypeError:
dbus.add_signal_receiver(name_owner_changed, signal_name='NameOwnerChanged')

View File

@ -0,0 +1,614 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import dbus.service
import logging
import traceback
import os
import weakref
from collections import defaultdict
from ve_utils import wrap_dbus_value, unwrap_dbus_value
# vedbus contains three classes:
# VeDbusItemImport -> use this to read data from the dbus, ie import
# VeDbusItemExport -> use this to export data to the dbus (one value)
# VeDbusService -> use that to create a service and export several values to the dbus
# Code for VeDbusItemImport is copied from busitem.py and thereafter modified.
# All projects that used busitem.py need to migrate to this package. And some
# projects used to define there own equivalent of VeDbusItemExport. Better to
# use VeDbusItemExport, or even better the VeDbusService class that does it all for you.
# TODOS
# 1 check for datatypes, it works now, but not sure if all is compliant with
# com.victronenergy.BusItem interface definition. See also the files in
# tests_and_examples. And see 'if type(v) == dbus.Byte:' on line 102. Perhaps
# something similar should also be done in VeDbusBusItemExport?
# 2 Shouldn't VeDbusBusItemExport inherit dbus.service.Object?
# 7 Make hard rules for services exporting data to the D-Bus, in order to make tracking
# changes possible. Does everybody first invalidate its data before leaving the bus?
# And what about before taking one object away from the bus, instead of taking the
# whole service offline?
# They should! And after taking one value away, do we need to know that someone left
# the bus? Or we just keep that value in invalidated for ever? Result is that we can't
# see the difference anymore between an invalidated value and a value that was first on
# the bus and later not anymore. See comments above VeDbusItemImport as well.
# 9 there are probably more todos in the code below.
# Some thoughts with regards to the data types:
#
# Text from: http://dbus.freedesktop.org/doc/dbus-python/doc/tutorial.html#data-types
# ---
# Variants are represented by setting the variant_level keyword argument in the
# constructor of any D-Bus data type to a value greater than 0 (variant_level 1
# means a variant containing some other data type, variant_level 2 means a variant
# containing a variant containing some other data type, and so on). If a non-variant
# is passed as an argument but introspection indicates that a variant is expected,
# it'll automatically be wrapped in a variant.
# ---
#
# Also the different dbus datatypes, such as dbus.Int32, and dbus.UInt32 are a subclass
# of Python int. dbus.String is a subclass of Python standard class unicode, etcetera
#
# So all together that explains why we don't need to explicitly convert back and forth
# between the dbus datatypes and the standard python datatypes. Note that all datatypes
# in python are objects. Even an int is an object.
# The signature of a variant is 'v'.
# Export ourselves as a D-Bus service.
class VeDbusService(object):
def __init__(self, servicename, bus=None):
# dict containing the VeDbusItemExport objects, with their path as the key.
self._dbusobjects = {}
self._dbusnodes = {}
self._ratelimiters = []
self._dbusname = None
# dict containing the onchange callbacks, for each object. Object path is the key
self._onchangecallbacks = {}
# Connect to session bus whenever present, else use the system bus
self._dbusconn = bus or (dbus.SessionBus() if 'DBUS_SESSION_BUS_ADDRESS' in os.environ else dbus.SystemBus())
# make the dbus connection available to outside, could make this a true property instead, but ach..
self.dbusconn = self._dbusconn
# Register ourselves on the dbus, trigger an error if already in use (do_not_queue)
self._dbusname = dbus.service.BusName(servicename, self._dbusconn, do_not_queue=True)
# Add the root item that will return all items as a tree
self._dbusnodes['/'] = VeDbusRootExport(self._dbusconn, '/', self)
logging.info("registered ourselves on D-Bus as %s" % servicename)
# To force immediate deregistering of this dbus service and all its object paths, explicitly
# call __del__().
def __del__(self):
for node in list(self._dbusnodes.values()):
node.__del__()
self._dbusnodes.clear()
for item in list(self._dbusobjects.values()):
item.__del__()
self._dbusobjects.clear()
if self._dbusname:
self._dbusname.__del__() # Forces call to self._bus.release_name(self._name), see source code
self._dbusname = None
def get_name(self):
return self._dbusname.get_name()
# @param callbackonchange function that will be called when this value is changed. First parameter will
# be the path of the object, second the new value. This callback should return
# True to accept the change, False to reject it.
def add_path(self, path, value, description="", writeable=False,
onchangecallback=None, gettextcallback=None, valuetype=None, itemtype=None):
if onchangecallback is not None:
self._onchangecallbacks[path] = onchangecallback
itemtype = itemtype or VeDbusItemExport
item = itemtype(self._dbusconn, path, value, description, writeable,
self._value_changed, gettextcallback, deletecallback=self._item_deleted, valuetype=valuetype)
spl = path.split('/')
for i in range(2, len(spl)):
subPath = '/'.join(spl[:i])
if subPath not in self._dbusnodes and subPath not in self._dbusobjects:
self._dbusnodes[subPath] = VeDbusTreeExport(self._dbusconn, subPath, self)
self._dbusobjects[path] = item
logging.debug('added %s with start value %s. Writeable is %s' % (path, value, writeable))
# Add the mandatory paths, as per victron dbus api doc
def add_mandatory_paths(self, processname, processversion, connection,
deviceinstance, productid, productname, firmwareversion, hardwareversion, connected):
self.add_path('/Mgmt/ProcessName', processname)
self.add_path('/Mgmt/ProcessVersion', processversion)
self.add_path('/Mgmt/Connection', connection)
# Create rest of the mandatory objects
self.add_path('/DeviceInstance', deviceinstance)
self.add_path('/ProductId', productid)
self.add_path('/ProductName', productname)
self.add_path('/FirmwareVersion', firmwareversion)
self.add_path('/HardwareVersion', hardwareversion)
self.add_path('/Connected', connected)
# Callback function that is called from the VeDbusItemExport objects when a value changes. This function
# maps the change-request to the onchangecallback given to us for this specific path.
def _value_changed(self, path, newvalue):
if path not in self._onchangecallbacks:
return True
return self._onchangecallbacks[path](path, newvalue)
def _item_deleted(self, path):
self._dbusobjects.pop(path)
for np in list(self._dbusnodes.keys()):
if np != '/':
for ip in self._dbusobjects:
if ip.startswith(np + '/'):
break
else:
self._dbusnodes[np].__del__()
self._dbusnodes.pop(np)
def __getitem__(self, path):
return self._dbusobjects[path].local_get_value()
def __setitem__(self, path, newvalue):
self._dbusobjects[path].local_set_value(newvalue)
def __delitem__(self, path):
self._dbusobjects[path].__del__() # Invalidates and then removes the object path
assert path not in self._dbusobjects
def __contains__(self, path):
return path in self._dbusobjects
def __enter__(self):
l = ServiceContext(self)
self._ratelimiters.append(l)
return l
def __exit__(self, *exc):
# pop off the top one and flush it. If with statements are nested
# then each exit flushes its own part.
if self._ratelimiters:
self._ratelimiters.pop().flush()
class ServiceContext(object):
def __init__(self, parent):
self.parent = parent
self.changes = {}
def __getitem__(self, path):
return self.parent[path]
def __setitem__(self, path, newvalue):
c = self.parent._dbusobjects[path]._local_set_value(newvalue)
if c is not None:
self.changes[path] = c
def flush(self):
if self.changes:
self.parent._dbusnodes['/'].ItemsChanged(self.changes)
class TrackerDict(defaultdict):
""" Same as defaultdict, but passes the key to default_factory. """
def __missing__(self, key):
self[key] = x = self.default_factory(key)
return x
class VeDbusRootTracker(object):
""" This tracks the root of a dbus path and listens for PropertiesChanged
signals. When a signal arrives, parse it and unpack the key/value changes
into traditional events, then pass it to the original eventCallback
method. """
def __init__(self, bus, serviceName):
self.importers = defaultdict(weakref.WeakSet)
self.serviceName = serviceName
self._match = bus.get_object(serviceName, '/', introspect=False).connect_to_signal(
"ItemsChanged", weak_functor(self._items_changed_handler))
def __del__(self):
self._match.remove()
self._match = None
def add(self, i):
self.importers[i.path].add(i)
def _items_changed_handler(self, items):
if not isinstance(items, dict):
return
for path, changes in items.items():
try:
v = changes['Value']
except KeyError:
continue
try:
t = changes['Text']
except KeyError:
t = str(unwrap_dbus_value(v))
for i in self.importers.get(path, ()):
i._properties_changed_handler({'Value': v, 'Text': t})
"""
Importing basics:
- If when we power up, the D-Bus service does not exist, or it does exist and the path does not
yet exist, still subscribe to a signal: as soon as it comes online it will send a signal with its
initial value, which VeDbusItemImport will receive and use to update local cache. And, when set,
call the eventCallback.
- If when we power up, save it
- When using get_value, know that there is no difference between services (or object paths) that don't
exist and paths that are invalid (= empty array, see above). Both will return None. In case you do
really want to know ifa path exists or not, use the exists property.
- When a D-Bus service leaves the D-Bus, it will first invalidate all its values, and send signals
with that update, and only then leave the D-Bus. (or do we need to subscribe to the NameOwnerChanged-
signal!?!) To be discussed and make sure. Not really urgent, since all existing code that uses this
class already subscribes to the NameOwnerChanged signal, and subsequently removes instances of this
class.
Read when using this class:
Note that when a service leaves that D-Bus without invalidating all its exported objects first, for
example because it is killed, VeDbusItemImport doesn't have a clue. So when using VeDbusItemImport,
make sure to also subscribe to the NamerOwnerChanged signal on bus-level. Or just use dbusmonitor,
because that takes care of all of that for you.
"""
class VeDbusItemImport(object):
def __new__(cls, bus, serviceName, path, eventCallback=None, createsignal=True):
instance = object.__new__(cls)
# If signal tracking should be done, also add to root tracker
if createsignal:
if "_roots" not in cls.__dict__:
cls._roots = TrackerDict(lambda k: VeDbusRootTracker(bus, k))
return instance
## Constructor
# @param bus the bus-object (SESSION or SYSTEM).
# @param serviceName the dbus-service-name (string), for example 'com.victronenergy.battery.ttyO1'
# @param path the object-path, for example '/Dc/V'
# @param eventCallback function that you want to be called on a value change
# @param createSignal only set this to False if you use this function to one time read a value. When
# leaving it to True, make sure to also subscribe to the NameOwnerChanged signal
# elsewhere. See also note some 15 lines up.
def __init__(self, bus, serviceName, path, eventCallback=None, createsignal=True):
# TODO: is it necessary to store _serviceName and _path? Isn't it
# stored in the bus_getobjectsomewhere?
self._serviceName = serviceName
self._path = path
self._match = None
# TODO: _proxy is being used in settingsdevice.py, make a getter for that
self._proxy = bus.get_object(serviceName, path, introspect=False)
self.eventCallback = eventCallback
assert eventCallback is None or createsignal == True
if createsignal:
self._match = self._proxy.connect_to_signal(
"PropertiesChanged", weak_functor(self._properties_changed_handler))
self._roots[serviceName].add(self)
# store the current value in _cachedvalue. When it doesn't exists set _cachedvalue to
# None, same as when a value is invalid
self._cachedvalue = None
try:
v = self._proxy.GetValue()
except dbus.exceptions.DBusException:
pass
else:
self._cachedvalue = unwrap_dbus_value(v)
def __del__(self):
if self._match is not None:
self._match.remove()
self._match = None
self._proxy = None
def _refreshcachedvalue(self):
self._cachedvalue = unwrap_dbus_value(self._proxy.GetValue())
## Returns the path as a string, for example '/AC/L1/V'
@property
def path(self):
return self._path
## Returns the dbus service name as a string, for example com.victronenergy.vebus.ttyO1
@property
def serviceName(self):
return self._serviceName
## Returns the value of the dbus-item.
# the type will be a dbus variant, for example dbus.Int32(0, variant_level=1)
# this is not a property to keep the name consistant with the com.victronenergy.busitem interface
# returns None when the property is invalid
def get_value(self):
return self._cachedvalue
## Writes a new value to the dbus-item
def set_value(self, newvalue):
r = self._proxy.SetValue(wrap_dbus_value(newvalue))
# instead of just saving the value, go to the dbus and get it. So we have the right type etc.
if r == 0:
self._refreshcachedvalue()
return r
## Resets the item to its default value
def set_default(self):
self._proxy.SetDefault()
self._refreshcachedvalue()
## Returns the text representation of the value.
# For example when the value is an enum/int GetText might return the string
# belonging to that enum value. Another example, for a voltage, GetValue
# would return a float, 12.0Volt, and GetText could return 12 VDC.
#
# Note that this depends on how the dbus-producer has implemented this.
def get_text(self):
return self._proxy.GetText()
## Returns true of object path exists, and false if it doesn't
@property
def exists(self):
# TODO: do some real check instead of this crazy thing.
r = False
try:
r = self._proxy.GetValue()
r = True
except dbus.exceptions.DBusException:
pass
return r
## callback for the trigger-event.
# @param eventCallback the event-callback-function.
@property
def eventCallback(self):
return self._eventCallback
@eventCallback.setter
def eventCallback(self, eventCallback):
self._eventCallback = eventCallback
## Is called when the value of the imported bus-item changes.
# Stores the new value in our local cache, and calls the eventCallback, if set.
def _properties_changed_handler(self, changes):
if "Value" in changes:
changes['Value'] = unwrap_dbus_value(changes['Value'])
self._cachedvalue = changes['Value']
if self._eventCallback:
# The reason behind this try/except is to prevent errors silently ending up the an error
# handler in the dbus code.
try:
self._eventCallback(self._serviceName, self._path, changes)
except:
traceback.print_exc()
os._exit(1) # sys.exit() is not used, since that also throws an exception
class VeDbusTreeExport(dbus.service.Object):
def __init__(self, bus, objectPath, service):
dbus.service.Object.__init__(self, bus, objectPath)
self._service = service
logging.debug("VeDbusTreeExport %s has been created" % objectPath)
def __del__(self):
# self._get_path() will raise an exception when retrieved after the call to .remove_from_connection,
# so we need a copy.
path = self._get_path()
if path is None:
return
self.remove_from_connection()
logging.debug("VeDbusTreeExport %s has been removed" % path)
def _get_path(self):
if len(self._locations) == 0:
return None
return self._locations[0][1]
def _get_value_handler(self, path, get_text=False):
logging.debug("_get_value_handler called for %s" % path)
r = {}
px = path
if not px.endswith('/'):
px += '/'
for p, item in self._service._dbusobjects.items():
if p.startswith(px):
v = item.GetText() if get_text else wrap_dbus_value(item.local_get_value())
r[p[len(px):]] = v
logging.debug(r)
return r
@dbus.service.method('com.victronenergy.BusItem', out_signature='v')
def GetValue(self):
value = self._get_value_handler(self._get_path())
return dbus.Dictionary(value, signature=dbus.Signature('sv'), variant_level=1)
@dbus.service.method('com.victronenergy.BusItem', out_signature='v')
def GetText(self):
return self._get_value_handler(self._get_path(), True)
def local_get_value(self):
return self._get_value_handler(self.path)
class VeDbusRootExport(VeDbusTreeExport):
@dbus.service.signal('com.victronenergy.BusItem', signature='a{sa{sv}}')
def ItemsChanged(self, changes):
pass
@dbus.service.method('com.victronenergy.BusItem', out_signature='a{sa{sv}}')
def GetItems(self):
return {
path: {
'Value': wrap_dbus_value(item.local_get_value()),
'Text': item.GetText() }
for path, item in self._service._dbusobjects.items()
}
class VeDbusItemExport(dbus.service.Object):
## Constructor of VeDbusItemExport
#
# Use this object to export (publish), values on the dbus
# Creates the dbus-object under the given dbus-service-name.
# @param bus The dbus object.
# @param objectPath The dbus-object-path.
# @param value Value to initialize ourselves with, defaults to None which means Invalid
# @param description String containing a description. Can be called over the dbus with GetDescription()
# @param writeable what would this do!? :).
# @param callback Function that will be called when someone else changes the value of this VeBusItem
# over the dbus. First parameter passed to callback will be our path, second the new
# value. This callback should return True to accept the change, False to reject it.
def __init__(self, bus, objectPath, value=None, description=None, writeable=False,
onchangecallback=None, gettextcallback=None, deletecallback=None,
valuetype=None):
dbus.service.Object.__init__(self, bus, objectPath)
self._onchangecallback = onchangecallback
self._gettextcallback = gettextcallback
self._value = value
self._description = description
self._writeable = writeable
self._deletecallback = deletecallback
self._type = valuetype
# To force immediate deregistering of this dbus object, explicitly call __del__().
def __del__(self):
# self._get_path() will raise an exception when retrieved after the
# call to .remove_from_connection, so we need a copy.
path = self._get_path()
if path == None:
return
if self._deletecallback is not None:
self._deletecallback(path)
self.remove_from_connection()
logging.debug("VeDbusItemExport %s has been removed" % path)
def _get_path(self):
if len(self._locations) == 0:
return None
return self._locations[0][1]
## Sets the value. And in case the value is different from what it was, a signal
# will be emitted to the dbus. This function is to be used in the python code that
# is using this class to export values to the dbus.
# set value to None to indicate that it is Invalid
def local_set_value(self, newvalue):
changes = self._local_set_value(newvalue)
if changes is not None:
self.PropertiesChanged(changes)
def _local_set_value(self, newvalue):
if self._value == newvalue:
return None
self._value = newvalue
return {
'Value': wrap_dbus_value(newvalue),
'Text': self.GetText()
}
def local_get_value(self):
return self._value
# ==== ALL FUNCTIONS BELOW THIS LINE WILL BE CALLED BY OTHER PROCESSES OVER THE DBUS ====
## Dbus exported method SetValue
# Function is called over the D-Bus by other process. It will first check (via callback) if new
# value is accepted. And it is, stores it and emits a changed-signal.
# @param value The new value.
# @return completion-code When successful a 0 is return, and when not a -1 is returned.
@dbus.service.method('com.victronenergy.BusItem', in_signature='v', out_signature='i')
def SetValue(self, newvalue):
if not self._writeable:
return 1 # NOT OK
newvalue = unwrap_dbus_value(newvalue)
# If value type is enforced, cast it. If the type can be coerced
# python will do it for us. This allows ints to become floats,
# or bools to become ints. Additionally also allow None, so that
# a path may be invalidated.
if self._type is not None and newvalue is not None:
try:
newvalue = self._type(newvalue)
except (ValueError, TypeError):
return 1 # NOT OK
if newvalue == self._value:
return 0 # OK
# call the callback given to us, and check if new value is OK.
if (self._onchangecallback is None or
(self._onchangecallback is not None and self._onchangecallback(self.__dbus_object_path__, newvalue))):
self.local_set_value(newvalue)
return 0 # OK
return 2 # NOT OK
## Dbus exported method GetDescription
#
# Returns the a description.
# @param language A language code (e.g. ISO 639-1 en-US).
# @param length Lenght of the language string.
# @return description
@dbus.service.method('com.victronenergy.BusItem', in_signature='si', out_signature='s')
def GetDescription(self, language, length):
return self._description if self._description is not None else 'No description given'
## Dbus exported method GetValue
# Returns the value.
# @return the value when valid, and otherwise an empty array
@dbus.service.method('com.victronenergy.BusItem', out_signature='v')
def GetValue(self):
return wrap_dbus_value(self._value)
## Dbus exported method GetText
# Returns the value as string of the dbus-object-path.
# @return text A text-value. '---' when local value is invalid
@dbus.service.method('com.victronenergy.BusItem', out_signature='s')
def GetText(self):
if self._value is None:
return '---'
# Default conversion from dbus.Byte will get you a character (so 'T' instead of '84'), so we
# have to convert to int first. Note that if a dbus.Byte turns up here, it must have come from
# the application itself, as all data from the D-Bus should have been unwrapped by now.
if self._gettextcallback is None and type(self._value) == dbus.Byte:
return str(int(self._value))
if self._gettextcallback is None and self.__dbus_object_path__ == '/ProductId':
return "0x%X" % self._value
if self._gettextcallback is None:
return str(self._value)
return self._gettextcallback(self.__dbus_object_path__, self._value)
## The signal that indicates that the value has changed.
# Other processes connected to this BusItem object will have subscribed to the
# event when they want to track our state.
@dbus.service.signal('com.victronenergy.BusItem', signature='a{sv}')
def PropertiesChanged(self, changes):
pass
## This class behaves like a regular reference to a class method (eg. self.foo), but keeps a weak reference
## to the object which method is to be called.
## Use this object to break circular references.
class weak_functor:
def __init__(self, f):
self._r = weakref.ref(f.__self__)
self._f = weakref.ref(f.__func__)
def __call__(self, *args, **kargs):
r = self._r()
f = self._f()
if r == None or f == None:
return
f(r, *args, **kargs)

View File

@ -0,0 +1,7 @@
#!/bin/bash
. /opt/victronenergy/serial-starter/run-service.sh
app=/opt/victronenergy/dbus-fzsonick-48tl/dbus-fzsonick-48tl.py
args="$tty"
start $args

1287
NodeRed/dvcc.py Normal file

File diff suppressed because it is too large Load Diff

5660
NodeRed/flows.json Normal file

File diff suppressed because one or more lines are too long

26
NodeRed/rc.local Executable file
View File

@ -0,0 +1,26 @@
#!/bin/bash
mount -o remount,rw /
# Source directory
source_dir="/data/dbus-fzsonick-48tl"
# Destination directory
destination_dir_upper="/opt/victronenergy/"
destination_dir="/opt/victronenergy/dbus-fzsonick-48tl/"
# Check if the destination directory exists
if [ -d "$destination_dir" ]; then
# Remove the destination directory
rm -r "$destination_dir"
fi
# Copy the contents of the source directory to the destination directory
cp -r "$source_dir" "$destination_dir_upper"
# Set MPPT network mode to 0
# sed -i "s|('/Link/NetworkMode', [^)]*)|('/Link/NetworkMode', 0)|g" /opt/victronenergy/dbus-systemcalc-py/delegates/dvcc.py
#sed -i "s|self._get_path('/Settings/BmsPresent') == 1|0|g" /opt/victronenergy/dbus-systemcalc-py/delegates/dvcc.py
sed -i "s/self._set_path('\/Link\/NetworkMode', v)/self._set_path('\/Link\/NetworkMode', 0)\n self._set_path('\/Settings\/BmsPresent',0)/" /opt/victronenergy/dbus-systemcalc-py/delegates/dvcc.py
exit 0

31
NodeRed/settings-user.js Normal file
View File

@ -0,0 +1,31 @@
module.exports = {
uiHost:"",
/* To password protect the Node-RED editor and admin API, the following
property can be used. See https://nodered.org/docs/security.html for details.
*/
adminAuth: {
sessionExpiryTime: 86400,
type: "credentials",
users: [{
username: "admin",
password: "$2b$08$d7A0gwkDh4KtultiCAVH6eQ.tQUwVApq.tDVOOYQ51EpLIMbYy2GW",//salidomo
permissions: "*"
}]
},
/* Context Storage
The following property can be used to enable context storage. The configuration
provided here will enable file-based context that flushes to disk every 30 seconds.
Refer to the documentation for further options: https://nodered.org/docs/api/context/
*/
//contextStorage: {
// default: {
// module:"localfilesystem"
// },
//},
contextStorage: {
default: "memoryOnly",
memoryOnly: { module: 'memory' },
file: { module: 'localfilesystem' }
},
}

View File

@ -677,7 +677,7 @@ internal static class Program
var request = s3Config.CreatePutRequest(s3Path); var request = s3Config.CreatePutRequest(s3Path);
// This is temporary for Wittman, but now it's for all Installation // This is temporary for Wittman, but now it's for all Installation
await File.WriteAllTextAsync("/var/www/html/status.csv", csv.SplitLines().Where(l => !l.Contains("Secret")).JoinLines()); //await File.WriteAllTextAsync("/var/www/html/status.csv", csv.SplitLines().Where(l => !l.Contains("Secret")).JoinLines());
//Use this for no compression //Use this for no compression
//var response = await request.PutAsync(new StringContent(csv)); //var response = await request.PutAsync(new StringContent(csv));

View File

@ -1,303 +0,0 @@
#!/usr/bin/python2 -u
# coding=utf-8
import os
import struct
from time import sleep
import serial
from os import system
from pymodbus.client.sync import ModbusSerialClient as Modbus
from pymodbus.exceptions import ModbusIOException
from pymodbus.pdu import ModbusResponse
from os.path import dirname, abspath
from sys import path, argv, exit
path.append(dirname(dirname(abspath(__file__))))
PAGE_SIZE = 0x100
HALF_PAGE = PAGE_SIZE / 2
WRITE_ENABLE = [1]
SERIAL_STARTER_DIR = '/opt/victronenergy/serial-starter/'
FIRMWARE_VERSION_REGISTER = 1054
ERASE_FLASH_REGISTER = 0x2084
RESET_REGISTER = 0x2087
# trick the pycharm type-checker into thinking Callable is in scope, not used at runtime
# noinspection PyUnreachableCode
if False:
from typing import List, NoReturn, Iterable, Optional
class LockTTY(object):
def __init__(self, tty):
# type: (str) -> None
self.tty = tty
def __enter__(self):
system(SERIAL_STARTER_DIR + 'stop-tty.sh ' + self.tty)
return self
def __exit__(self, exc_type, exc_val, exc_tb):
system(SERIAL_STARTER_DIR + 'start-tty.sh ' + self.tty)
def calc_stm32_crc_round(crc, data):
# type: (int, int) -> int
crc = crc ^ data
for _ in range(32):
xor = (crc & 0x80000000) != 0
crc = (crc & 0x7FFFFFFF) << 1 # clear bit 31 because python ints have "infinite" bits
if xor:
crc = crc ^ 0x04C11DB7
return crc
def calc_stm32_crc(data):
# type: (Iterable[int]) -> int
crc = 0xFFFFFFFF
for dw in data:
crc = calc_stm32_crc_round(crc, dw)
return crc
def init_modbus(tty):
# type: (str) -> Modbus
return Modbus(
port='/dev/' + tty,
method='rtu',
baudrate=115200,
stopbits=1,
bytesize=8,
timeout=0.15, # seconds
parity=serial.PARITY_ODD)
def failed(response):
# type: (ModbusResponse) -> bool
# Todo 'ModbusIOException' object has no attribute 'function_code'
return response.function_code > 0x80
def clear_flash(modbus, slave_address):
# type: (Modbus, int) -> bool
print ('erasing flash...')
write_response = modbus.write_registers(address=0x2084, values=[1], unit=slave_address)
if failed(write_response):
print('erasing flash FAILED')
return False
flash_countdown = 17
while flash_countdown > 0:
read_response = modbus.read_holding_registers(address=0x2085, count=1, unit=slave_address)
if failed(read_response):
print('erasing flash FAILED')
return False
if read_response.registers[0] != flash_countdown:
flash_countdown = read_response.registers[0]
msg = str(100 * (16 - flash_countdown) / 16) + '%'
print('\r{0} '.format(msg), end=' ')
print('done!')
return True
# noinspection PyShadowingBuiltins
def bytes_to_words(bytes):
# type: (str) -> List[int]
return list(struct.unpack('>' + len(bytes)/2 * 'H', bytes))
def send_half_page_1(modbus, slave_address, data, page):
# type: (Modbus, int, str, int) -> NoReturn
first_half = [page] + bytes_to_words(data[:HALF_PAGE])
write_first_half = modbus.write_registers(0x2000, first_half, unit=slave_address)
if failed(write_first_half):
raise Exception("Failed to write page " + str(page))
def send_half_page_2(modbus, slave_address, data, page):
# type: (Modbus, int, str, int) -> NoReturn
registers = bytes_to_words(data[HALF_PAGE:]) + calc_crc(page, data) + WRITE_ENABLE
result = modbus.write_registers(0x2041, registers, unit=slave_address)
if failed(result):
raise Exception("Failed to write page " + str(page))
def get_fw_name(fw_path):
# type: (str) -> str
return fw_path.split('/')[-1].split('.')[0]
def upload_fw(modbus, slave_id, fw_path, fw_name):
# type: (Modbus, int, str, str) -> NoReturn
with open(fw_path, "rb") as f:
size = os.fstat(f.fileno()).st_size
n_pages = size / PAGE_SIZE
print('uploading firmware ' + fw_name + ' to BMS ...')
for page in range(0, n_pages):
page_data = f.read(PAGE_SIZE)
msg = "page " + str(page + 1) + '/' + str(n_pages) + ' ' + str(100 * page / n_pages + 1) + '%'
print('\r{0} '.format(msg), end=' ')
if is_page_empty(page_data):
continue
send_half_page_1(modbus, slave_id, page_data, page)
send_half_page_2(modbus, slave_id, page_data, page)
def is_page_empty(page):
# type: (str) -> bool
return page.count('\xff') == len(page)
def reset_bms(modbus, slave_id):
# type: (Modbus, int) -> bool
print ('resetting BMS...')
result = modbus.write_registers(RESET_REGISTER, [1], unit=slave_id)
# expecting a ModbusIOException (timeout)
# BMS can no longer reply because it is already reset
success = isinstance(result, ModbusIOException)
if success:
print('done')
else:
print('FAILED to reset battery!')
return success
def calc_crc(page, data):
# type: (int, str) -> List[int]
crc = calc_stm32_crc([page] + bytes_to_words(data))
crc_bytes = struct.pack('>L', crc)
return bytes_to_words(crc_bytes)
def identify_battery(modbus, slave_id):
# type: (Modbus, int) -> Optional[str]
target = 'battery #' + str(slave_id) + ' at ' + modbus.port
try:
print(('contacting ' + target + ' ...'))
response = modbus.read_input_registers(address=FIRMWARE_VERSION_REGISTER, count=1, unit=slave_id)
fw = '{0:0>4X}'.format(response.registers[0])
print(('found battery with firmware ' + fw))
return fw
except:
print(('failed to communicate with ' + target + ' !'))
return None
def print_usage():
print(('Usage: ' + __file__ + ' <serial device> <battery id> <firmware>'))
print(('Example: ' + __file__ + ' ttyUSB0 2 A08C.bin'))
def parse_cmdline_args(argv):
# type: (List[str]) -> (str, str, str, str)
def fail_with(msg):
print(msg)
print_usage()
exit(1)
if len(argv) < 1:
fail_with('missing argument for tty device')
if len(argv) < 2:
fail_with('missing argument for battery ID')
if len(argv) < 3:
fail_with('missing argument for firmware')
return argv[0], int(argv[1]), argv[2], get_fw_name(argv[2])
def verify_firmware(modbus, battery_id, fw_name):
# type: (Modbus, int, str) -> NoReturn
fw_verify = identify_battery(modbus, battery_id)
if fw_verify == fw_name:
print('SUCCESS')
else:
print('FAILED to verify uploaded firmware!')
if fw_verify is not None:
print('expected firmware version ' + fw_name + ' but got ' + fw_verify)
def wait_for_bms_reboot():
# type: () -> NoReturn
# wait 20s for the battery to reboot
print('waiting for BMS to reboot...')
for t in range(20, 0, -1):
print('\r{0} '.format(t), end=' ')
sleep(1)
print('0')
def main(argv):
# type: (List[str]) -> NoReturn
tty, battery_id, fw_path, fw_name = parse_cmdline_args(argv)
with LockTTY(tty), init_modbus(tty) as modbus:
if identify_battery(modbus, battery_id) is None:
return
clear_flash(modbus, battery_id)
upload_fw(modbus, battery_id, fw_path, fw_name)
if not reset_bms(modbus, battery_id):
return
wait_for_bms_reboot()
verify_firmware(modbus, battery_id, fw_name)
main(argv[1:])

File diff suppressed because it is too large Load Diff

View File

@ -20,6 +20,7 @@
"date-fns": "^2.28.0", "date-fns": "^2.28.0",
"dayjs": "^1.11.10", "dayjs": "^1.11.10",
"history": "5.3.0", "history": "5.3.0",
"jszip": "^3.10.1",
"linq-to-typescript": "^11.0.0", "linq-to-typescript": "^11.0.0",
"nprogress": "0.2.0", "nprogress": "0.2.0",
"numeral": "2.0.6", "numeral": "2.0.6",

View File

@ -107,6 +107,55 @@ function Log(props: LogProps) {
}); });
}; };
const warningDescriptionMap: { [key: string]: string } = {
"TaM1": "TaM1: BMS temperature high",
"TbM1": "TbM1: Battery temperature high",
"VBm1": "VBm1: Bus voltage low",
"VBM1": "VBM1: Bus voltage high",
"IDM1": "IDM1: Discharge current high",
"vsm1": "vsm1: String voltage low",
"vsM1": "vsM1: String voltage high",
"iCM1": "iCM1: Charge current high",
"iDM1": "iDM1: Discharge current high",
"MID1": "MID1: String voltages unbalanced",
"BLPW": "BLPW: Not enough charging power on bus",
"CCBF": "CCBF: Internal charger hardware failure",
"Ah_W": "Ah_W: String SOC low",
"MPMM": "MPMM: Midpoint wiring problem",
"TCdi": "TCdi: Temperature difference between strings high",
"LMPW": "LMPW: String voltages unbalance warning",
"TOCW": "TOCW: Top of Charge requested"
};
const errorDescriptionMap: { [key: string]: string } = {
"Tam": "Tam: Recoverable, BMS temperature too low",
"TaM2": "TaM2: Recoverable, BMS temperature too high",
"Tbm": "Tbm: Recoverable, Battery temperature too low",
"TbM2": "TbM2: Recoverable, Battery temperature too high",
"VBm2": "VBm2: Recoverable, Recoverable: Bus voltage too low",
"VBM2": "VBM2: Recoverable,Recoverable: Bus voltage too high",
"IDM2": "IDM2: Recoverable, Discharge current too high",
"ISOB": "ISOB: Unrecoverable, Electrical insulation failure",
"MSWE": "MSWE: Unrecoverable, Main switch failure",
"FUSE": "FUSE: Unrecoverable, Main fuse blown",
"HTRE": "HTRE: Recoverable, Battery failed to warm up",
"TCPE": "TCPE: Unrecoverable, Temperature sensor failure",
"STRE": "STRE: Recoverable, Voltage measurement circuit fails",
"CME": "CME: Recoverable, Current sensor failure",
"HWFL": "HWFL: Recoverable, BMS hardware failure",
"HWEM": "HWEM: Recoverable, Hardware protection tripped",
"ThM": "ThM: Recoverable, Heatsink temperature too high",
"vsm2": "vsm2: Unrecoverable, Low string voltage failure",
"vsM2": "vsM2: Recoverable, String voltage too high",
"iCM2": "iCM2: Unrecoverable, Charge current too high",
"iDM2": "iDM2: Recoverable, Discharge current too high",
"MID2": "MID2: Recoverable, String voltage unbalance too high",
"HTFS": "HTFS: Recoverable, Unrecoverable: Heater Fuse Blown",
"DATA": "DATA: Recoverable, Unrecoverable: Parameters out of range",
"LMPA": "LMPA: Unrecoverable, String voltages unbalance alarm",
"HEBT": "HEBT: Recoverable, oss of heartbeat"
};
return ( return (
<Container maxWidth="xl"> <Container maxWidth="xl">
<Grid container> <Grid container>
@ -304,7 +353,7 @@ function Log(props: LogProps) {
gutterBottom gutterBottom
noWrap noWrap
> >
{error.description} {errorDescriptionMap[error.description] || error.description}
</Typography> </Typography>
</div> </div>
<div <div
@ -619,7 +668,7 @@ function Log(props: LogProps) {
gutterBottom gutterBottom
noWrap noWrap
> >
{warning.description} {warningDescriptionMap[warning.description] || warning.description}
</Typography> </Typography>
</div> </div>
<div <div