srne-mqtt/srnemqtt/solar_types.py

301 lines
11 KiB
Python

# -*- coding: utf-8 -*-
import struct
from abc import ABC, abstractmethod
from enum import Enum, unique
from typing import Any, Callable, Dict, Optional
@unique
class DataName(str, Enum):
BATTERY_CHARGE = "battery_charge"
BATTERY_VOLTAGE = "battery_voltage"
BATTERY_CURRENT = "battery_current"
INTERNAL_TEMPERATURE = "internal_temperature"
BATTERY_TEMPERATURE = "battery_temperature"
LOAD_VOLTAGE = "load_voltage"
LOAD_CURRENT = "load_current"
LOAD_POWER = "load_power"
PANEL_VOLTAGE = "panel_voltage"
PANEL_CURRENT = "panel_current"
PANEL_POWER = "panel_power"
LOAD_ENABLED = "load_enabled"
BATTERY_VOLTAGE_MIN = "battery_voltage_min"
BATTERY_VOLTAGE_MAX = "battery_voltage_max"
CHARGE_MAX_CURRENT = "charge_max_current"
DISCHARGE_MAX_CURRENT = "discharge_max_current"
CHARGE_MAX_POWER = "charge_max_power"
DISCHARGE_MAX_POWER = "discharge_max_power"
CHARGE_AMP_HOUR = "charge_amp_hour"
DISCHARGE_AMP_HOUR = "discharge_amp_hour"
PRODUCTION_ENERGY = "production_energy"
CONSUMPTION_ENERGY = "consumption_energy"
RUN_DAYS = "run_days"
DISCHARGE_COUNT = "discharge_count"
FULL_CHARGE_COUNT = "full_charge_count"
TOTAL_CHARGE_AMP_HOURS = "total_charge_amp_hours"
TOTAL_DISCHARGE_AMP_HOURS = "total_discharge_amp_hours"
TOTAL_PRODUCTION_ENERGY = "total_production_energy"
TOTAL_CONSUMPTION_ENERGY = "total_consumption_energy"
CALCULATED_BATTERY_POWER = "calculated_battery_power"
CALCULATED_PANEL_POWER = "calculated_panel_power"
CALCULATED_LOAD_POWER = "calculated_load_power"
def __repr__(self):
return repr(self.value)
def __str__(self):
return self.value
class DataItem:
name: DataName
st_format: str
unit: Optional[str]
transformation: Optional[Callable]
def __init__(
self,
name: DataName,
st_format: str,
unit: Optional[str] = None,
transform: Optional[Callable] = None,
):
self.name = name
self.st_format = st_format
self.unit = unit
self.transformation = transform
if self.st_format[0] not in "@=<>!":
self.st_format = "!" + self.st_format
@property
def st_size(self) -> int:
return struct.calcsize(self.st_format)
def transform(self, data):
if self.transformation is None:
return data
return self.transformation(data)
def parse_temperature(bin):
if bin & 0x80:
return (bin & 0x7F) * -1
return bin & 0x7F
DATA_BATTERY_STATE = [
DataItem(DataName.BATTERY_CHARGE, "H", "%"),
DataItem(DataName.BATTERY_VOLTAGE, "H", "V", lambda n: n / 10),
DataItem(DataName.BATTERY_CURRENT, "H", "A", lambda n: n / 100),
DataItem(DataName.INTERNAL_TEMPERATURE, "B", "°C", parse_temperature),
DataItem(DataName.BATTERY_TEMPERATURE, "B", "°C", parse_temperature),
DataItem(DataName.LOAD_VOLTAGE, "H", "V", lambda n: n / 10),
DataItem(DataName.LOAD_CURRENT, "H", "A", lambda n: n / 100),
DataItem(DataName.LOAD_POWER, "H", "W"),
DataItem(DataName.PANEL_VOLTAGE, "H", "V", lambda n: n / 10),
DataItem(DataName.PANEL_CURRENT, "H", "A", lambda n: n / 100),
DataItem(DataName.PANEL_POWER, "H", "W"),
DataItem(DataName.LOAD_ENABLED, "x?", transform=bool),
]
HISTORICAL_DATA = [
DataItem(DataName.BATTERY_VOLTAGE_MIN, "H", "V", lambda n: n / 10),
DataItem(DataName.BATTERY_VOLTAGE_MAX, "H", "V", lambda n: n / 10),
DataItem(DataName.CHARGE_MAX_CURRENT, "H", "A", lambda n: n / 100),
DataItem(DataName.DISCHARGE_MAX_CURRENT, "H", "A", lambda n: n / 100),
DataItem(DataName.CHARGE_MAX_POWER, "H", "W"),
DataItem(DataName.DISCHARGE_MAX_POWER, "H", "W"),
DataItem(DataName.CHARGE_AMP_HOUR, "H", "Ah"),
DataItem(DataName.DISCHARGE_AMP_HOUR, "H", "Ah"),
DataItem(DataName.PRODUCTION_ENERGY, "H", "Wh"),
DataItem(DataName.CONSUMPTION_ENERGY, "H", "Wh"),
#
DataItem(DataName.RUN_DAYS, "H"),
DataItem(DataName.DISCHARGE_COUNT, "H"),
DataItem(DataName.FULL_CHARGE_COUNT, "H"),
DataItem(DataName.TOTAL_CHARGE_AMP_HOURS, "L", "Ah"),
DataItem(DataName.TOTAL_DISCHARGE_AMP_HOURS, "L", "Ah"),
DataItem(DataName.TOTAL_PRODUCTION_ENERGY, "L", "Wh"),
DataItem(DataName.TOTAL_CONSUMPTION_ENERGY, "L", "Wh"),
]
class DecodedData(ABC):
@abstractmethod
def __init__(self, data: bytes | bytearray | memoryview) -> None:
...
@abstractmethod
def as_dict(self) -> Dict[DataName, Any]:
...
class ChargerState(DecodedData):
battery_charge: int
battery_voltage: float
battery_current: float
internal_temperature: int
battery_temperature: int
load_voltage: float
load_current: float
load_power: float
panel_voltage: float
panel_current: float
panel_power: float
load_enabled: bool
def __init__(self, data: bytes | bytearray | memoryview) -> None:
(
_battery_charge,
_battery_voltage,
_battery_current,
_internal_temperature,
_battery_temperature,
_load_voltage,
_load_current,
_load_power,
_panel_voltage,
_panel_current,
_panel_power,
_load_enabled,
) = struct.unpack("!HHHBBHHHHHHx?", data)
self.battery_charge = _battery_charge
self.battery_voltage = _battery_voltage / 10
self.battery_current = _battery_current / 100
self.internal_temperature = parse_temperature(_internal_temperature)
self.battery_temperature = parse_temperature(_battery_temperature)
self.load_voltage = _load_voltage / 10
self.load_current = _load_current / 100
self.load_power = _load_power
self.panel_voltage = _panel_voltage / 10
self.panel_current = _panel_current / 100
self.panel_power = _panel_power
self.load_enabled = bool(_load_enabled)
@property
def calculated_battery_power(self) -> float:
return self.battery_voltage * self.battery_current
@property
def calculated_panel_power(self) -> float:
return self.panel_voltage * self.panel_current
@property
def calculated_load_power(self) -> float:
return self.load_voltage * self.load_current
def as_dict(self):
return {
DataName.BATTERY_CHARGE: self.battery_charge,
DataName.BATTERY_VOLTAGE: self.battery_voltage,
DataName.BATTERY_CURRENT: self.battery_current,
DataName.INTERNAL_TEMPERATURE: self.internal_temperature,
DataName.BATTERY_TEMPERATURE: self.battery_temperature,
DataName.LOAD_VOLTAGE: self.load_voltage,
DataName.LOAD_CURRENT: self.load_current,
DataName.LOAD_POWER: self.load_power,
DataName.PANEL_VOLTAGE: self.panel_voltage,
DataName.PANEL_CURRENT: self.panel_current,
DataName.PANEL_POWER: self.panel_power,
DataName.LOAD_ENABLED: self.load_enabled,
DataName.CALCULATED_BATTERY_POWER: self.calculated_battery_power,
DataName.CALCULATED_PANEL_POWER: self.calculated_panel_power,
DataName.CALCULATED_LOAD_POWER: self.calculated_load_power,
}
class HistoricalData(DecodedData):
battery_voltage_min: float
battery_voltage_max: float
charge_max_current: float
discharge_max_current: float
charge_max_power: int
discharge_max_power: int
charge_amp_hour: int
discharge_amp_hour: int
production_energy: int
consumption_energy: int
def __init__(self, data: bytes | bytearray | memoryview) -> None:
(
_battery_voltage_min,
_battery_voltage_max,
_charge_max_current,
_discharge_max_current,
_charge_max_power,
_discharge_max_power,
_charge_amp_hour,
_discharge_amp_hour,
_production_energy,
_consumption_energy,
) = struct.unpack("!HHHHHHHHHH", data)
self.battery_voltage_min = _battery_voltage_min / 10
self.battery_voltage_max = _battery_voltage_max / 10
self.charge_max_current = _charge_max_current / 100
self.discharge_max_current = _discharge_max_current / 100
self.charge_max_power = _charge_max_power
self.discharge_max_power = _discharge_max_power
self.charge_amp_hour = _charge_amp_hour
self.discharge_amp_hour = _discharge_amp_hour
self.production_energy = _production_energy
self.consumption_energy = _consumption_energy
def as_dict(self):
return {
DataName.BATTERY_VOLTAGE_MIN: self.battery_voltage_min,
DataName.BATTERY_VOLTAGE_MAX: self.battery_voltage_max,
DataName.CHARGE_MAX_CURRENT: self.charge_max_current,
DataName.DISCHARGE_MAX_CURRENT: self.discharge_max_current,
DataName.CHARGE_MAX_POWER: self.charge_max_power,
DataName.DISCHARGE_MAX_POWER: self.discharge_max_power,
DataName.CHARGE_AMP_HOUR: self.charge_amp_hour,
DataName.DISCHARGE_AMP_HOUR: self.discharge_amp_hour,
DataName.PRODUCTION_ENERGY: self.production_energy,
DataName.CONSUMPTION_ENERGY: self.consumption_energy,
}
class HistoricalExtraInfo(DecodedData):
run_days: int
discharge_count: int
full_charge_count: int
total_charge_amp_hours: int
total_discharge_amp_hours: int
total_production_energy: int
total_consumption_energy: int
def __init__(self, data: bytes | bytearray | memoryview) -> None:
(
_run_days,
_discharge_count,
_full_charge_count,
_total_charge_amp_hours,
_total_discharge_amp_hours,
_total_production_energy,
_total_consumption_energy,
) = struct.unpack("!HHHLLLL", data)
self.run_days = _run_days
self.discharge_count = _discharge_count
self.full_charge_count = _full_charge_count
self.total_charge_amp_hours = _total_charge_amp_hours
self.total_discharge_amp_hours = _total_discharge_amp_hours
self.total_production_energy = _total_production_energy
self.total_consumption_energy = _total_consumption_energy
def as_dict(self):
return {
DataName.RUN_DAYS: self.run_days,
DataName.DISCHARGE_COUNT: self.discharge_count,
DataName.FULL_CHARGE_COUNT: self.full_charge_count,
DataName.TOTAL_CHARGE_AMP_HOURS: self.total_charge_amp_hours,
DataName.TOTAL_DISCHARGE_AMP_HOURS: self.total_discharge_amp_hours,
DataName.TOTAL_PRODUCTION_ENERGY: self.total_production_energy,
DataName.TOTAL_CONSUMPTION_ENERGY: self.total_consumption_energy,
}