split store.py into a module with multiple small files

Author: yankee42

.flake8

@@ -0,0 +1,3 @@
+[flake8]
+per-file-ignores = */__init__.py:F401
+max-line-length = 120

.github/workflows/github-actions-python.yml

@@ -19,7 +19,6 @@ jobs:
         uses: TrueBrain/[email protected]
         with:
           path: packages
-          max_line_length: 120
       - name: Test with pytest
         run: |
           cd packages && python -m pytest

packages/modules/common/store.py

@@ -0,0 +1,5 @@
+from modules.common.store._api import ValueStore
+from modules.common.store._battery import get_bat_value_store
+from modules.common.store._car import get_car_value_store
+from modules.common.store._counter import get_counter_value_store
+from modules.common.store._inverter import get_inverter_value_store

packages/modules/common/store/__init__.py

@@ -0,0 +1,10 @@
+from abc import abstractmethod
+from typing import Generic, TypeVar
+
+T = TypeVar("T")
+
+
+class ValueStore(Generic[T]):
+    @abstractmethod
+    def set(self, state: T) -> None:
+        pass

packages/modules/common/store/_api.py

@@ -0,0 +1,43 @@
+from helpermodules import log, compatibility
+from modules.common.component_state import BatState
+from modules.common.store import ValueStore
+from modules.common.store._broker import pub_to_broker
+from modules.common.store._ramdisk import write_to_file
+from modules.common.store._util import process_error
+
+
+class BatteryValueStoreRamdisk(ValueStore[BatState]):
+    def __init__(self, component_num: int) -> None:
+        self.num = component_num
+
+    def set(self, bat_state: BatState):
+        try:
+            power = write_to_file("/speicherleistung", bat_state.power, 0)
+            write_to_file("/speichersoc", bat_state.soc, 0)
+            write_to_file("/speicherikwh", bat_state.imported, 2)
+            write_to_file("/speicherekwh", bat_state.exported, 2)
+            log.MainLogger().info('BAT Watt: ' + str(power))
+            log.MainLogger().info('BAT Einspeisung: ' + str(bat_state.exported))
+            log.MainLogger().info('BAT Bezug: ' + str(bat_state.imported))
+        except Exception as e:
+            process_error(e)
+
+
+class BatteryValueStoreBroker(ValueStore[BatState]):
+    def __init__(self, component_num: int) -> None:
+        self.num = component_num
+
+    def set(self, bat_state: BatState):
+        try:
+            pub_to_broker("openWB/set/bat/"+str(self.num)+"/get/power", bat_state.power, 2)
+            pub_to_broker("openWB/set/bat/"+str(self.num)+"/get/soc", bat_state.soc, 0)
+            pub_to_broker("openWB/set/bat/"+str(self.num)+"/get/imported", bat_state.imported, 2)
+            pub_to_broker("openWB/set/bat/"+str(self.num)+"/get/exported", bat_state.exported, 2)
+        except Exception as e:
+            process_error(e)
+
+
+def get_bat_value_store(component_num: int) -> ValueStore[BatState]:
+    if compatibility.is_ramdisk_in_use():
+        return BatteryValueStoreRamdisk(component_num)
+    return BatteryValueStoreBroker(component_num)

packages/modules/common/store/_battery.py

@@ -0,0 +1,15 @@
+from typing import Union
+
+from helpermodules.pub import Pub
+from modules.common.store._util import get_rounding_function_by_digits, process_error
+
+
+def pub_to_broker(topic: str, value, digits: Union[int, None] = None) -> None:
+    rounding = get_rounding_function_by_digits(digits)
+    try:
+        if isinstance(value, list):
+            Pub().pub(topic, [rounding(v) for v in value])
+        else:
+            Pub().pub(topic, rounding(value))
+    except Exception as e:
+        process_error(e)

packages/modules/common/store/_broker.py

