rewrite configuration

Author: benderl

packages/control/io_device.py

@@ -101,10 +101,10 @@ def ripple_control_receiver(self, device: Dict) -> float:
         else:
             return 1
 
-    def stepwise_control(self, device_id: int) -> Optional[str]:
+    def stepwise_control(self, device_id: int) -> Optional[float]:
         for action in self.actions.values():
             if isinstance(action, StepwiseControl):
-                if device_id == action.config.configuration.pv_id:
+                if device_id in [component["id"] for component in action.config.configuration.devices]:
                     self._check_fault_state_io_device(action.config.configuration.io_device)
                     return action.control_stepwise()
         else:

packages/control/pv_all.py

@@ -68,8 +68,13 @@ def calc_power_for_all_components(self) -> None:
                             else:
                                 if fault_state < module_data.get.fault_state:
                                     fault_state = module_data.get.fault_state
-                            msg = data.data.io_actions.stepwise_control(data.data.pv_data[module].num)
-                            if msg is not None and data.data.pv_data[module].data.get.fault_state == 0:
+                            limit_value = data.data.io_actions.stepwise_control(data.data.pv_data[module].num)
+                            if limit_value is not None and data.data.pv_data[module].data.get.fault_state == 0:
+                                msg = (
+                                    f"Leistung begrenzt auf {int(limit_value * 100)}%"
+                                    if limit_value < 1
+                                    else "Keine Leistungsbegrenzung aktiv."
+                                )
                                 data.data.pv_data[module].data.get.fault_str = msg
                                 Pub().pub(f"openWB/set/pv/{data.data.pv_data[module].num}/get/fault_str", msg)
                     except Exception:

packages/dataclass_utils/factories.py

@@ -17,7 +17,7 @@ def voltages_list_factory() -> List[float]:
     return [230.0]*3
 
 
-def empty_io_pattern_factory():
+def empty_io_pattern_boolean_factory():
     return [
         {
             "value": True,  # dimmen
@@ -28,3 +28,20 @@ def empty_io_pattern_factory():
             "input_matrix": {}
         }
     ]
+
+
+def empty_io_pattern_stepwise_factory():
+    return [
+        {
+            "value": 0.6,  # Stufe 1
+            "input_matrix": {}
+        },
+        {
+            "value": 0.3,  # Stufe 2
+            "input_matrix": {}
+        },
+        {
+            "value": 0,  # Stufe 3
+            "input_matrix": {}
+        }
+    ]

packages/modules/io_actions/controllable_consumers/dimming/config.py

@@ -1,13 +1,13 @@
 from dataclasses import dataclass, field
 from typing import Dict, List, Optional
-from dataclass_utils.factories import empty_io_pattern_factory, empty_list_factory
+from dataclass_utils.factories import empty_io_pattern_boolean_factory, empty_list_factory
 from modules.io_actions.groups import ActionGroup
 
 
 @dataclass
 class DimmingConfig:
     io_device: Optional[int] = None
-    input_pattern: List[Dict] = field(default_factory=empty_io_pattern_factory)
+    input_pattern: List[Dict] = field(default_factory=empty_io_pattern_boolean_factory)
     devices: List[Dict] = field(default_factory=empty_list_factory)
     # [{"type": "cp", "id": 0},
     # {"type": "io", "id": 1, "digital_output": "SofortLa"}]

packages/modules/io_actions/controllable_consumers/dimming_direct_control/config.py

@@ -1,13 +1,13 @@
 from dataclasses import dataclass, field
 from typing import Dict, List, Optional
-from dataclass_utils.factories import empty_io_pattern_factory, empty_list_factory
+from dataclass_utils.factories import empty_io_pattern_boolean_factory, empty_list_factory
 from modules.io_actions.groups import ActionGroup
 
 
 @dataclass
 class DimmingDirectControlConfig:
     io_device: Optional[int] = None
-    input_pattern: List[Dict] = field(default_factory=empty_io_pattern_factory)
+    input_pattern: List[Dict] = field(default_factory=empty_io_pattern_boolean_factory)
     devices: List[Dict] = field(default_factory=empty_list_factory)
     # [{"type": "cp", "id": 0},
     # {"type": "io", "id": 1, "digital_output": "SofortLa"}]

packages/modules/io_actions/generator_systems/stepwise_control/api.py

@@ -1,10 +1,11 @@
 import logging
-from control import data
 from typing import Optional
+from control import data
 from helpermodules.logger import ModifyLoglevelContext
+from helpermodules.pub import Pub
+from helpermodules.timecheck import create_timestamp
 from modules.common.abstract_device import DeviceDescriptor
 from modules.common.abstract_io import AbstractIoAction
-from modules.common.utils.component_parser import get_component_name_by_id
 from modules.io_actions.generator_systems.stepwise_control.config import StepwiseControlSetup
 
 control_command_log = logging.getLogger("steuve_control_command")
@@ -13,57 +14,62 @@
 class StepwiseControl(AbstractIoAction):
     def __init__(self, config: StepwiseControlSetup):
         self.config = config
-        control_command_log.info(f"Stufenweise Steuerung einer EZA: Eingang {self.config.configuration.s1} für S1, "
-                                 f"Eingang {self.config.configuration.s2} für S2, und Eingang "
-                                 f"{self.config.configuration.w3} für W3 wird überwacht. Die Beschränkung muss in "
-                                 "der EZA vorgenommen werden.")
+        self.__unique_inputs = []
+        for pattern in self.config.configuration.input_pattern:
+            for key in pattern["input_matrix"].keys():
+                if key not in self.__unique_inputs:
+                    self.__unique_inputs.append(key)
+        assigned_devices = [io_device["id"] for io_device in self.config.configuration.devices]
+        control_command_log.info(
+            f"Stufenweise Steuerung von EZA: I/O-Gerät: {self.config.configuration.io_device}, "
+            f"Überwachte digitale Eingänge: {self.__unique_inputs}, "
+            f"zugeordnete Erzeugungsanlagen: {assigned_devices} "
+            "Die Begrenzung muss in den EZA vorgenommen werden!"
+        )
         super().__init__()
 
     def setup(self) -> None:
