extarct s3 data with decompression

This commit is contained in:
kostas 2024-06-03 12:53:39 +02:00
parent bac3e17224
commit c73ef581e3
14 changed files with 2941 additions and 0 deletions

View File

@ -0,0 +1,59 @@
import serial
import logging
from data import read_file_one_line
# dbus configuration
CONNECTION = 'Modbus RTU'
PRODUCT_NAME = 'FIAMM 48TL Series Battery'
PRODUCT_ID = 0xB012 # assigned by victron
DEVICE_INSTANCE = 1
SERVICE_NAME_PREFIX = 'com.victronenergy.battery.'
# driver configuration
SOFTWARE_VERSION = '3.0.0'
UPDATE_INTERVAL = 2000 # milliseconds
#LOG_LEVEL = logging.INFO
LOG_LEVEL = logging.DEBUG
# battery config
V_MAX = 54.2
V_MIN = 42
R_STRING_MIN = 0.125
R_STRING_MAX = 0.250
I_MAX_PER_STRING = 15
AH_PER_STRING = 40
NUM_OF_STRINGS_PER_BATTERY = 5
# modbus configuration
BASE_ADDRESS = 999
NO_OF_REGISTERS = 64
MAX_SLAVE_ADDRESS = 25
# RS 485 configuration
PARITY = serial.PARITY_ODD
TIMEOUT = 0.1 # seconds
BAUD_RATE = 115200
BYTE_SIZE = 8
STOP_BITS = 1
MODE = 'rtu'
# InnovEnergy IOT configuration
INSTALLATION_NAME = read_file_one_line('/data/innovenergy/openvpn/installation-name')
INNOVENERGY_SERVER_IP = '10.2.0.1'
INNOVENERGY_SERVER_PORT = 8134
INNOVENERGY_PROTOCOL_VERSION = '48TL200V3'
# S3 Credentials
S3BUCKET = "5-c0436b6a-d276-4cd8-9c44-1eae86cf5d0e"
S3KEY = "EXO6bb63d9bbe5f938a68fa444b"
S3SECRET = "A4-5wIjIlAqn-p0cUkQu0f9fBIrX1V5PGTBDwjsrlR8"

View File

