dimm kit: improve error handling, handle deleted output devices

packages/control/io_device.py

@@ -1,7 +1,11 @@
 from dataclasses import dataclass, field
-from typing import Dict, Optional, Tuple, Union
+import logging
+from typing import Dict, Iterator, Optional, Tuple, Union
+from control import data
 from control.limiting_value import LoadmanagementLimit
 from helpermodules.constants import NO_ERROR
+from modules.common.configurable_io import ConfigurableIo
+from modules.common.fault_state import FaultStateContext
 from modules.io_actions.controllable_consumers.dimming.api_eebus import DimmingEebus
 from modules.io_actions.controllable_consumers.dimming.api_io import DimmingIo
 
@@ -12,6 +16,18 @@
 from modules.io_actions.generator_systems.stepwise_control.api_eebus import StepwiseControlEebus
 from modules.io_actions.generator_systems.stepwise_control.api_io import StepwiseControlIo
 
+log = logging.getLogger(__name__)
+
+IoAction = Union[
+    DimmingIo,
+    DimmingEebus,
+    DimmingDirectControl,
+    RippleControlReceiver,
+    StepwiseControlEebus,
+    StepwiseControlIo,
+    LoadManager,
+]
+
 
 @dataclass
 class Get:
@@ -57,53 +73,69 @@ def __init__(self, num: Union[int, str]):
 
 class IoActions:
     def __init__(self):
-        self.actions: Dict[int, Union[DimmingIo, DimmingEebus, DimmingDirectControl,
-                                      RippleControlReceiver, StepwiseControlEebus, StepwiseControlIo,
-                                      LoadManager]] = {}
+        self.actions: Dict[int, IoAction] = {}
 
-    def setup(self):
+    def iter_actions_with_io_device(self) -> Iterator[Tuple[IoAction, ConfigurableIo]]:
         for action in self.actions.values():
-            action.setup()
+            io_device = data.data.system_data.get(f"io{action.config.configuration.io_device}")
+            if not isinstance(io_device, ConfigurableIo):
+                log.error(f"I/O device io{action.config.configuration.io_device} not found.")
+                continue
+            yield action, io_device
 
-    def dimming_get_import_power_left(self, device: Dict) -> Tuple[Optional[float], LoadmanagementLimit]:
-        for action in self.actions.values():
-            if isinstance(action, (DimmingIo, DimmingEebus)):
-                for d in action.config.configuration.devices:
-                    if device == d:
-                        return action.dimming_get_import_power_left()
+    def setup(self):
+        for action, io_device in self.iter_actions_with_io_device():
+            with FaultStateContext(io_device.fault_state, update_always=False):
+                action.setup()
+
+    def dimming_get_import_power_left(self,
+                                      device: Dict[str, Union[int, str]]
+                                      ) -> Tuple[Optional[float], LoadmanagementLimit]:
+        for action, io_device in self.iter_actions_with_io_device():
+            with FaultStateContext(io_device.fault_state, update_always=False):
+                if isinstance(action, (DimmingIo, DimmingEebus)):
+                    for d in action.config.configuration.devices:
+                        if device == d:
+                            return action.dimming_get_import_power_left()
         else:
             return None, LoadmanagementLimit(None, None)
 
-    def dimming_set_import_power_left(self, device: Dict, used_power: float) -> Optional[float]:
-        for action in self.actions.values():
-            if isinstance(action, (DimmingIo, DimmingEebus)):
-                for d in action.config.configuration.devices:
-                    if d == device:
-                        return action.dimming_set_import_power_left(used_power)
-
-    def dimming_via_direct_control(self, device: Dict) -> Tuple[Optional[float], LoadmanagementLimit]:
-        for action in self.actions.values():
-            if isinstance(action, DimmingDirectControl):
-                for d in action.config.configuration.devices:
-                    if device == d:
-                        return action.dimming_via_direct_control()
+    def dimming_set_import_power_left(self,
+                                      device: Dict[str, Union[int, str]], used_power: float) -> Optional[float]:
+        for action, io_device in self.iter_actions_with_io_device():
+            with FaultStateContext(io_device.fault_state, update_always=False):
+                if isinstance(action, (DimmingIo, DimmingEebus)):
+                    for d in action.config.configuration.devices:
+                        if d == device:
+                            return action.dimming_set_import_power_left(used_power)
+
+    def dimming_via_direct_control(self,
+                                   device: Dict[str, Union[int, str]]) -> Tuple[Optional[float], LoadmanagementLimit]:
+        for action, io_device in self.iter_actions_with_io_device():
+            with FaultStateContext(io_device.fault_state, update_always=False):
+                if isinstance(action, DimmingDirectControl):
+                    for d in action.config.configuration.devices:
+                        if device == d:
+                            return action.dimming_via_direct_control()
         else:
             return None, LoadmanagementLimit(None, None)
 