@@ -0,0 +1,27 @@
+from helpermodules import compatibility
+from modules.common.component_state import CarState
+from modules.common.store import ValueStore
+from modules.common.store._broker import pub_to_broker
+from modules.common.store._ramdisk import write_to_file
+
+
+class CarValueStoreRamdisk(ValueStore[CarState]):
+    def __init__(self, charge_point: int):
+        self.filename = "soc" if charge_point == 1 else "soc1"
+
+    def set(self, state: CarState) -> None:
+        write_to_file(self.filename, state.soc, 0)
+
+
+class CarValueStoreBroker(ValueStore[CarState]):
+    def __init__(self, vehicle_id: int):
+        self.topic = "openWB/set/ev/{}/get/counter".format(vehicle_id)
+
+    def set(self, state: CarState) -> None:
+        pub_to_broker(self.topic, state.soc)
+
+
+def get_car_value_store(id: int) -> ValueStore[CarState]:
+    if compatibility.is_ramdisk_in_use():
+        return CarValueStoreRamdisk(id)
+    return CarValueStoreBroker(id)

packages/modules/common/store/_car.py

@@ -0,0 +1,51 @@
+from helpermodules import log, compatibility
+from modules.common.component_state import CounterState
+from modules.common.store import ValueStore
+from modules.common.store._broker import pub_to_broker
+from modules.common.store._ramdisk import write_array_to_files, write_to_file
+from modules.common.store._util import process_error
+
+
+class CounterValueStoreRamdisk(ValueStore[CounterState]):
+    def __init__(self, component_num: int) -> None:
+        self.num = component_num
+
+    def set(self, counter_state: CounterState):
+        try:
+            write_array_to_files("/evuv", counter_state.voltages, 1)
+            write_array_to_files("/bezuga", counter_state.currents, 1)
+            write_array_to_files("/bezugw", counter_state.powers, 0)
+            write_array_to_files("/evupf", counter_state.power_factors, 2)
+            imported = write_to_file("/bezugkwh", counter_state.imported)
+            exported = write_to_file("/einspeisungkwh", counter_state.exported)
+            power_all = write_to_file("/wattbezug", counter_state.power_all, 0)
+            write_to_file("/evuhz", counter_state.frequency, 2)
+            log.MainLogger().info('EVU Watt: ' + str(power_all))
+            log.MainLogger().info('EVU Bezug: ' + str(imported))
+            log.MainLogger().info('EVU Einspeisung: ' + str(exported))
+        except Exception as e:
+            process_error(e)
+
+
+class CounterValueStoreBroker(ValueStore[CounterState]):
+    def __init__(self, component_num: int) -> None:
+        self.num = component_num
+
+    def set(self, counter_state: CounterState):
+        try:
+            pub_to_broker("openWB/set/counter/" + str(self.num) + "/get/voltage", counter_state.voltages, 2)
+            pub_to_broker("openWB/set/counter/" + str(self.num) + "/get/current", counter_state.currents, 2)
+            pub_to_broker("openWB/set/counter/" + str(self.num) + "/get/power_phase", counter_state.powers, 2)
+            pub_to_broker("openWB/set/counter/" + str(self.num) + "/get/power_factors", counter_state.power_factors, 2)
+            pub_to_broker("openWB/set/counter/" + str(self.num) + "/get/imported", counter_state.imported)
+            pub_to_broker("openWB/set/counter/" + str(self.num) + "/get/exported", counter_state.exported)
+            pub_to_broker("openWB/set/counter/" + str(self.num) + "/get/power_all", counter_state.power_all)
+            pub_to_broker("openWB/set/counter/" + str(self.num) + "/get/frequency", counter_state.frequency)
+        except Exception as e:
+            process_error(e)
+
+
+def get_counter_value_store(component_num: int) -> ValueStore[CounterState]:
+    if compatibility.is_ramdisk_in_use():
+        return CounterValueStoreRamdisk(component_num)
+    return CounterValueStoreBroker(component_num)