@ -0,0 +1,644 @@
#!/usr/bin/python -u
# coding=utf-8
import logging
import os
import time
import states as State
import target_type as TargetType
from random import randint
from python_libs.ie_dbus.dbus_service import DBusService
from python_libs.ie_utils.main_loop import run_on_main_loop
# noinspection PyUnreachableCode
if False:
from typing import NoReturn, Optional, Any, Iterable, List
logging.basicConfig(level=logging.INFO)
_log = logging.getLogger(__name__)
VERSION = '1.0.0'
PRODUCT = 'Controller'
GRID_SERVICE_PREFIX = 'com.victronenergy.grid.'
BATTERY_SERVICE_PREFIX = 'com.victronenergy.battery.'
INVERTER_SERVICE_PREFIX = 'com.victronenergy.vebus.'
SYSTEM_SERVICE_PREFIX = 'com.victronenergy.system'
HUB4_SERVICE_PREFIX = 'com.victronenergy.hub4'
SETTINGS_SERVICE_PREFIX = 'com.victronenergy.settings'
UPDATE_PERIOD_MS = 2000
MAX_POWER_PER_BATTERY = 2500
MAX_DAYS_WITHOUT_EOC = 7
SECONDS_PER_DAY = 24 * 60 * 60
GRID_SET_POINT_SETTING = PRODUCT + '/GridSetPoint'
LAST_EOC_SETTING = PRODUCT + '/LastEOC'
CALIBRATION_CHARGE_START_TIME_OF_DAY_SETTING = PRODUCT + '/CalibrationChargeStartTime'
HEAT_LOSS = 150 # W
P_CONST = 0.5 # W/W
Epoch = int
Seconds = int
def time_now():
return int(time.time())
class Controller(object):
def __init__(self, measurement, target, target_type, state):
# type: (float, float, int, int) -> NoReturn
self.target_type = target_type
self.target = target
self.measurement = measurement
self.state = state
d_p = target - measurement
self.delta = d_p * P_CONST
@staticmethod
def min(controllers):
# type: (Iterable[Controller]) -> Controller
return min(controllers, key=lambda c: c.delta)
@staticmethod
def max(controllers):
# type: (Iterable[Controller]) -> Controller
return max(controllers, key=lambda c: c.delta)
def clamp(self, lower_limit_controllers, upper_limit_controllers):
# type: (List[Controller],List[Controller]) -> Controller
c_min = Controller.min(upper_limit_controllers + [self])
return Controller.max(lower_limit_controllers + [c_min])
# noinspection PyMethodMayBeStatic
class InnovEnergyController(DBusService):
def __init__(self):
super(InnovEnergyController, self).__init__(PRODUCT.lower())
self.settings.add_setting(path=LAST_EOC_SETTING, default_value=0) # unix epoch timestamp
self.settings.add_setting(path=GRID_SET_POINT_SETTING, default_value=0) # grid setpoint, Watts
self.settings.add_setting(path=CALIBRATION_CHARGE_START_TIME_OF_DAY_SETTING, default_value=32400) # 09:00
self.own_properties.set('/ProductName', PRODUCT)
self.own_properties.set('/Mgmt/ProcessName', __file__)
self.own_properties.set('/Mgmt/ProcessVersion', VERSION)
self.own_properties.set('/Mgmt/Connection', 'dbus')
self.own_properties.set('/ProductId', PRODUCT)
self.own_properties.set('/FirmwareVersion', VERSION)
self.own_properties.set('/HardwareVersion', VERSION)
self.own_properties.set('/Connected', 1)
self.own_properties.set('/TimeToCalibrationCharge', 'unknown')
self.own_properties.set('/State', 0)
self.phases = [
p for p in ['/Hub4/L1/AcPowerSetpoint', '/Hub4/L2/AcPowerSetpoint', '/Hub4/L3/AcPowerSetpoint']
if self.remote_properties.exists(self.inverter_service + p)
]
self.n_phases = len(self.phases)
print ('The system has ' + str(self.n_phases) + ' phase' + ('s' if self.n_phases != 1 else ''))
self.max_inverter_power = 32700
# ^ defined in https://github.com/victronenergy/dbus_modbustcp/blob/master/CCGX-Modbus-TCP-register-list.xlsx
def clamp_power_command(self, value):
# type: (float) -> int
value = max(value, -self.max_inverter_power)
value = min(value, self.max_inverter_power)
return int(value)
def get_service(self, prefix):
# type: (str) -> Optional[unicode]
service = next((s for s in self.available_services if s.startswith(prefix)), None)
if service is None:
raise Exception('no service matching ' + prefix + '* available')
return service
def is_service_available(self, prefix):
# type: (str) -> bool
return next((True for s in self.available_services if s.startswith(prefix)), False)
@property
def battery_service(self):
# type: () -> Optional[unicode]
return self.get_service(BATTERY_SERVICE_PREFIX)
@property
def battery_available(self):
# type: () -> bool
return self.is_service_available(BATTERY_SERVICE_PREFIX)
@property
def grid_service(self):
# type: () -> Optional[unicode]
return self.get_service(GRID_SERVICE_PREFIX)
@property
def grid_meter_available(self):
# type: () -> bool
return self.is_service_available(GRID_SERVICE_PREFIX)
@property
def inverter_service(self):
# type: () -> Optional[unicode]
return self.get_service(INVERTER_SERVICE_PREFIX)
@property
def inverter_available(self):
# type: () -> bool
return self.is_service_available(INVERTER_SERVICE_PREFIX)
@property
def system_service(self):
# type: () -> Optional[unicode]
return self.get_service(SYSTEM_SERVICE_PREFIX)
@property
def system_service_available(self):
# type: () -> bool
return self.is_service_available(SYSTEM_SERVICE_PREFIX)
@property
def hub4_service(self):
# type: () -> Optional[unicode]
return self.get_service(HUB4_SERVICE_PREFIX)
@property
def hub4_service_available(self):
# type: () -> bool
return self.is_service_available(HUB4_SERVICE_PREFIX)
@property
def inverter_power_setpoint(self):
# type: () -> float
return sum((self.get_inverter_prop(p) for p in self.phases))
def get_battery_prop(self, dbus_path):
# type: (str) -> Any
battery_service = self.battery_service
return self.remote_properties.get(battery_service + dbus_path).value
def get_grid_prop(self, dbus_path):
# type: (str) -> Any
return self.remote_properties.get(self.grid_service + dbus_path).value
def get_inverter_prop(self, dbus_path):
# type: (str) -> Any
return self.remote_properties.get(self.inverter_service + dbus_path).value
def get_system_prop(self, dbus_path):
# type: (str) -> Any
system_service = self.system_service
return self.remote_properties.get(system_service + dbus_path).value
def get_hub4_prop(self, dbus_path):
# type: (str) -> Any
hub4_service = self.hub4_service
return self.remote_properties.get(hub4_service + dbus_path).value
def set_settings_prop(self, dbus_path, value):
# type: (str, Any) -> bool
return self.remote_properties.set(SETTINGS_SERVICE_PREFIX + dbus_path, value)
def set_inverter_prop(self, dbus_path, value):
# type: (str, Any) -> bool
inverter_service = self.inverter_service
return self.remote_properties.set(inverter_service + dbus_path, value)
@property
def max_battery_charge_power(self):
# type: () -> int
return self.get_battery_prop('/Info/MaxChargePower')
@property
def max_battery_discharge_power(self):
# type: () -> int
return self.get_battery_prop('/Info/MaxDischargePower')
@property
def max_configured_charge_power(self):
# type: () -> Optional[int]
max_power = self.settings.get('/Settings/CGwacs/MaxChargePower')
return max_power if max_power >= 0 else None
@property
def max_configured_discharge_power(self): # unsigned
# type: () -> Optional[int]
max_power = self.settings.get('/Settings/CGwacs/MaxDischargePower')
return max_power if max_power >= 0 else None
@property
def max_charge_power(self):
# type: () -> int
if self.max_configured_charge_power is None:
return self.max_battery_charge_power
else:
return min(self.max_battery_charge_power, self.max_configured_charge_power)
@property
def max_discharge_power(self): # unsigned
# type: () -> int
if self.max_configured_discharge_power is None:
return self.max_battery_discharge_power
else:
return min(self.max_battery_discharge_power, self.max_configured_discharge_power)
def set_inverter_power_setpoint(self, power):
# type: (float) -> NoReturn
if self.settings.get('/Settings/CGwacs/BatteryLife/State') == 9:
self.settings.set('/Settings/CGwacs/BatteryLife/State', 0) # enables scheduled charge
self.settings.set('/Settings/CGwacs/Hub4Mode', 3) # disable hub4
self.set_inverter_prop('/Hub4/DisableCharge', 0)
self.set_inverter_prop('/Hub4/DisableFeedIn', 0)
power = self.clamp_power_command(power / self.n_phases)
for p in self.phases:
self.set_inverter_prop(p, power + randint(-1, 1)) # use randint to force dbus re-send
def set_controller_state(self, state):
# type: (int) -> NoReturn
self.own_properties.set('/State', state)
@property
def grid_power(self):
# type: () -> Optional[float]
try:
return self.get_grid_prop('/Ac/Power')
except:
return None
@property
def battery_cold(self):
# type: () -> bool
return self.get_battery_prop('/IoStatus/BatteryCold') == 1
@property
def eoc_reached(self):
# type: () -> bool
if not self.battery_available:
return False
return min(self.get_battery_prop('/EOCReached')) == 1
@property
def battery_power(self):
# type: () -> float
return self.get_battery_prop('/Dc/0/Power')
@property
def inverter_ac_in_power(self):
# type: () -> float
return self.get_inverter_prop('/Ac/ActiveIn/P')
@property
def inverter_ac_out_power(self):
# type: () -> float
return self.get_inverter_prop('/Ac/Out/P')
@property
def soc(self):
# type: () -> float
return self.get_battery_prop('/Soc')
@property
def n_batteries(self):
# type: () -> int
return self.get_battery_prop('/NbOfBatteries')
@property
def min_soc(self):
# type: () -> float
return self.settings.get('/Settings/CGwacs/BatteryLife/MinimumSocLimit')
@property
def should_hold_min_soc(self):
# type: () -> bool
return self.min_soc <= self.soc <= self.min_soc + 5
@property
def utc_offset(self):
# type: () -> int
# stackoverflow.com/a/1301528
# stackoverflow.com/a/3168394
os.environ['TZ'] = self.settings.get('/Settings/System/TimeZone')
time.tzset()
is_dst = time.daylight and time.localtime().tm_isdst > 0
return -(time.altzone if is_dst else time.timezone)
@property
def grid_set_point(self):
# type: () -> float
return self.settings.get('/Settings/CGwacs/AcPowerSetPoint')
@property
def time_to_calibration_charge_str(self):
# type: () -> str
return self.own_properties.get('/TimeToCalibrationCharge').text
@property
def calibration_charge_deadline(self):
# type: () -> Epoch
utc_offset = self.utc_offset
ultimate_deadline = self.settings.get(LAST_EOC_SETTING) + MAX_DAYS_WITHOUT_EOC * SECONDS_PER_DAY
midnight_before_udl = int((ultimate_deadline + utc_offset) / SECONDS_PER_DAY) * SECONDS_PER_DAY - utc_offset # round off to last midnight
dead_line = midnight_before_udl + self.calibration_charge_start_time_of_day
while dead_line > ultimate_deadline: # should fire at most once, but let's be defensive...
dead_line -= SECONDS_PER_DAY # too late, advance one day
return dead_line
@property
def time_to_calibration_charge(self):
# type: () -> Seconds
return self.calibration_charge_deadline - time_now()
@property
def grid_blackout(self):
# type: () -> bool
return self.get_inverter_prop('/Leds/Mains') < 1
@property
def scheduled_charge(self):
# type: () -> bool
return self.get_hub4_prop('/Overrides/ForceCharge') != 0
@property
def calibration_charge_start_time_of_day(self):
# type: () -> Seconds
return self.settings.get(CALIBRATION_CHARGE_START_TIME_OF_DAY_SETTING) # seconds since midnight
@property
def must_do_calibration_charge(self):
# type: () -> bool
return self.time_to_calibration_charge <= 0
def controller_charge_to_min_soc(self):
# type: () -> Controller
return Controller(
measurement=self.battery_power,
target=self.max_charge_power,
target_type=TargetType.BATTERY_DC,
state=State.CHARGE_TO_MIN_SOC
)
def controller_hold_min_soc(self):
# type: () -> Controller
# TODO: explain
a = -4 * HEAT_LOSS * self.n_batteries
b = -a * (self.min_soc + .5)
target_dc_power = a * self.soc + b
return Controller(
measurement = self.battery_power,
target = target_dc_power,
target_type = TargetType.BATTERY_DC,
state = State.HOLD_MIN_SOC
)
def controller_calibration_charge(self):
# type: () -> Controller
return Controller(
measurement = self.battery_power,
target = self.max_charge_power,
target_type = TargetType.BATTERY_DC,
state = State.CALIBRATION_CHARGE
)
def controller_limit_discharge_power(self): # signed
# type: () -> Controller
return Controller(
measurement = self.battery_power,
target = -self.max_discharge_power, # add sign!
target_type = TargetType.BATTERY_DC,
state = State.LIMIT_DISCHARGE_POWER
)
def controller_limit_charge_power(self):
# type: () -> Controller
return Controller(
measurement = self.battery_power,
target = self.max_charge_power,
target_type = TargetType.BATTERY_DC,
state = State.LIMIT_CHARGE_POWER
)
def controller_optimize_self_consumption(self):
# type: () -> Controller
return Controller(
measurement = self.grid_power,
target = self.grid_set_point,
target_type = TargetType.GRID_AC,
state = State.OPTIMIZE_SELF_CONSUMPTION
)
def controller_heating(self):
# type: () -> Controller
return Controller(
measurement = self.battery_power,
target = self.max_charge_power,
target_type = TargetType.BATTERY_DC,
state = State.HEATING
)
def controller_scheduled_charge(self):
# type: () -> Controller
return Controller(
measurement = self.battery_power,
target = self.max_charge_power,
target_type = TargetType.BATTERY_DC,
state = State.SCHEDULED_CHARGE
)
def controller_no_grid_meter(self):
# type: () -> Controller
return Controller(
measurement = self.battery_power,
target = self.max_charge_power,
target_type = TargetType.BATTERY_DC,
state = State.NO_GRID_METER_AVAILABLE
)
def controller_no_battery(self):
# type: () -> Controller
return Controller(
measurement = self.inverter_ac_in_power,
target = 0,
target_type = TargetType.INVERTER_AC_IN,
state = State.NO_BATTERY_AVAILABLE
)
def controller_bridge_grid_blackout(self):
# type: () -> Controller
return Controller(
measurement = 0,
target = 0,
target_type = TargetType.GRID_AC,
state = State.BRIDGE_GRID_BLACKOUT
)
def update_eoc(self):
if self.eoc_reached:
print('battery has reached EOC')
self.settings.set(LAST_EOC_SETTING, time_now())
self.publish_time_to_calibration_charge()
def publish_time_to_calibration_charge(self):
total_seconds = self.time_to_calibration_charge
if total_seconds <= 0:
time_to_eoc_str = 'now'
else:
total_minutes, seconds = divmod(total_seconds, 60)
total_hours, minutes = divmod(total_minutes, 60)
total_days, hours = divmod(total_hours, 24)
days_str = (str(total_days) + 'd') if total_days > 0 else ''
hours_str = (str(hours) + 'h') if total_hours > 0 else ''
minutes_str = (str(minutes) + 'm') if total_days == 0 else ''
time_to_eoc_str = "{0} {1} {2}".format(days_str, hours_str, minutes_str).strip()
self.own_properties.set('/TimeToCalibrationCharge', time_to_eoc_str)
def print_system_stats(self, controller):
# type: (Controller) -> NoReturn
def soc_setpoint():
if controller.state == State.CALIBRATION_CHARGE or controller.state == State.NO_GRID_METER_AVAILABLE:
return ' => 100%'
if controller.state == State.CHARGE_TO_MIN_SOC:
return ' => ' + str(int(self.min_soc)) + '%'
return ''
def setpoint(target_type):
if target_type != controller.target_type:
return ''
return ' => ' + str(int(controller.target)) + 'W'
def p(power):
# type: (Optional[float]) -> str
if power is None:
return ' --- W'
else:
return str(int(power)) + 'W'
ac_loads = None if self.grid_power is None else self.grid_power - self.inverter_ac_in_power
delta = p(controller.delta) if controller.delta < 0 else '+' + p(controller.delta)
battery_power = self.battery_power if self.battery_available else None
soc_ = str(self.soc) + '%' if self.battery_available else '---'
print (State.name_of[controller.state])
print ('')
print ('time to CC: ' + self.time_to_calibration_charge_str)
print (' SOC: ' + soc_ + soc_setpoint())
print (' grid: ' + p(self.grid_power) + setpoint(TargetType.GRID_AC))
print (' battery: ' + p(battery_power) + setpoint(TargetType.BATTERY_DC))
print (' AC in: ' + p(self.inverter_ac_in_power) + ' ' + delta)
print (' AC out: ' + p(self.inverter_ac_out_power))
print (' AC loads: ' + p(ac_loads))
def choose_controller(self):
# type: () -> Controller
if self.grid_blackout:
return self.controller_bridge_grid_blackout()
if not self.battery_available:
return self.controller_no_battery()
if self.battery_cold:
return self.controller_heating()
if self.scheduled_charge:
return self.controller_scheduled_charge()
if self.must_do_calibration_charge:
return self.controller_calibration_charge()
if self.soc < self.min_soc:
return self.controller_charge_to_min_soc()
if not self.grid_meter_available:
return self.controller_no_grid_meter()
hold_min_soc = self.controller_hold_min_soc()
limit_discharge_power = self.controller_limit_discharge_power() # signed
lower_limit = [limit_discharge_power, hold_min_soc]
# No upper limit. We no longer actively limit charge power. DC/DC Charger inside the BMS will do that for us.
upper_limit = []
optimize_self_consumption = self.controller_optimize_self_consumption()
return optimize_self_consumption.clamp(lower_limit, upper_limit)
def update(self):
print('iteration started\n')
self.update_eoc()
if self.inverter_available:
controller = self.choose_controller()
power = self.inverter_ac_in_power + controller.delta
self.set_inverter_power_setpoint(power)
self.set_controller_state(controller.state)
self.print_system_stats(controller) # for debug
else:
self.set_controller_state(State.NO_INVERTER_AVAILABLE)
print('inverter not available!')
print('\niteration finished\n')
def main():
print('starting ' + __file__)
with InnovEnergyController() as service:
run_on_main_loop(service.update, UPDATE_PERIOD_MS)
print(__file__ + ' has shut down')
if __name__ == '__main__':
main()

