dimm kit: improve error handling, handle deleted output devices

packages/control/io_device.py

@@ -1,7 +1,10 @@
 from dataclasses import dataclass, field
 from typing import Dict, 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
 
@@ -63,47 +66,63 @@
 
     def setup(self):
         for action in self.actions.values():
-            action.setup()
+            io_device = data.data.system_data[f"io{action.config.configuration.io_device}"]
+            with FaultStateContext(io_device.fault_state, update_always=False):
+                action.setup()
 
-    def dimming_get_import_power_left(self, device: Dict) -> Tuple[Optional[float], LoadmanagementLimit]:
+    def dimming_get_import_power_left(self,
+                                      device: Dict[str, Union[int, str]]
+                                      ) -> 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()
+            io_device = data.data.system_data[f"io{action.config.configuration.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]:
+    def dimming_set_import_power_left(self,
+                                      device: Dict[str, Union[int, str]], 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]:
+            io_device = data.data.system_data[f"io{action.config.configuration.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 in self.actions.values():
-            if isinstance(action, DimmingDirectControl):
-                for d in action.config.configuration.devices:
-                    if device == d:
-                        return action.dimming_via_direct_control()
+            io_device = data.data.system_data[f"io{action.config.configuration.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]:
+    def ripple_control_receiver(self, device: Dict[str, Union[int, str]]) -> 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()
+            io_device = data.data.system_data[f"io{action.config.configuration.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()
+            io_device = data.data.system_data[f"io{action.config.configuration.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

@@ -13,6 +13,7 @@
 from helpermodules.utils._thread_handler import joined_thread_handler
 from modules.common.abstract_io import AbstractIoDevice
 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
@@ -72,30 +73,36 @@ def process_algorithm_results(self) -> None:
                                   data.data.bat_data[f"bat{bat_component.component_config.id}"].data.set.power_limit),
                             name=f"set power limit {bat_component.component_config.id}"))
             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]
+                io_device = data.data.system_data[f"io{action.config.configuration.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:
+                        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)
+            return message, FaultStateLevel.ERROR
         if isinstance(entry.handler, str):
             return entry.handler, FaultStateLevel.ERROR
         result = entry.handler(exception)

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,31 @@ 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 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()
+        if self.reraise is False or exc_value is None:
+            return True
+        else:
+            return False
+
+
 T_C = TypeVar("T_C", bound=Callable)

packages/modules/io_actions/common.py

@@ -1,6 +1,29 @@
+import logging
+from typing import Dict, Union
+
 from control import data
 from modules.common.fault_state_level import FaultStateLevel
 
 
+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"] == "io":
+            io = f"io{device['id']}"
+            return (f"{data.data.system_data[io].config.name}: "
+                    "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 "Unbekanntes Gerät, "

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_device_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
@@ -64,14 +64,7 @@ def setup(self) -> None:
                 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")
+                    msg += get_device_log_message(device)
                 control_command_log.info(msg)
             elif self.timestamp:
                 Pub().pub(f"openWB/set/io/action/{self.config.id}/timestamp", None)

packages/modules/io_actions/controllable_consumers/dimming/api_io.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_device_log_message
 from modules.io_actions.controllable_consumers.dimming.config import DimmingSetup
 from modules.io_actions.controllable_consumers.dimming.utils import calc_dimming_surplus
 
@@ -62,14 +62,7 @@ def setup(self) -> None:
                 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")
+                    msg += get_device_log_message(device)
                 control_command_log.info(msg)
             elif self.timestamp:
                 Pub().pub(f"openWB/set/io/action/{self.config.id}/timestamp", None)

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

@@ -8,7 +8,7 @@
 from modules.common.abstract_device import DeviceDescriptor
 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_device_log_message
 from modules.io_actions.controllable_consumers.dimming_direct_control.config import DimmingDirectControlSetup
 
 control_command_log = logging.getLogger("steuve_control_command")
@@ -51,13 +51,7 @@ def setup(self) -> None:
 
                 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"
-                if device["type"] == "cp":
-                    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", IO-Gerät {data.data.system_data[io].config.name}: "
-                            "Leistung unbekannt")
+                msg += get_device_log_message(device)
                 control_command_log.info(msg)
             elif self.timestamp:
                 Pub().pub(f"openWB/set/io/action/{self.config.id}/timestamp", None)

packages/modules/io_actions/controllable_consumers/ripple_control_receiver/api.py

@@ -8,7 +8,7 @@
 from modules.common.abstract_device import DeviceDescriptor
 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_device_log_message
 from modules.io_actions.controllable_consumers.ripple_control_receiver.config import RippleControlReceiverSetup
 
 control_command_log = logging.getLogger("steuve_control_command")
@@ -33,14 +33,7 @@ def log_active_ripple_control_receiver():
             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", IO-Gerät {data.data.io_data[io].data.config.name}: "
-                            "Leistung unbekannt")
+                msg += get_device_log_message(device)
             control_command_log.info(msg)
 
         with ModifyLoglevelContext(control_command_log, logging.DEBUG):