-    def ripple_control_receiver(self, device: Dict) -> Tuple[float, LoadmanagementLimit]:
-        for action in self.actions.values():
-            if isinstance(action, RippleControlReceiver):
-                for d in action.config.configuration.devices:
-                    if device == d:
-                        return action.ripple_control_receiver()
+    def ripple_control_receiver(self, device: Dict[str, Union[int, str]]) -> Tuple[float, LoadmanagementLimit]:
+        for action, io_device in self.iter_actions_with_io_device():
+            with FaultStateContext(io_device.fault_state, update_always=False):
+                if isinstance(action, RippleControlReceiver):
+                    for d in action.config.configuration.devices:
+                        if device == d:
+                            return action.ripple_control_receiver()
         else:
             return 1, LoadmanagementLimit(None, None)
 
     def stepwise_control(self, device_id: int) -> Tuple[Optional[float], LoadmanagementLimit]:
-        for action in self.actions.values():
-            if isinstance(action, (StepwiseControlEebus, StepwiseControlIo)):
-                if device_id in [component["id"] for component in action.config.configuration.devices]:
-                    return action.control_stepwise()
+        for action, io_device in self.iter_actions_with_io_device():
+            with FaultStateContext(io_device.fault_state, update_always=False):
+                if isinstance(action, (StepwiseControlEebus, StepwiseControlIo)):
+                    if device_id in [component["id"] for component in action.config.configuration.devices]:
+                        return action.control_stepwise()
         else:
             return None, LoadmanagementLimit(None, None)
 

packages/control/process.py

@@ -19,13 +19,15 @@
 from modules.common.abstract_io import AbstractIoDevice
 from modules.common.configurable_consumer import SetLimitData
 from modules.common.configurable_device import set_power_limit_wrapper
+from modules.common.fault_state import FaultStateContext
 from modules.common.fault_state_level import FaultStateLevel
 from modules.io_actions.controllable_consumers.dimming.api_io import DimmingIo
 from modules.io_actions.controllable_consumers.dimming_direct_control.api import DimmingDirectControl
 from modules.io_actions.generator_systems.stepwise_control.api_eebus import StepwiseControlEebus
 from modules.io_actions.generator_systems.stepwise_control.api_io import StepwiseControlIo
 
 log = logging.getLogger(__name__)
+control_command_log = logging.getLogger("steuve_control_command")
 
 
 class Process:
@@ -96,31 +98,38 @@ def process_algorithm_results(self) -> None:
                         modules_threads.append(consumer_thread)
                 except Exception:
                     log.exception("Fehler im Process-Modul für Verbaucher "+str(consumer))