View File

@ -0,0 +1,192 @@
import struct
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, List, Iterable, Union, AnyStr, Any
def read_bool(base_register, bit):
# type: (int, int) -> Callable[[BatteryStatus], bool]
# TODO: explain base register offset
register = base_register + int(bit/16)
bit = bit % 16
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):
# 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 @#%^&!
return (value + offset) * scale_factor
return get_value
def read_registers(register, count):
# type: (int, int) -> Callable[[BatteryStatus], List[int]]
start = register - cfg.BASE_ADDRESS
end = start + count
def get_value(status):
# type: (BatteryStatus) -> List[int]
return [x for x in status.modbus_data[start:end]]
return get_value
def comma_separated(values):
# type: (Iterable[str]) -> str
return ", ".join(set(values))
def count_bits(base_register, nb_of_registers, nb_of_bits, first_bit=0):
# type: (int, int, int, int) -> Callable[[BatteryStatus], int]
get_registers = read_registers(base_register, nb_of_registers)
end_bit = first_bit + nb_of_bits
def get_value(status):
# type: (BatteryStatus) -> int
registers = get_registers(status)
bin_registers = [bin(x)[-1:1:-1] for x in registers] # reverse the bits in each register so that bit0 is to the left
str_registers = [str(x).ljust(16, "0") for x in bin_registers] # add leading zeroes, so all registers are 16 chars long
bit_string = ''.join(str_registers) # join them, one long string of 0s and 1s
filtered_bits = bit_string[first_bit:end_bit] # take the first nb_of_bits bits starting at first_bit
return filtered_bits.count('1') # count 1s
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
# noinspection PyShadowingNames
def 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 const(constant):
# type: (any) -> Callable[[any], any]
def get(*args):
return constant
return get
def mean(numbers):
# type: (List[Union[float,int]]) -> float
return float(sum(numbers)) / len(numbers)
def first(ts, default=None):
return next((t for t in ts), default)
def bitfields_to_str(lists):
# type: (List[List[int]]) -> str
def or_lists():
# type: () -> Iterable[int]
length = len(first(lists))
n_lists = len(lists)
for i in range(0, length):
e = 0
for l in range(0, n_lists):
e = e | lists[l][i]
yield e
hexed = [
'{0:0>4X}'.format(x)
for x in or_lists()
]
return ' '.join(hexed)
def pack_string(string):
# type: (AnyStr) -> Any
data = string.encode('UTF-8')
return struct.pack('B', len(data)) + data
def read_bitmap(register):
# type: (int) -> Callable[[BatteryStatus], int]
def get_value(status):
# type: (BatteryStatus) -> int
value = status.modbus_data[register - cfg.BASE_ADDRESS]
return value
return get_value
def return_in_list(ts):
return ts
def first(ts):
return next(t for t in ts)
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

View File

