Innovenergy_trunk/firmware/opt/dbus-fz-sonick-48tl-with-s3/data.py

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 = ""