-        pass
-
-    def control_stepwise(self) -> Optional[str]:
-        text = (f"Die Einspeiseleistung von {get_component_name_by_id(self.config.configuration.pv_id)} ist auf "
-                "{} % beschränkt. Die Beschränkung muss in der EZA vorgenommen werden.")
-        msg = None
-        digital_input = data.data.io_states[f"io_states{self.config.configuration.io_device}"].data.get.digital_input
-        digital_input_prev = data.data.io_states[
-            f"io_states{self.config.configuration.io_device}"].data.get.digital_input_prev
+        with ModifyLoglevelContext(control_command_log, logging.DEBUG):
+            digital_input = (
+                data.data.io_states[
+                    f"io_states{self.config.configuration.io_device}"
+                ].data.get.digital_input
+            )
+            digital_input_prev = data.data.io_states[
+                f"io_states{self.config.configuration.io_device}"].data.get.digital_input_prev
+            changed = len([
+                input_name for input_name in self.__unique_inputs
+                if digital_input[input_name] != digital_input_prev[input_name]
+            ]) > 0
 
-        active_inputs = [
-            digital_input[self.config.configuration.s1],
-            digital_input[self.config.configuration.s2],
-            digital_input[self.config.configuration.w3]
-        ]
-        num_active = sum(1 for v in active_inputs if v)
+            for pattern in self.config.configuration.input_pattern:
+                for action_input, value in pattern["input_matrix"].items():
+                    if digital_input[action_input] != value:
+                        break
+                else:
+                    # Alle digitalen Eingänge entsprechen dem Pattern
+                    if pattern["value"] != 1:
+                        if changed:
+                            Pub().pub(f"openWB/set/io/action/{self.config.id}/timestamp", create_timestamp())
+                            control_command_log.info(f"EZA-Begrenzung mit Wert {int(pattern['value']*100)}% aktiviert.")
+                        break
+            else:
+                if changed:
+                    Pub().pub(f"openWB/set/io/action/{self.config.id}/timestamp", None)
+                    control_command_log.info("EZA-Begrenzung aufgehoben.")
 
-        if num_active > 1:
-            error_msg = (f"Fehler: Mehr als ein Eingang ist aktiv für die stufenweise Steuerung der EZA! "
-                         f"S1: {digital_input[self.config.configuration.s1]}, "
-                         f"S2: {digital_input[self.config.configuration.s2]}, "
-                         f"W3: {digital_input[self.config.configuration.w3]}")
-            with ModifyLoglevelContext(control_command_log, logging.ERROR):
-                control_command_log.error(error_msg)
-            raise ValueError(error_msg)
-
-        if digital_input[self.config.configuration.s1]:
-            msg = text.format(60)
-        elif digital_input[self.config.configuration.s2]:
-            msg = text.format(30)
-        elif digital_input[self.config.configuration.w3]:
-            msg = text.format(0)
+    def control_stepwise(self) -> Optional[float]:
+        for pattern in self.config.configuration.input_pattern:
+            for digital_input, value in pattern["input_matrix"].items():
+                if data.data.io_states[f"io_states{self.config.configuration.io_device}"
+                                       ].data.get.digital_input[digital_input] != value:
+                    break
+            else:
+                # Alle digitalen Eingänge entsprechen dem Pattern
+                return pattern['value']
         else:
-            # Keine Beschränkung soll nicht dauerhaft im WR angezeigt werden.
-            msg = (f"Die Einspeiseleistung von {get_component_name_by_id(self.config.configuration.pv_id)} ist "
-                   "nicht beschränkt. Die Beschränkung muss in der EZA vorgenommen werden.")
-
-        if not (digital_input[self.config.configuration.s1] == digital_input_prev[self.config.configuration.s1] and
-                digital_input[self.config.configuration.s2] == digital_input_prev[self.config.configuration.s2] and
-                digital_input[self.config.configuration.w3] == digital_input_prev[self.config.configuration.w3]):
-            # Wenn sich was geändert hat, loggen
-            with ModifyLoglevelContext(control_command_log, logging.DEBUG):
-                control_command_log.info(msg)
-        return msg
+            # Zustand entspricht keinem Pattern, Leistungsbegrenzung aufheben
+            return 1
 
 
 def create_action(config: StepwiseControlSetup):

packages/modules/io_actions/generator_systems/stepwise_control/config.py

@@ -1,20 +1,20 @@
-from dataclasses import dataclass
-from typing import Optional
+from dataclasses import dataclass, field
+from typing import Dict, List, Optional
+from dataclass_utils.factories import empty_list_factory, empty_io_pattern_stepwise_factory
 from modules.io_actions.groups import ActionGroup
 
 
 @dataclass
 class StepwiseControlConfig:
     io_device: Optional[int] = None
-    s1: str = None
-    s2: str = None
-    w3: str = None
-    pv_id: int = None
+    input_pattern: List[Dict] = field(default_factory=empty_io_pattern_stepwise_factory)
+    devices: List[Dict] = field(default_factory=empty_list_factory)
+    # [{"type": "inverter", "id": 1},...]
 
 
 class StepwiseControlSetup:
     def __init__(self,
-                 name: str = "Stufenweise Steuerung einer EZA",
+                 name: str = "Stufenweise Steuerung von EZA",
                  type: str = "stepwise_control",
                  id: int = 0,
                  configuration: StepwiseControlConfig = None):