@ -0,0 +1,134 @@
import config as cfg
# trick the pycharm type-checker into thinking Callable is in scope, not used at runtime
# noinspection PyUnreachableCode
if False:
from typing import Callable, List, Optional, AnyStr, Union, Any
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 ServiceSignal(object):
def __init__(self, dbus_path, get_value_or_const, unit=''):
# type: (str, Union[Callable[[],Any],Any], Optional[AnyStr] )->None
self.get_value_or_const = get_value_or_const
self.dbus_path = dbus_path
self.unit = unit
@property
def value(self):
try:
return self.get_value_or_const() # callable
except:
return self.get_value_or_const # value
class BatterySignal(object):
def __init__(self, dbus_path, aggregate, get_value, unit=''):
# type: (str, Callable[[List[any]],any], Callable[[BatteryStatus],any], Optional[AnyStr] )->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[any] -> any
function that combines the values of multiple batteries into one.
e.g. sum for currents, or mean for voltages
:param get_value: (BatteryStatus) -> any
function to extract the datum from the modbus record,
"""
self.dbus_path = dbus_path
self.aggregate = aggregate
self.get_value = get_value
self.unit = unit
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
self.n_strings = int(ampere_hours/cfg.AH_PER_STRING)
self.i_max = self.n_strings * cfg.I_MAX_PER_STRING
self.v_min = cfg.V_MIN
self.v_max = cfg.V_MAX
self.r_int_min = cfg.R_STRING_MIN / self.n_strings
self.r_int_max = cfg.R_STRING_MAX / self.n_strings
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
def serialize(self):
# type: () -> str
b = self.battery
s = cfg.INNOVENERGY_PROTOCOL_VERSION + '\n'
s += cfg.INSTALLATION_NAME + '\n'
s += str(b.slave_address) + '\n'
s += b.hardware_version + '\n'
s += b.firmware_version + '\n'
s += b.bms_version + '\n'
s += str(b.ampere_hours) + '\n'
for d in self.modbus_data:
s += str(d) + '\n'
return s
def read_file_one_line(file_name):
with open(file_name, 'r') as file:
return file.read().replace('\n', '').replace('\r', '').strip()
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 = ""

View File

@ -0,0 +1,674 @@
#!/usr/bin/python2 -u
# coding=utf-8
import logging
import re
import socket
import sys
import gobject
import signals
import config as cfg
from dbus.mainloop.glib import DBusGMainLoop
from pymodbus.client.sync import ModbusSerialClient as Modbus
from pymodbus.exceptions import ModbusException, ModbusIOException
from pymodbus.other_message import ReportSlaveIdRequest
from pymodbus.pdu import ExceptionResponse
from pymodbus.register_read_message import ReadInputRegistersResponse
from data import BatteryStatus, BatterySignal, Battery, ServiceSignal
from python_libs.ie_dbus.dbus_service import DBusService
import time
import os
import csv
import pika
import zipfile
import hashlib
import base64
import hmac
import requests
from datetime import datetime
import io
import json
from convert import first
CSV_DIR = "/data/csv_files/"
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, List, Iterable, NoReturn
RESET_REGISTER = 0x2087
def compress_csv_data(csv_data, file_name="data.csv"):
memory_stream = io.BytesIO()
# Create a zip archive in the memory buffer
with zipfile.ZipFile(memory_stream, 'w', zipfile.ZIP_DEFLATED) as archive:
# Add CSV data to the ZIP archive using writestr
archive.writestr(file_name, csv_data.encode('utf-8'))
# Get the compressed byte array from the memory buffer
compressed_bytes = memory_stream.getvalue()
# Encode the compressed byte array as a Base64 string
base64_string = base64.b64encode(compressed_bytes).decode('utf-8')
return base64_string
class S3config:
def __init__(self):
self.bucket = cfg.S3BUCKET
self.region = "sos-ch-dk-2"
self.provider = "exo.io"
self.key = cfg.S3KEY
self.secret = cfg.S3SECRET
self.content_type = "application/base64; charset=utf-8"
@property
def host(self):
return "{}.{}.{}".format(self.bucket, self.region, self.provider)
@property
def url(self):
return "https://{}".format(self.host)
def create_put_request(self, s3_path, data):
headers = self._create_request("PUT", s3_path)
url = "{}/{}".format(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 = "{}\n{}\n{}\n{}\n/{}/{}".format(
method, md5_hash, content_type, date, bucket.strip('/'), s3_path.strip('/')
)
signature = base64.b64encode(
hmac.new(s3_secret.encode(), payload.encode(), hashlib.sha1).digest()
).decode()
return "AWS {}:{}".format(s3_key, signature)
def SubscribeToQueue():
try:
connection = pika.BlockingConnection(pika.ConnectionParameters(host="10.2.0.11",
port=5672,
virtual_host="/",
credentials=pika.PlainCredentials("producer", "b187ceaddb54d5485063ddc1d41af66f")))
channel = connection.channel()
channel.queue_declare(queue="statusQueue", durable=True)
print("Subscribed to queue")
except Exception as ex:
print("An error occurred while connecting to the RabbitMQ queue:", ex)
return channel
previous_warnings = {}
previous_alarms = {}
class MessageType:
ALARM_OR_WARNING = "AlarmOrWarning"
HEARTBEAT = "Heartbeat"
class AlarmOrWarning:
def __init__(self, description, created_by):
self.date = datetime.now().strftime('%Y-%m-%d')
self.time = datetime.now().strftime('%H:%M:%S')
self.description = description
self.created_by = created_by
def to_dict(self):
return {
"Date": self.date,
"Time": self.time,
"Description": self.description,
"CreatedBy": self.created_by
}
channel = SubscribeToQueue()
# Create an S3config instance
s3_config = S3config()
INSTALLATION_ID=int(s3_config.bucket.split('-')[0])
PRODUCT_ID = 1
is_first_update = True
prev_status = 0
subscribed_to_queue_first_time = False
heartbit_interval = 0
def update_state_from_dictionaries(current_warnings, current_alarms, node_numbers):
global previous_warnings, previous_alarms, INSTALLATION_ID, PRODUCT_ID, is_first_update, channel, prev_status, heartbit_interval, subscribed_to_queue_first_time
heartbit_interval += 1
if is_first_update:
changed_warnings = current_warnings
changed_alarms = current_alarms
is_first_update = False
else:
changed_alarms = {}
changed_warnings = {}
# calculate the diff in warnings and alarms
prev_alarm_value_list = list(previous_alarms.values())
alarm_keys = list(previous_alarms.keys())
for i, alarm in enumerate(current_alarms.values()):
if alarm != prev_alarm_value_list[i]:
changed_alarms[alarm_keys[i]] = True
else:
changed_alarms[alarm_keys[i]] = False
prev_warning_value_list=list(previous_warnings.values())
warning_keys=list(previous_warnings.keys())
for i, warning in enumerate(current_warnings.values()):
if warning!=prev_warning_value_list[i]:
changed_warnings[warning_keys[i]]=True
else:
changed_warnings[warning_keys[i]]=False
status_message = {
"InstallationId": INSTALLATION_ID,
"Product": PRODUCT_ID,
"Status": 0,
"Type": 1,
"Warnings": [],
"Alarms": []
}
alarms_number_list = []
for node_number in node_numbers:
cnt = 0
for alarm_value in current_alarms.values():
if alarm_value:
cnt+=1
alarms_number_list.append(cnt)
warnings_number_list = []
for node_number in node_numbers:
cnt = 0
for warning_value in current_warnings.values():
if warning_value:
cnt+=1
warnings_number_list.append(cnt)
# Evaluate alarms
if any(changed_alarms.values()):
for i, changed_alarm in enumerate(changed_alarms.values()):
if changed_alarm and list(current_alarms.values())[i]:
status_message["Alarms"].append(AlarmOrWarning(list(current_alarms.keys())[i],"System").to_dict())
if any(changed_warnings.values()):
for i, changed_warning in enumerate(changed_warnings.values()):
if changed_warning and list(current_warnings.values())[i]:
status_message["Warnings"].append(AlarmOrWarning(list(current_warnings.keys())[i],"System").to_dict())
if any(current_alarms.values()):
status_message["Status"]=2
if not any(current_alarms.values()) and any(current_warnings.values()):
status_message["Status"]=1
if not any(current_alarms.values()) and not any(current_warnings.values()):
status_message["Status"]=0
if status_message["Status"]!=prev_status or len(status_message["Warnings"])>0 or len(status_message["Alarms"])>0:
prev_status=status_message["Status"]
status_message["Type"]=0
status_message = json.dumps(status_message)
channel.basic_publish(exchange="", routing_key="statusQueue", body=status_message)
print(status_message)
print("Message sent successfully")
elif heartbit_interval>=15 or not subscribed_to_queue_first_time:
print("Send heartbit message to rabbitmq")
heartbit_interval=0
subscribed_to_queue_first_time=True
status_message = json.dumps(status_message)
channel.basic_publish(exchange="", routing_key="statusQueue", body=status_message)
previous_warnings = current_warnings.copy()
previous_alarms = current_alarms.copy()
return status_message, alarms_number_list, warnings_number_list
def read_csv_as_string(file_path):
"""
Reads a CSV file from the given path and returns its content as a single string.
"""
try:
# Note: 'encoding' is not available in open() in Python 2.7, so we'll use 'codecs' module.
import codecs
with codecs.open(file_path, 'r', encoding='utf-8') as file:
return file.read()
except IOError as e:
if e.errno == 2: # errno 2 corresponds to "No such file or directory"
print("Error: The file {} does not exist.".format(file_path))
else:
print("IO error occurred: {}".format(str(e)))
return None
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_udp_socket():
# type: () -> socket
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
s.setblocking(False)
return s
def report_slave_id(modbus, slave_address):
# type: (Modbus, int) -> str
slave = str(slave_address)
logging.debug('requesting slave id from node ' + slave)
with modbus:
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
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():
slave_address = 0
n_missing = -255
while n_missing < 3:
slave_address += 1
try:
yield identify_battery(modbus, slave_address)
n_missing = 0
except Exception as e:
logging.info('failed to identify battery at {0} : {1}'.format(str(slave_address), str(e)))
n_missing += 1
logging.info('giving up searching for further batteries')
batteries = list(_identify_batteries()) # dont be lazy!
n = len(batteries)
logging.info('found ' + str(n) + (' battery' if n == 1 else ' batteries'))
return batteries
def parse_slave_id(modbus, slave_address):
# type: (Modbus, int) -> (str, str, int)
slave_id = report_slave_id(modbus, slave_address)
sid = re.sub(r'[^\x20-\x7E]', '', slave_id) # remove weird special chars
match = re.match('(?P<hw>48TL(?P<ah>[0-9]+)) *(?P<bms>.*)', sid)
if match is None:
raise Exception('no known battery found')
return match.group('hw').strip(), match.group('bms').strip(), int(match.group('ah').strip())
def read_firmware_version(modbus, slave_address):
# type: (Modbus, int) -> str
logging.debug('reading firmware version')
with modbus:
response = read_modbus_registers(modbus, slave_address, base_address=1054, count=1)
register = response.registers[0]
return '{0:0>4X}'.format(register)
def read_modbus_registers(modbus, slave_address, base_address=cfg.BASE_ADDRESS, count=cfg.NO_OF_REGISTERS):
# type: (Modbus, int, int, 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')
with modbus:
data = read_modbus_registers(modbus, battery.slave_address)
return BatteryStatus(battery, data.registers)
def publish_values_on_dbus(service, battery_signals, battery_statuses):
# type: (DBusService, Iterable[BatterySignal], Iterable[BatteryStatus]) -> ()
publish_individuals(service, battery_signals, battery_statuses)
publish_aggregates(service, battery_signals, battery_statuses)
def publish_aggregates(service, signals, battery_statuses):
# type: (DBusService, Iterable[BatterySignal], Iterable[BatteryStatus]) -> ()
for s in signals:
if s.aggregate is None:
continue
values = [s.get_value(battery_status) for battery_status in battery_statuses]
value = s.aggregate(values)
service.own_properties.set(s.dbus_path, value, s.unit)
def publish_individuals(service, signals, battery_statuses):
# type: (DBusService, Iterable[BatterySignal], Iterable[BatteryStatus]) -> ()
for signal in signals:
for battery_status in battery_statuses:
address = battery_status.battery.slave_address
dbus_path = '/_Battery/' + str(address) + signal.dbus_path
value = signal.get_value(battery_status)
service.own_properties.set(dbus_path, value, signal.unit)
def publish_service_signals(service, signals):
# type: (DBusService, Iterable[ServiceSignal]) -> NoReturn
for signal in signals:
service.own_properties.set(signal.dbus_path, signal.value, signal.unit)
def upload_status_to_innovenergy(sock, statuses):
# type: (socket, Iterable[BatteryStatus]) -> bool
logging.debug('upload status')
try:
for s in statuses:
sock.sendto(s.serialize(), (cfg.INNOVENERGY_SERVER_IP, cfg.INNOVENERGY_SERVER_PORT))
except:
logging.debug('FAILED')
return False
else:
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]
def reset_batteries(modbus, batteries):
# type: (Modbus, Iterable[Battery]) -> NoReturn
logging.info('Resetting batteries...')
for battery in batteries:
result = modbus.write_registers(RESET_REGISTER, [1], unit=battery.slave_address)
# expecting a ModbusIOException (timeout)
# BMS can no longer reply because it is already reset
success = isinstance(result, ModbusIOException)
outcome = 'successfully' if success else 'FAILED to'
logging.info('Battery {0} {1} reset'.format(str(battery.slave_address), outcome))
logging.info('Shutting down fz-sonick driver')
exit(0)
alive = True # global alive flag, watchdog_task clears it, update_task sets it
def create_update_task(modbus, service, batteries):
# type: (Modbus, DBusService, Iterable[Battery]) -> Callable[[],bool]
"""
Creates an update task which runs the main update function
and resets the alive flag
"""
_socket = init_udp_socket()
_signals = signals.init_battery_signals()
csv_signals = signals.create_csv_signals(first(batteries).firmware_version)
node_numbers = [battery.slave_address for battery in batteries]
warnings_signals, alarm_signals = signals.read_warning_and_alarm_flags()
current_warnings = {}
current_alarms = {}
def update_task():
# type: () -> bool
global alive
logging.debug('starting update cycle')
if service.own_properties.get('/ResetBatteries').value == 1:
reset_batteries(modbus, batteries)
statuses = [read_battery_status(modbus, battery) for battery in batteries]
# Iterate over each node and signal to create rows in the new format
for i, node in enumerate(node_numbers):
for s in warnings_signals:
signal_name = insert_id(s.name, i+1)
value = s.get_value(statuses[i])
current_warnings[signal_name] = value
for s in alarm_signals:
signal_name = insert_id(s.name, i+1)
value = s.get_value(statuses[i])
current_alarms[signal_name] = value
status_message, alarms_number_list, warnings_number_list = update_state_from_dictionaries(current_warnings, current_alarms, node_numbers)
publish_values_on_dbus(service, _signals, statuses)
create_csv_files(csv_signals, statuses, node_numbers, alarms_number_list, warnings_number_list)
upload_status_to_innovenergy(_socket, statuses)
logging.debug('finished update cycle\n')
alive = True
return True
return update_task
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 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, alarms_number_list, warnings_number_list):
timestamp = int(time.time())
if timestamp % 2 != 0:
timestamp-=1
if not os.path.exists(CSV_DIR):
os.makedirs(CSV_DIR)
csv_filename = "{}.csv".format(timestamp)
csv_path = os.path.join(CSV_DIR, csv_filename)
with open(csv_path, 'ab') as csvfile:
csv_writer = csv.writer(csvfile, delimiter=';')
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)
for i, node in enumerate(node_numbers):
csv_writer.writerow(["/Battery/Devices/{}/Alarms".format(str(i+1)), alarms_number_list[i], ""])
csv_writer.writerow(["/Battery/Devices/{}/Warnings".format(str(i+1)), warnings_number_list[i], ""])
for s in signals:
signal_name = insert_id(s.name, i+1)
value = s.get_value(statuses[i])
row_values = [signal_name, value, s.get_text]
csv_writer.writerow(row_values)
csv_data = read_csv_as_string(csv_path)
if csv_data is None:
print("error while reading csv as string")
return
# zip-comp additions
compressed_csv = compress_csv_data(csv_data)
compressed_filename = "{}.csv".format(timestamp)
response = s3_config.create_put_request(compressed_filename, compressed_csv)
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 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("INSIDE DBUS SONICK")
logging.basicConfig(level=cfg.LOG_LEVEL)
logging.info('starting ' + __file__)
tty = parse_cmdline_args(argv)
modbus = init_modbus(tty)
batteries = identify_batteries(modbus)
if len(batteries) <= 0:
sys.exit(2)
service = DBusService(service_name=cfg.SERVICE_NAME_PREFIX + tty)
service.own_properties.set('/ResetBatteries', value=False, writable=True) # initial value = False
main_loop = gobject.MainLoop()
service_signals = signals.init_service_signals(batteries)
publish_service_signals(service, service_signals)
update_task = create_update_task(modbus, service, batteries)
update_task() # run it right away, so that all props are initialized before anyone can ask
watchdog_task = create_watchdog_task(main_loop)
gobject.timeout_add(cfg.UPDATE_INTERVAL * 2, watchdog_task, priority = gobject.PRIORITY_LOW) # add watchdog first
gobject.timeout_add(cfg.UPDATE_INTERVAL, update_task, priority = gobject.PRIORITY_LOW) # call update once every update_interval
logging.info('starting gobject.MainLoop')
main_loop.run()
logging.info('gobject.MainLoop was shut down')
sys.exit(0xFF) # reaches this only on error
main(sys.argv[1:])

View File

@ -0,0 +1,156 @@
from logging import getLogger
import dbus
_log = getLogger(__name__)
# noinspection PyUnreachableCode
if False:
from typing import Any, Union, Dict
DbusString = Union[dbus.String, dbus.UTF8String, dbus.ObjectPath, dbus.Signature]
DbusInt = Union[dbus.Int16, dbus.Int32, dbus.Int64]
DbusDouble = dbus.Double
DbusBool = dbus.Boolean
DbusStringVariant = DbusString # TODO: variant_level constraint ?
DbusIntVariant = DbusInt
DbusDoubleVariant = DbusDouble
DbusBoolVariant = DbusBool
DbusValue = Union[DbusString, DbusInt, DbusDouble, DbusBool, DBUS_NONE]
DbusVariant = Union[DbusStringVariant, DbusIntVariant, DbusDoubleVariant, DbusBoolVariant, DBUS_NONE]
DbusTextDict = dbus.Dictionary
DbusVariantDict = dbus.Dictionary
DbusType = Union[DbusValue, DbusVariant, DbusVariantDict, DbusTextDict]
DBUS_NONE = dbus.Array([], signature=dbus.Signature('i'), variant_level=1) # DEFINED by victron
MAX_INT16 = 2 ** 15 - 1
MAX_INT32 = 2 ** 31 - 1
def dbus_uint32(value):
# type: (int) -> dbus.UInt32
if value < 0:
raise Exception('cannot convert negative value to UInt32')
return dbus.UInt32(value)
def dbus_int(value):
# type: (Union[int, long]) -> Union[dbus.Int16, dbus.Int32, dbus.Int64]
abs_value = abs(value)
if abs_value < MAX_INT16:
return dbus.Int16(value)
elif abs_value < MAX_INT32:
return dbus.Int32(value)
else:
return dbus.Int64(value)
def dbus_string(value):
# type: (Union[str, unicode]) -> DbusString
if isinstance(value, unicode):
return dbus.UTF8String(value)
else:
return dbus.String(value)
def dbus_double(value):
# type: (float) -> DbusDouble
return dbus.Double(value)
def dbus_bool(value):
# type: (bool) -> DbusBool
return dbus.Boolean(value)
# VARIANTS
def dbus_int_variant(value):
# type: (Union[int, long]) -> DbusIntVariant
abs_value = abs(value)
if abs_value < MAX_INT16:
return dbus.Int16(value, variant_level=1)
elif abs_value < MAX_INT32:
return dbus.Int32(value, variant_level=1)
else:
return dbus.Int64(value, variant_level=1)
def dbus_string_variant(value):
# type: (Union[str, unicode]) -> DbusStringVariant
if isinstance(value, unicode):
return dbus.UTF8String(value, variant_level=1)
else:
return dbus.String(value, variant_level=1)
def dbus_double_variant(value):
# type: (float) -> DbusDoubleVariant
return dbus.Double(value, variant_level=1)
def dbus_bool_variant(value):
# type: (bool) -> DbusBoolVariant
return dbus.Boolean(value, variant_level=1)
def dbus_variant(value):
# type: (Any) -> DbusVariant
if value is None:
return DBUS_NONE
if isinstance(value, float):
return dbus_double_variant(value)
if isinstance(value, bool):
return dbus_bool_variant(value)
if isinstance(value, (int, long)):
return dbus_int_variant(value)
if isinstance(value, (str, unicode)):
return dbus_string_variant(value)
# TODO: container types
if isinstance(value, list):
# Convert each element in the list to a dbus variant
dbus_array = [dbus_variant(item) for item in value]
if not dbus_array:
return dbus.Array([], signature='v') # Empty array with variant type
first_element = value[0]
if isinstance(first_element, float):
signature = 'd'
elif isinstance(first_element, bool):
signature = 'b'
elif isinstance(first_element, (int, long)):
signature = 'x'
elif isinstance(first_element, (str, unicode)):
signature = 's'
else:
signature = 'v' # default to variant if unknown
return dbus.Array(dbus_array, signature=signature)
raise TypeError('unsupported python type: ' + str(type(value)) + ' ' + str(value))
def dbus_value(value):
# type: (Any) -> DbusVariant
if value is None:
return DBUS_NONE
if isinstance(value, float):
return dbus_double(value)
if isinstance(value, bool):
return dbus_bool(value)
if isinstance(value, (int, long)):
return dbus_int(value)
if isinstance(value, (str, unicode)):
return dbus_string_variant(value)
# TODO: container types
raise TypeError('unsupported python type: ' + str(type(value)) + ' ' + str(value))

View File

@ -0,0 +1,202 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
from traceback import print_exc
from os import _exit as os_exit
from os import statvfs
import logging
from functools import update_wrapper
import dbus
logger = logging.getLogger(__name__)
VEDBUS_INVALID = dbus.Array([], signature=dbus.Signature('i'), variant_level=1)
# Use this function to make sure the code quits on an unexpected exception. Make sure to use it
# when using gobject.idle_add and also gobject.timeout_add.
# Without this, the code will just keep running, since gobject does not stop the mainloop on an
# exception.
# Example: gobject.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 tryed 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():
# For the CCGX, the 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.
# nice coincidence is that this also works fine when running on your (linux) development computer.
global __vrm_portal_id
if __vrm_portal_id:
return __vrm_portal_id
# Assume we are on linux
import fcntl, socket, struct
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
info = fcntl.ioctl(s.fileno(), 0x8927, struct.pack('256s', 'eth0'[:15]))
__vrm_portal_id = ''.join(['%02x' % ord(char) for char in info[18:24]])
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, ex:
logger.info("Error while retrieving free space for path %s: %s" % (path, ex))
return result
def get_load_averages():
c = read_file('/proc/loadavg')
return c.split(' ')[:3]
# Returns False if it cannot find a machine name. Otherwise returns the string
# containing the name
def get_machine_name():
c = read_file('/proc/device-tree/model')
if c != False:
return c.strip('\x00')
return read_file('/etc/venus/machine')
# 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, 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):
return dbus.Int32(value, variant_level=1)
if isinstance(value, str):
return dbus.String(value, variant_level=1)
if isinstance(value, unicode):
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, long):
return dbus.Int64(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 unicode(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([str(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
class reify(object):
""" Decorator to replace a property of an object with the calculated value,
to make it concrete. """
def __init__(self, wrapped):
self.wrapped = wrapped
update_wrapper(self, wrapped)
def __get__(self, inst, objtype=None):
if inst is None:
return self
v = self.wrapped(inst)
setattr(inst, self.wrapped.__name__, v)
return v

View File

@ -0,0 +1,496 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
import dbus.service
import logging
import traceback
import os
import weakref
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 = {}
# 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['/'] = self._create_tree_export(self._dbusconn, '/', self._get_tree_dict)
logging.info("registered ourselves on D-Bus as %s" % servicename)
def _get_tree_dict(self, path, get_text=False):
logging.debug("_get_tree_dict called for %s" % path)
r = {}
px = path
if not px.endswith('/'):
px += '/'
for p, item in self._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
# To force immediate deregistering of this dbus service and all its object paths, explicitly
# call __del__().
def __del__(self):
for node in self._dbusnodes.values():
node.__del__()
self._dbusnodes.clear()
for item in 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
# @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):
if onchangecallback is not None:
self._onchangecallbacks[path] = onchangecallback
item = VeDbusItemExport(
self._dbusconn, path, value, description, writeable,
self._value_changed, gettextcallback, deletecallback=self._item_deleted)
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] = self._create_tree_export(self._dbusconn, subPath, self._get_tree_dict)
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)
def _create_tree_export(self, bus, objectPath, get_value_handler):
return VeDbusTreeExport(bus, objectPath, get_value_handler)
# 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 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
"""
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):
## 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))
# 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 != 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
## 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, get_value_handler):
dbus.service.Object.__init__(self, bus, objectPath)
self._get_value_handler = get_value_handler
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]
@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 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):
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
# 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.local_set_value(None)
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):
if self._value == newvalue:
return
self._value = newvalue
changes = {}
changes['Value'] = wrap_dbus_value(newvalue)
changes['Text'] = self.GetText()
self.PropertiesChanged(changes)
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 newvalue == self._value:
return 0 # OK
# call the callback given to us, and check if new value is OK.
if (self._onchangecallback is None or
(self._onchangecallback is not None and self._onchangecallback(self.__dbus_object_path__, newvalue))):
self.local_set_value(newvalue)
return 0 # OK
return 2 # NOT OK
## Dbus exported method GetDescription
#
# Returns the a description.
# @param language A language code (e.g. ISO 639-1 en-US).
# @param length Lenght of the language string.
# @return description
@dbus.service.method('com.victronenergy.BusItem', in_signature='si', out_signature='s')
def GetDescription(self, language, length):
return self._description if self._description is not None else 'No description given'
## Dbus exported method GetValue
# Returns the value.
# @return the value when valid, and otherwise an empty array
@dbus.service.method('com.victronenergy.BusItem', out_signature='v')
def GetValue(self):
return wrap_dbus_value(self._value)
## Dbus exported method GetText
# Returns the value as string of the dbus-object-path.
# @return text A text-value. '---' when local value is invalid
@dbus.service.method('com.victronenergy.BusItem', out_signature='s')
def GetText(self):
if self._value is None:
return '---'
# Default conversion from dbus.Byte will get you a character (so 'T' instead of '84'), so we
# have to convert to int first. Note that if a dbus.Byte turns up here, it must have come from
# the application itself, as all data from the D-Bus should have been unwrapped by now.
if self._gettextcallback is None and type(self._value) == dbus.Byte:
return str(int(self._value))
if self._gettextcallback is None and self.__dbus_object_path__ == '/ProductId':
return "0x%X" % self._value
if self._gettextcallback is None:
return str(self._value)
return self._gettextcallback(self.__dbus_object_path__, self._value)
## The signal that indicates that the value has changed.
# Other processes connected to this BusItem object will have subscribed to the
# event when they want to track our state.
@dbus.service.signal('com.victronenergy.BusItem', signature='a{sv}')
def PropertiesChanged(self, changes):
pass
## This class behaves like a regular reference to a class method (eg. self.foo), but keeps a weak reference
## to the object which method is to be called.
## Use this object to break circular references.
class weak_functor:
def __init__(self, f):
self._r = weakref.ref(f.__self__)
self._f = weakref.ref(f.__func__)
def __call__(self, *args, **kargs):
r = self._r()
f = self._f()
if r == None or f == None:
return
f(r, *args, **kargs)