-            for action in data.data.io_actions.actions.values():
-                if isinstance(action, DimmingDirectControl):
-                    for d in action.config.configuration.devices:
-                        if d["type"] == "io":
-                            data.data.io_states[f"io_states{d['id']}"].data.set.digital_output[d["digital_output"]] = (
-                                action.dimming_via_direct_control()[0] is None  # active output (True) if no dimming
-                            )
-                if isinstance(action, DimmingIo):
-                    for d in action.config.configuration.devices:
-                        if d["type"] == "io":
-                            data.data.io_states[f"io_states{d['id']}"].data.set.digital_output[d["digital_output"]] = (
-                                not action.dimming_active()  # active output (True) if no dimming
-                            )
-                if isinstance(action, (StepwiseControlEebus, StepwiseControlIo)):
-                    # check if passthrough is enabled
-                    if (action.config.configuration.passthrough_enabled and
-                            action.config.configuration.io_output_device is not None):
-                        # find output pattern by value
-                        for pattern in action.config.configuration.output_pattern:
-                            if pattern["value"] == action.control_stepwise()[0]:
-                                # set digital outputs according to matching output_pattern
-                                for output in pattern["matrix"].keys():
-                                    data.data.io_states[
-                                        f"io_states{action.config.configuration.io_output_device}"
-                                    ].data.set.digital_output[output] = pattern["matrix"][output]
+            for action, io_device in data.data.io_actions.iter_actions_with_io_device():
+                with FaultStateContext(io_device.fault_state, update_always=False):
+                    try:
+                        if isinstance(action, DimmingDirectControl):
+                            for d in action.config.configuration.devices:
+                                if d["type"] == "io":
+                                    digital_output = data.data.io_states[f"io_states{d['id']}"].data.set.digital_output
+                                    # active output (True) if no dimming
+                                    digital_output[d["digital_output"]] = action.dimming_via_direct_control()[0] is None
+                        if isinstance(action, DimmingIo):
+                            for d in action.config.configuration.devices:
+                                if d["type"] == "io":
+                                    digital_output = data.data.io_states[f"io_states{d['id']}"].data.set.digital_output
+                                    # active output (True) if no dimming
+                                    digital_output[d["digital_output"]] = not action.dimming_active()
+                        if isinstance(action, (StepwiseControlEebus, StepwiseControlIo)):
+                            # check if passthrough is enabled
+                            if (action.config.configuration.passthrough_enabled and
+                                    action.config.configuration.io_output_device is not None):
+                                # find output pattern by value
+                                for pattern in action.config.configuration.output_pattern:
+                                    if pattern["value"] == action.control_stepwise()[0]:
+                                        # set digital outputs according to matching output_pattern
+                                        for output in pattern["matrix"].keys():
+                                            data.data.io_states[
+                                                f"io_states{action.config.configuration.io_output_device}"
+                                            ].data.set.digital_output[output] = pattern["matrix"][output]
+                    except KeyError as e:
+                        control_command_log.error(
+                            f"Ausgang konnte für die Aktion {action.config.name} nicht zugeordnet werden.")
+                        raise KeyError(f"Ausgang konnte für die Aktion {action.config.name} nicht zugeordnet werden. "
+                                       "Bitte prüfen Sie die Konfiguration.") from e
             for io in data.data.system_data.values():
                 if isinstance(io, AbstractIoDevice):
                     modules_threads.append(

packages/helpermodules/exceptions/registry.py

@@ -25,7 +25,8 @@ class ExceptionRegistry:
     def translate_exception(self, exception: Exception) -> Tuple[str, FaultStateLevel]:
         entry = self.find_registry_entry(exception)
         if entry is None:
-            return str(exception.args[0]), FaultStateLevel.ERROR
+            message = (str(exception.args[0]) if exception.args else str(exception)) or exception.__class__.__name__
+            return message, FaultStateLevel.ERROR
         if isinstance(entry.handler, str):
             return entry.handler, FaultStateLevel.ERROR
         result = entry.handler(exception)

packages/helpermodules/exceptions/registry_test.py

@@ -31,6 +31,11 @@ class ErrorF(ErrorD):
     pass
 
 
+class EmptyStringException(Exception):
+    def __str__(self):
+        return ""
+
+
 @pytest.mark.parametrize("exception,expected_message", [
     [ErrorRoot, "ErrorRoot"],
     [ErrorB, "B"],
@@ -68,3 +73,21 @@ def test_accepts_all_supported_formats(handler):
     # evaluation
     assert isinstance(actual[0], str)
     assert actual[0] == "msg"
+
+
+@pytest.mark.parametrize("exception,expected_message", [
+    pytest.param(Exception(), "Exception", id="no args fallback"),
+    pytest.param(EmptyStringException(), "EmptyStringException", id="empty string fallback"),
+])
+def test_translate_exception_falls_back_to_class_name_for_empty_message(exception: Exception,
+                                                                        expected_message: str,
+                                                                        monkeypatch):
+    # setup
+    monkeypatch.setattr(ExceptionRegistry, "registry", [])
+    registry = ExceptionRegistry()
+
+    # execution
+    actual = registry.translate_exception(exception)
+
+    # evaluation
+    assert actual[0] == expected_message

packages/modules/common/fault_state.py

@@ -1,6 +1,7 @@
 import logging
 import traceback
-from typing import Optional, Callable, TypeVar
+from types import TracebackType
+from typing import Optional, Callable, Type, TypeVar
 
 from helpermodules import exceptions
 from helpermodules.pub import Pub
@@ -87,4 +88,30 @@ def from_exception(self, exception: Optional[Exception] = None) -> None:
                 exception)
 
 
+class FaultStateContext:
+    def __init__(self, fault_state: FaultState, update_always: bool = True, reraise: bool = False) -> None:
+        self.__fault_state = fault_state
+        self.update_always = update_always
+        self.reraise = reraise
+
+    def __enter__(self) -> None:
+        if self.update_always:
+            self.__fault_state.no_error()
+        return None
+
+    def __exit__(self,
+                 exc_type: Optional[Type[BaseException]],
+                 exc_value: Optional[BaseException],
+                 traceback: Optional[TracebackType]) -> bool:
+        if exc_value is not None and not isinstance(exc_value, Exception):
+            return False
+        if isinstance(exc_value, Exception):
+            self.__fault_state.from_exception(exc_value)
+        elif self.update_always is False and self.__fault_state.fault_state == 0:
+            # Fehlerstatus nicht überschreiben
+            return True
+        self.__fault_state.store_error()
+        return self.reraise is False or exc_value is None
+
+
 T_C = TypeVar("T_C", bound=Callable)

packages/modules/io_actions/common.py

@@ -1,6 +1,42 @@
+import logging
+from typing import Dict, List, Union
+
 from control import data
 from modules.common.fault_state_level import FaultStateLevel
+from modules.common.utils.component_parser import get_component_name_by_id
+
+
+control_command_log = logging.getLogger("steuve_control_command")
 
 
 def check_fault_state_io_device(io_device: int) -> bool:
     return data.data.io_states[f"io_states{io_device}"].data.get.fault_state == FaultStateLevel.ERROR
+
+
+def get_device_log_message(device: Dict[str, Union[int, str]]) -> str:
+    try:
+        if device["type"] == "cp":
+            cp = f"cp{device['id']}"
+            return (f", Ladepunkt {data.data.cp_data[cp].data.config.name}: "
+                    f"{data.data.cp_data[cp].data.get.powers}W")
+        if device["type"] == "inverter":
+            inverter = f"pv{device['id']}"
+            return (f", Erzeugungsanlage {get_component_name_by_id(device['id'])}: "
+                    f"{data.data.pv_data[inverter].data.get.power}W")
+        if device["type"] == "io":
+            io = f"io{device['id']}"
+            return (f", {data.data.system_data[io].config.name} {device['digital_output']}: "
+                    "Leistung unbekannt")
+    except KeyError:
+        control_command_log.warning(f"Zugriff auf gelöschtes Gerät nicht möglich: {device}")
+    except Exception:
+        control_command_log.exception(f"Fehler beim Zugriff auf Gerät {device}")
+    return f", Unbekanntes Gerät: {device}"
+
+
+def get_power_log_message(devices: List[Dict[str, Union[int, str]]]) -> str:
+    evu_counter = data.data.counter_data[data.data.counter_all_data.get_evu_counter_str()]
+    msg = f"EVU-Zähler: {evu_counter.data.get.powers}W, {evu_counter.data.get.power}W"
+    for device in devices:
+        msg += get_device_log_message(device)
+    return msg

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

@@ -9,7 +9,7 @@
 from dataclass_utils import asdict
 from modules.common.abstract_io import AbstractIoAction
 from modules.common.utils.component_parser import get_io_name_by_id
-from modules.io_actions.common import check_fault_state_io_device
+from modules.io_actions.common import check_fault_state_io_device, get_power_log_message
 from modules.io_actions.controllable_consumers.dimming.config import DimmingSetup
 from modules.io_actions.controllable_consumers.dimming.utils import calc_dimming_surplus
 from modules.io_devices.eebus.config import AnalogInputMapping, DigitalInputMapping
@@ -61,18 +61,7 @@ def setup(self) -> None:
                         control_command_log.info(f"Dimmen aktiviert. Übermittelter LPC-Wert: {lpc_value/1000}kWh. "
                                                  "Leistungswerte vor Ausführung des Steuerbefehls:")
 
-                evu_counter = data.data.counter_data[data.data.counter_all_data.get_evu_counter_str()]
-                msg = f"EVU-Zähler: {evu_counter.data.get.powers}W, {evu_counter.data.get.power}W"
-                for device in self.config.configuration.devices:
-                    if device["type"] == "cp":
-                        cp = f"cp{device['id']}"
-                        msg += (f", Ladepunkt {data.data.cp_data[cp].data.config.name}: "
-                                f"{data.data.cp_data[cp].data.get.powers}W")
-                    if device["type"] == "io":
-                        io = f"io{device['id']}"
-                        msg += (f", {data.data.system_data[io].config.name}: "
-                                "Leistung unbekannt")
-                control_command_log.info(msg)
+                control_command_log.info(get_power_log_message(self.config.configuration.devices))
             elif self.timestamp:
                 Pub().pub(f"openWB/set/io/action/{self.config.id}/timestamp", None)
                 control_command_log.info("Dimmen deaktiviert.")