NodeRed/dbus-csv-files/config.py

@@ -1,51 +0,0 @@
-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
-
-

NodeRed/dbus-csv-files/convert.py

@@ -1,119 +0,0 @@
-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
-
-

NodeRed/dbus-csv-files/data.py

@@ -1,63 +0,0 @@
-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

NodeRed/dbus-csv-files/dbus-csv-files.py

@@ -1,731 +0,0 @@
-#! /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:])

NodeRed/dbus-csv-files/start.sh

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

NodeRed/dbus-fzsonick-48tl/config.py

@@ -1,51 +0,0 @@
-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
-
-

NodeRed/dbus-fzsonick-48tl/convert.py

@@ -1,119 +0,0 @@
-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
-
-

NodeRed/dbus-fzsonick-48tl/data.py

@@ -1,97 +0,0 @@
-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

NodeRed/dbus-fzsonick-48tl/dbus-fzsonick-48tl.py

@@ -1,980 +0,0 @@
-#!/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:])

NodeRed/dbus-fzsonick-48tl/ext/velib_python/ve_utils.py

@@ -1,276 +0,0 @@
-#!/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')

NodeRed/dbus-fzsonick-48tl/ext/velib_python/vedbus.py

@@ -1,614 +0,0 @@
-#!/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)

NodeRed/dbus-fzsonick-48tl/start.sh

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

NodeRed/rc.local

@@ -1,26 +0,0 @@
-#!/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

NodeRed/settings-user.js

@@ -1,31 +0,0 @@
-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' }
-    },
-    }

firmware/opt/innovenergy/scripts/upload-bms-firmware

@@ -0,0 +1,303 @@
+#!/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:])

typescript/frontend-marios2/package.json

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

typescript/frontend-marios2/src/content/dashboards/Log/Log.tsx

@@ -107,55 +107,6 @@ 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 (
     <Container maxWidth="xl">
       <Grid container>
@@ -353,7 +304,7 @@ function Log(props: LogProps) {
                             gutterBottom
                             noWrap
                           >
-                            {errorDescriptionMap[error.description] || error.description}
+                            {error.description}
                           </Typography>
                         </div>
                         <div
@@ -668,7 +619,7 @@ function Log(props: LogProps) {
                             gutterBottom
                             noWrap
                           >
-                            {warningDescriptionMap[warning.description] || warning.description}
+                            {warning.description}
                           </Typography>
                         </div>
                         <div