View File

@ -0,0 +1,3 @@
#!/bin/sh
exec 2>&1
exec multilog t s25000 n4 /var/log/dbus-fzsonick-48tl.TTY

View File

@ -0,0 +1,374 @@
# coding=utf-8
import config as cfg
from convert import mean, read_float, read_led_state, read_bool, count_bits, comma_separated, read_bitmap, return_in_list, first, read_hex_string
from data import BatterySignal, Battery, LedColor, ServiceSignal, BatteryStatus, LedState, CsvSignal
# noinspection PyUnreachableCode
if False:
from typing import List, Iterable
def read_voltage():
return read_float(register=999, scale_factor=0.01, offset=0)
def read_current():
return read_float(register=1000, scale_factor=0.01, offset=-10000)
def read_limb_bitmap():
return read_bitmap(1059)
def read_power(status):
return int(read_current()(status) * read_voltage()(status))
def interpret_limb_bitmap(bitmap_value):
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 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):
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):
di = i_limit - i
dv = di * r_int
p_limit = i_limit * (v + dv)
return p_limit
def calc_max_charge_power(status):
n_strings = cfg.NUM_OF_STRINGS_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 = max(p_limit, 0)
return int(p_limit)
def calc_max_discharge_power(status):
n_strings = cfg.NUM_OF_STRINGS_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 read_switch_closed(status):
value = read_bool(base_register=1013, bit=0)(status)
if value:
return False
return True
def read_alarm_out_active(status):
value = read_bool(base_register=1013, bit=1)(status)
if value:
return False
return True
def read_aux_relay(status):
value = read_bool(base_register=1013, bit=4)(status)
if value:
return False
return True
def hex_string_to_ascii(hex_string):
hex_string = hex_string.replace(" ", "")
ascii_string = ''.join([chr(int(hex_string[i:i+2], 16)) for i in range(0, len(hex_string), 2)])
return ascii_string
def init_service_signals(batteries):
print("INSIDE INIT SERVICE SIGNALS")
n_batteries = len(batteries)
product_name = cfg.PRODUCT_NAME + ' x' + str(n_batteries)
return [
ServiceSignal('/NbOfBatteries', n_batteries),
ServiceSignal('/Mgmt/ProcessName', __file__),
ServiceSignal('/Mgmt/ProcessVersion', cfg.SOFTWARE_VERSION),
ServiceSignal('/Mgmt/Connection', cfg.CONNECTION),
ServiceSignal('/DeviceInstance', cfg.DEVICE_INSTANCE),
ServiceSignal('/ProductName', product_name),
ServiceSignal('/ProductId', cfg.PRODUCT_ID),
ServiceSignal('/Connected', 1)
]
def init_battery_signals():
print("START INIT SIGNALS")
battery_status_reader = read_hex_string(1060, 2)
def read_eoc_reached(status):
battery_status_string = battery_status_reader(status)
return hex_string_to_ascii(battery_status_string) == "EOC_"
def read_battery_cold(status):
return \
read_led_state(register=1004, led=LedColor.green)(status) >= LedState.blinking_slow and \
read_led_state(register=1004, led=LedColor.blue)(status) >= LedState.blinking_slow
def read_soc(status):
soc = read_float(register=1053, scale_factor=0.1, offset=0)(status)
if soc > 99.9 and not read_eoc_reached(status):
return 99.9
if soc >= 99.9 and read_eoc_reached(status):
return 100
return soc
def number_of_active_strings(status):
return cfg.NUM_OF_STRINGS_PER_BATTERY - limp_strings_value(status)
def max_discharge_current(status):
return number_of_active_strings(status) * cfg.I_MAX_PER_STRING
def max_charge_current(status):
return status.battery.ampere_hours / 2
return [
BatterySignal('/TimeToTOCRequest', max, read_float(register=1052)),
BatterySignal('/EOCReached', return_in_list, read_eoc_reached),
BatterySignal('/NumOfLimbStrings', return_in_list, limp_strings_value),
BatterySignal('/Dc/0/Voltage', mean, get_value=read_voltage(), unit='V'),
BatterySignal('/Dc/0/Current', sum, get_value=read_current(), unit='A'),
BatterySignal('/Dc/0/Power', sum, get_value=read_power, unit='W'),
BatterySignal('/BussVoltage', mean, read_float(register=1001, scale_factor=0.01, offset=0), unit='V'),
BatterySignal('/Soc', mean, read_soc, unit='%'),
BatterySignal('/LowestSoc', min, read_float(register=1053, scale_factor=0.1, offset=0), unit='%'),
BatterySignal('/Dc/0/Temperature', mean, read_float(register=1003, scale_factor=0.1, offset=-400), unit='C'),
BatterySignal('/Dc/0/LowestTemperature', min, read_float(register=1003, scale_factor=0.1, offset=-400), unit='C'),
BatterySignal('/WarningFlags/TaM1', return_in_list, read_bool(base_register=1005, bit=1)),
BatterySignal('/WarningFlags/TbM1', return_in_list, read_bool(base_register=1005, bit=4)),
BatterySignal('/WarningFlags/VBm1', return_in_list, read_bool(base_register=1005, bit=6)),
BatterySignal('/WarningFlags/VBM1', return_in_list, read_bool(base_register=1005, bit=8)),
BatterySignal('/WarningFlags/IDM1', return_in_list, read_bool(base_register=1005, bit=10)),
BatterySignal('/WarningFlags/vsm1', return_in_list, read_bool(base_register=1005, bit=22)),
BatterySignal('/WarningFlags/vsM1', return_in_list, read_bool(base_register=1005, bit=24)),
BatterySignal('/WarningFlags/iCM1', return_in_list, read_bool(base_register=1005, bit=26)),
BatterySignal('/WarningFlags/iDM1', return_in_list, read_bool(base_register=1005, bit=28)),
BatterySignal('/WarningFlags/MID1', return_in_list, read_bool(base_register=1005, bit=30)),
BatterySignal('/WarningFlags/BLPW', return_in_list, read_bool(base_register=1005, bit=32)),
BatterySignal('/WarningFlags/CCBF', return_in_list, read_bool(base_register=1005, bit=33)),
BatterySignal('/WarningFlags/Ah_W', return_in_list, read_bool(base_register=1005, bit=35)),
BatterySignal('/WarningFlags/MPMM', return_in_list, read_bool(base_register=1005, bit=38)),
BatterySignal('/WarningFlags/TCdi', return_in_list, read_bool(base_register=1005, bit=40)),
BatterySignal('/WarningFlags/LMPW', return_in_list, read_bool(base_register=1005, bit=44)),
BatterySignal('/WarningFlags/TOCW', return_in_list, read_bool(base_register=1005, bit=47)),
BatterySignal('/WarningFlags/BUSL', return_in_list, read_bool(base_register=1005, bit=49)),
BatterySignal('/AlarmFlags/Tam', return_in_list, read_bool(base_register=1005, bit=0)),
BatterySignal('/AlarmFlags/TaM2', return_in_list, read_bool(base_register=1005, bit=2)),
BatterySignal('/AlarmFlags/Tbm', return_in_list, read_bool(base_register=1005, bit=3)),
BatterySignal('/AlarmFlags/TbM2', return_in_list, read_bool(base_register=1005, bit=5)),
BatterySignal('/AlarmFlags/VBm2', return_in_list, read_bool(base_register=1005, bit=7)),
BatterySignal('/AlarmFlags/VBM2', return_in_list, read_bool(base_register=1005, bit=9)),
BatterySignal('/AlarmFlags/IDM2', return_in_list, read_bool(base_register=1005, bit=11)),
BatterySignal('/AlarmFlags/ISOB', return_in_list, read_bool(base_register=1005, bit=12)),
BatterySignal('/AlarmFlags/MSWE', return_in_list, read_bool(base_register=1005, bit=13)),
BatterySignal('/AlarmFlags/FUSE', return_in_list, read_bool(base_register=1005, bit=14)),
BatterySignal('/AlarmFlags/HTRE', return_in_list, read_bool(base_register=1005, bit=15)),
BatterySignal('/AlarmFlags/TCPE', return_in_list, read_bool(base_register=1005, bit=16)),
BatterySignal('/AlarmFlags/STRE', return_in_list, read_bool(base_register=1005, bit=17)),
BatterySignal('/AlarmFlags/CME', return_in_list, read_bool(base_register=1005, bit=18)),
BatterySignal('/AlarmFlags/HWFL', return_in_list, read_bool(base_register=1005, bit=19)),
BatterySignal('/AlarmFlags/HWEM', return_in_list, read_bool(base_register=1005, bit=20)),
BatterySignal('/AlarmFlags/ThM', return_in_list, read_bool(base_register=1005, bit=21)),
BatterySignal('/AlarmFlags/vsm2', return_in_list, read_bool(base_register=1005, bit=23)),
BatterySignal('/AlarmFlags/vsM2', return_in_list, read_bool(base_register=1005, bit=25)),
BatterySignal('/AlarmFlags/iCM2', return_in_list, read_bool(base_register=1005, bit=27)),
BatterySignal('/AlarmFlags/iDM2', return_in_list, read_bool(base_register=1005, bit=29)),
BatterySignal('/AlarmFlags/MID2', return_in_list, read_bool(base_register=1005, bit=31)),
BatterySignal('/AlarmFlags/HTFS', return_in_list, read_bool(base_register=1005, bit=42)),
BatterySignal('/AlarmFlags/DATA', return_in_list, read_bool(base_register=1005, bit=43)),
BatterySignal('/AlarmFlags/LMPA', return_in_list, read_bool(base_register=1005, bit=45)),
BatterySignal('/AlarmFlags/HEBT', return_in_list, read_bool(base_register=1005, bit=46)),
BatterySignal('/AlarmFlags/CURM', return_in_list, read_bool(base_register=1005, bit=48)),
BatterySignal('/Diagnostics/LedStatus/Red', first, read_led_state(register=1004, led=LedColor.red)),
BatterySignal('/Diagnostics/LedStatus/Blue', first, read_led_state(register=1004, led=LedColor.blue)),
BatterySignal('/Diagnostics/LedStatus/Green', first, read_led_state(register=1004, led=LedColor.green)),
BatterySignal('/Diagnostics/LedStatus/Amber', first, read_led_state(register=1004, led=LedColor.amber)),
BatterySignal('/Diagnostics/IoStatus/MainSwitchClosed', return_in_list, read_switch_closed),
BatterySignal('/Diagnostics/IoStatus/AlarmOutActive', return_in_list, read_alarm_out_active),
BatterySignal('/Diagnostics/IoStatus/InternalFanActive', return_in_list, read_bool(base_register=1013, bit=2)),
BatterySignal('/Diagnostics/IoStatus/VoltMeasurementAllowed', return_in_list, read_bool(base_register=1013, bit=3)),
BatterySignal('/Diagnostics/IoStatus/AuxRelay', return_in_list, read_aux_relay),
BatterySignal('/Diagnostics/IoStatus/RemoteState', return_in_list, read_bool(base_register=1013, bit=5)),
BatterySignal('/Diagnostics/IoStatus/RiscOn', return_in_list, read_bool(base_register=1013, bit=6)),
BatterySignal('/IoStatus/BatteryCold', any, read_battery_cold),
BatterySignal('/Info/MaxDischargeCurrent', sum, max_discharge_current, unit='A'),
BatterySignal('/Info/MaxChargeCurrent', sum, max_charge_current, unit='A'),
BatterySignal('/Info/MaxChargeVoltage', min, lambda bs: bs.battery.v_max, unit='V'),
BatterySignal('/Info/MinDischargeVoltage', max, lambda bs: bs.battery.v_min, unit='V'),
BatterySignal('/Info/BatteryLowVoltage', max, lambda bs: bs.battery.v_min - 2, unit='V'),
BatterySignal('/Info/NumberOfStrings', sum, number_of_active_strings),
BatterySignal('/Info/MaxChargePower', sum, calc_max_charge_power),
BatterySignal('/Info/MaxDischargePower', sum, calc_max_discharge_power),
BatterySignal('/FirmwareVersion', comma_separated, lambda bs: bs.battery.firmware_version),
BatterySignal('/HardwareVersion', comma_separated, lambda bs: bs.battery.hardware_version),
BatterySignal('/BmsVersion', comma_separated, lambda bs: bs.battery.bms_version)
]
def create_csv_signals(firmware_version):
total_current = read_float(register=1062, scale_factor=0.01, offset=-10000)
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 = read_float(register=1002, scale_factor=0.1, offset=-10000)
def read_soc_ah(status):
return soc_ah(status)
def return_led_state(status, color):
led_state = read_led_state(register=1004, led=color)(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_blue(status):
return return_led_state(status, LedColor.blue)
def return_led_state_red(status):
return return_led_state(status, LedColor.red)
def return_led_state_green(status):
return return_led_state(status, LedColor.green)
def return_led_state_amber(status):
return return_led_state(status, LedColor.amber)
battery_status_reader = read_hex_string(1060, 2)
def read_eoc_reached(status):
battery_status_string = battery_status_reader(status)
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:
hex_value_fun = read_hex_string(reg, 1)
hex_value = hex_value_fun(status)
serial_parts.append(hex_value.replace(' ', ''))
serial_number = ''.join(serial_parts).rstrip('0')
return serial_number
def time_since_toc_in_time_format(status):
time_in_minutes = read_float(register=1052)(status)
total_seconds = int(time_in_minutes * 60)
days = total_seconds // (24 * 3600)
total_seconds = total_seconds % (24 * 3600)
hours = total_seconds // 3600
total_seconds %= 3600
minutes = total_seconds // 60
seconds = total_seconds % 60
return "{}.{:02}:{:02}:{:02}".format(days, hours, minutes, seconds)
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', read_float(register=1053, scale_factor=0.1, offset=0), '%'),
CsvSignal('/Battery/Devices/Temperatures/Cells/Average', read_float(register=1003, scale_factor=0.1, offset=-400), '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', lambda status: int((read_limb_bitmap()(status) & 0b00001) != 0)),
CsvSignal('/Battery/Devices/BatteryStrings/String2Active', lambda status: int((read_limb_bitmap()(status) & 0b00010) != 0)),
CsvSignal('/Battery/Devices/BatteryStrings/String3Active', lambda status: int((read_limb_bitmap()(status) & 0b00100) != 0)),
CsvSignal('/Battery/Devices/BatteryStrings/String4Active', lambda status: int((read_limb_bitmap()(status) & 0b01000) != 0)),
CsvSignal('/Battery/Devices/BatteryStrings/String5Active', lambda status: int((read_limb_bitmap()(status) & 0b10000) != 0)),
CsvSignal('/Battery/Devices/IoStatus/ConnectedToDcBus', read_switch_closed),
CsvSignal('/Battery/Devices/IoStatus/AlarmOutActive', read_alarm_out_active),
CsvSignal('/Battery/Devices/IoStatus/InternalFanActive', read_bool(base_register=1013, bit=2)),
CsvSignal('/Battery/Devices/IoStatus/VoltMeasurementAllowed', read_bool(base_register=1013, bit=3)),
CsvSignal('/Battery/Devices/IoStatus/AuxRelayBus', read_aux_relay),
CsvSignal('/Battery/Devices/IoStatus/RemoteStateActive', read_bool(base_register=1013, bit=5)),
CsvSignal('/Battery/Devices/IoStatus/RiscActive', read_bool(base_register=1013, bit=6)),
CsvSignal('/Battery/Devices/Eoc', read_eoc_reached),
CsvSignal('/Battery/Devices/SerialNumber', read_serial_number),
CsvSignal('/Battery/Devices/TimeSinceTOC', time_since_toc_in_time_format),
CsvSignal('/Battery/Devices/MaxChargePower', calc_max_charge_power),
CsvSignal('/Battery/Devices/MaxDischargePower', calc_max_discharge_power),
]
def read_warning_and_alarm_flags():
return [
# Warnings
CsvSignal('/Battery/Devices/WarningFlags/TaM1', read_bool(base_register=1005, bit=1)),
CsvSignal('/Battery/Devices/WarningFlags/TbM1', read_bool(base_register=1005, bit=4)),
CsvSignal('/Battery/Devices/WarningFlags/VBm1', read_bool(base_register=1005, bit=6)),
CsvSignal('/Battery/Devices/WarningFlags/VBM1', read_bool(base_register=1005, bit=8)),
CsvSignal('/Battery/Devices/WarningFlags/IDM1', read_bool(base_register=1005, bit=10)),
CsvSignal('/Battery/Devices/WarningFlags/vsm1', read_bool(base_register=1005, bit=22)),
CsvSignal('/Battery/Devices/WarningFlags/vsM1', read_bool(base_register=1005, bit=24)),
CsvSignal('/Battery/Devices/WarningFlags/iCM1', read_bool(base_register=1005, bit=26)),
CsvSignal('/Battery/Devices/WarningFlags/iDM1', read_bool(base_register=1005, bit=28)),
CsvSignal('/Battery/Devices/WarningFlags/MID1', read_bool(base_register=1005, bit=30)),
CsvSignal('/Battery/Devices/WarningFlags/BLPW', read_bool(base_register=1005, bit=32)),
CsvSignal('/Battery/Devices/WarningFlags/CCBF', read_bool(base_register=1005, bit=33)),
CsvSignal('/Battery/Devices/WarningFlags/Ah_W', read_bool(base_register=1005, bit=35)),
CsvSignal('/Battery/Devices/WarningFlags/MPMM', read_bool(base_register=1005, bit=38)),
CsvSignal('/Battery/Devices/WarningFlags/TCdi', read_bool(base_register=1005, bit=40)),
CsvSignal('/Battery/Devices/WarningFlags/LMPW', read_bool(base_register=1005, bit=44)),
CsvSignal('/Battery/Devices/WarningFlags/TOCW', read_bool(base_register=1005, bit=47)),
CsvSignal('/Battery/Devices/WarningFlags/BUSL', read_bool(base_register=1005, bit=49)),
], [
# Alarms
CsvSignal('/Battery/Devices/AlarmFlags/Tam', read_bool(base_register=1005, bit=0)),
CsvSignal('/Battery/Devices/AlarmFlags/TaM2', read_bool(base_register=1005, bit=2)),
CsvSignal('/Battery/Devices/AlarmFlags/Tbm', read_bool(base_register=1005, bit=3)),
CsvSignal('/Battery/Devices/AlarmFlags/TbM2', read_bool(base_register=1005, bit=5)),
CsvSignal('/Battery/Devices/AlarmFlags/VBm2', read_bool(base_register=1005, bit=7)),
CsvSignal('/Battery/Devices/AlarmFlags/VBM2', read_bool(base_register=1005, bit=9)),
CsvSignal('/Battery/Devices/AlarmFlags/IDM2', read_bool(base_register=1005, bit=11)),
CsvSignal('/Battery/Devices/AlarmFlags/ISOB', read_bool(base_register=1005, bit=12)),
CsvSignal('/Battery/Devices/AlarmFlags/MSWE', read_bool(base_register=1005, bit=13)),
CsvSignal('/Battery/Devices/AlarmFlags/FUSE', read_bool(base_register=1005, bit=14)),
CsvSignal('/Battery/Devices/AlarmFlags/HTRE', read_bool(base_register=1005, bit=15)),
CsvSignal('/Battery/Devices/AlarmFlags/TCPE', read_bool(base_register=1005, bit=16)),
CsvSignal('/Battery/Devices/AlarmFlags/STRE', read_bool(base_register=1005, bit=17)),
CsvSignal('/Battery/Devices/AlarmFlags/CME', read_bool(base_register=1005, bit=18)),
CsvSignal('/Battery/Devices/AlarmFlags/HWFL', read_bool(base_register=1005, bit=19)),
CsvSignal('/Battery/Devices/AlarmFlags/HWEM', read_bool(base_register=1005, bit=20)),
CsvSignal('/Battery/Devices/AlarmFlags/ThM', read_bool(base_register=1005, bit=21)),
CsvSignal('/Battery/Devices/AlarmFlags/vsm2', read_bool(base_register=1005, bit=23)),
CsvSignal('/Battery/Devices/AlarmFlags/vsM2', read_bool(base_register=1005, bit=25)),
CsvSignal('/Battery/Devices/AlarmFlags/iCM2', read_bool(base_register=1005, bit=27)),
CsvSignal('/Battery/Devices/AlarmFlags/iDM2', read_bool(base_register=1005, bit=29)),
CsvSignal('/Battery/Devices/AlarmFlags/MID2', read_bool(base_register=1005, bit=31)),
CsvSignal('/Battery/Devices/AlarmFlags/HTFS', read_bool(base_register=1005, bit=42)),
CsvSignal('/Battery/Devices/AlarmFlags/DATA', read_bool(base_register=1005, bit=43)),
CsvSignal('/Battery/Devices/AlarmFlags/LMPA', read_bool(base_register=1005, bit=45)),
CsvSignal('/Battery/Devices/AlarmFlags/HEBT', read_bool(base_register=1005, bit=46)),
CsvSignal('/Battery/Devices/AlarmFlags/CURM', read_bool(base_register=1005, bit=48)),
]

View File

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