se automatic phase switch, enum handling with dataclass <->dict conversion

packages/control/algorithm/additional_current.py

@@ -30,7 +30,8 @@ def set_additional_current(self) -> None:
                     available_currents, limit = Loadmanagement().get_available_currents(missing_currents, counter, cp)
                     log.debug(f"cp {cp.num} available currents {available_currents} missing currents "
                               f"{missing_currents} limit {limit.message}")
-                    cp.data.control_parameter.limit = limit
+                    if limit.limiting_value is not None:
+                        cp.data.control_parameter.limit = limit
                     available_for_cp = common.available_current_for_cp(cp, counts, available_currents, missing_currents)
                     current = common.get_current_to_set(
                         cp.data.set.current, available_for_cp, cp.data.set.target_current)

packages/control/algorithm/algorithm.py

@@ -29,6 +29,8 @@ def calc_current(self) -> None:
             self._check_auto_phase_switch_delay()
             self.surplus_controlled.check_submode_pv_charging()
             common.reset_current()
+            for cp in data.data.cp_data.values():
+                cp.reset_values_before_algorithm()
             log.info("**Mindestrom setzen**")
             self.min_current.set_min_current()
             log.info("**Soll-Strom setzen**")

packages/control/algorithm/min_current.py

@@ -27,7 +27,8 @@ def set_min_current(self) -> None:
                     if max(missing_currents) > 0:
                         available_currents, limit = Loadmanagement().get_available_currents(
                             missing_currents, counter, cp)
-                        cp.data.control_parameter.limit = limit
+                        if limit.limiting_value is not None:
+                            cp.data.control_parameter.limit = limit
                         available_for_cp = common.available_current_for_cp(
                             cp, counts, available_currents, missing_currents)
                         current = common.get_current_to_set(

packages/control/algorithm/surplus_controlled.py

@@ -61,7 +61,9 @@ def _set(self,
                 cp,
                 feed_in=feed_in_yield
             )
-            cp.data.control_parameter.limit = limit
+            # im PV-Laden wird der Strom immer durch die Leistung begrenzt
+            if limit.limiting_value is not None and limit.limiting_value != LimitingValue.POWER:
+                cp.data.control_parameter.limit = limit
             available_for_cp = common.available_current_for_cp(cp, counts, available_currents, missing_currents)
             if counter.get_control_range_state(feed_in_yield) == ControlRangeState.MIDDLE:
                 pv_charging = data.data.general_data.data.chargemode_config.pv_charging

packages/control/auto_phase_switch_test.py

@@ -5,7 +5,7 @@
 from control.chargepoint.control_parameter import ControlParameter
 from control.counter import Counter, CounterData, Set
 
-from control.limiting_value import LoadmanagementLimit
+from control.limiting_value import LimitingValue, LoadmanagementLimit
 from control.ev.charge_template import ChargeTemplate
 from control.pv_all import PvAll
 from control.bat_all import BatAll
@@ -153,21 +153,32 @@ def test_auto_phase_switch(monkeypatch, vehicle: Ev, params: Params):
 
 
 @pytest.mark.parametrize(
-    "evse_current, get_currents, all_surplus, expected",
+    "evse_current, get_currents, all_surplus, limit, expected",
     [
-        pytest.param(8, [7.7]*3, 100, (False, Ev.ENOUGH_POWER), id="kein 1p3p, genug Leistung für mehrphasige Ladung"),
-        pytest.param(10, [9.8, 0, 0], 50, (False, Ev.NOT_ENOUGH_POWER),
+        pytest.param(8, [7.7]*3, 100, LoadmanagementLimit(None, None), (False, Ev.ENOUGH_POWER),
+                     id="kein 1p3p, genug Leistung für mehrphasige Ladung"),
+        pytest.param(10, [9.8, 0, 0], 50, LoadmanagementLimit(None, None), (False, Ev.NOT_ENOUGH_POWER),
                      id="kein 1p3p, nicht genug Leistung, um auf 3p zu schalten"),
-        pytest.param(16, [14, 0, 0], 5000, (False, Ev.CURRENT_OUT_OF_NOMINAL_DIFFERENCE),
+        pytest.param(16, [14, 0, 0], 5000, LoadmanagementLimit(None, None),
+                     (False, Ev.CURRENT_OUT_OF_NOMINAL_DIFFERENCE),
                      id="kein 1p3p, Auto lädt nicht mit vorgegebener Maximalstromstärke"),
-        pytest.param(6, [7.5]*3, -20, (False, Ev.CURRENT_OUT_OF_NOMINAL_DIFFERENCE),
+        pytest.param(6, [7.5]*3, -20, LoadmanagementLimit(None, None), (False, Ev.CURRENT_OUT_OF_NOMINAL_DIFFERENCE),
                      id="kein 1p3p, Auto lädt nicht mit vorgegebener Minimalstromstärke"),
-        pytest.param(16, [15.8, 0, 0], 5000, (True, None), id="1p3p"),
-        pytest.param(6, [5.8]*3, -10, (True, None), id="3p1p"),
+        pytest.param(16, [15.8, 0, 0], 5000, LoadmanagementLimit(None, None), (True, None), id="1p3p"),
+        pytest.param(6, [5.8]*3, -10, LoadmanagementLimit(None, None), (True, None), id="3p1p"),
+        pytest.param(10, [9.8, 0, 0], 5000,
+                     LoadmanagementLimit(message=", da der Maximal-Strom an Zähler Test erreicht ist.",
+                                         limiting_value=LimitingValue.CURRENT), (True, None),
+                     id="1p3p, da durch die Begrenzung des LM nicht mit maximalem Strom geladen wird"),
+        pytest.param(10, [9.8, 0, 0], 5000,
+                     LoadmanagementLimit(message=", da die maximale Schieflast an Zähler Test erreicht ist.",
+                                         limiting_value=LimitingValue.UNBALANCED_LOAD), (True, None),
+                     id="1p3p, da durch die Begrenzung der Schieflast nicht mit maximalem Strom geladen wird"),
     ])
 def test_check_phase_switch_conditions(evse_current: int,
                                        get_currents: List[float],
                                        all_surplus: int,
+                                       limit: LoadmanagementLimit,
                                        expected: Tuple[bool, Optional[str]],
                                        monkeypatch):
     # setup
@@ -183,7 +194,7 @@ def test_check_phase_switch_conditions(evse_current: int,
         get_currents,
         sum(get_currents)*230,
         16,
-        LoadmanagementLimit(None, None))
+        limit)
 
     # evaluation
     assert (phase_switch, condition_msg) == expected

packages/control/chargepoint/chargepoint.py

@@ -19,6 +19,7 @@
 from control.ev.ev import Ev
 from control import phase_switch
 from control.chargepoint.chargepoint_state import CHARGING_STATES, ChargepointState
+from control.limiting_value import loadmanagement_limit_factory
 from control.text import BidiState
 from helpermodules.phase_handling import convert_single_evu_phase_to_cp_phase
 from helpermodules.pub import Pub
@@ -266,6 +267,9 @@ def reset_control_parameter_at_charge_stop(self) -> None:
         control_parameter = control_parameter_factory()
         self.data.control_parameter = control_parameter
 
+    def reset_values_before_algorithm(self) -> None:
+        self.data.control_parameter.limit = loadmanagement_limit_factory()
+
     def initiate_control_pilot_interruption(self):
         """ prüft, ob eine Control Pilot- Unterbrechung erforderlich ist und führt diese durch.
         """

packages/control/ev/ev.py

@@ -277,8 +277,9 @@ def _check_phase_switch_conditions(self,
         all_surplus = data.data.counter_all_data.get_evu_counter().get_usable_surplus(feed_in_yield)
         required_surplus = control_parameter.min_current * max_phases_ev * 230 - get_power
         unbalanced_load_limit_reached = limit.limiting_value == LimitingValue.UNBALANCED_LOAD
-        condition_1_to_3 = (((get_medium_charging_current(get_currents) > max_current_range and
-                            all_surplus > required_surplus) or unbalanced_load_limit_reached) and
+        current_limit_reached = limit.limiting_value == LimitingValue.CURRENT
+        condition_1_to_3 = ((((get_medium_charging_current(get_currents) > max_current_range or current_limit_reached)
+                            and all_surplus > required_surplus) or unbalanced_load_limit_reached) and
                             phases_in_use == 1)
         condition_3_to_1 = get_medium_charging_current(
             get_currents) < min_current_range and all_surplus <= 0 and phases_in_use > 1

packages/dataclass_utils/_dataclass_asdict_test.py

@@ -1,6 +1,7 @@
 import pytest
 
 from dataclass_utils import asdict
+from dataclass_utils.conftest import MyDataclass
 
 
 class SingleValue:
@@ -37,3 +38,77 @@ def test_asdict(object, expected_dict: dict):
 
     # evaluation
     assert actual == expected_dict
+
+
+MY_DATACLASS_AS_DICT = {
+    "str_value": "string_value",
+    "float_value": 5.2,
+    "int_value": 6,
+    "enum_value": "value1",
+    "nested_dataclass": {
+        "nested_str": "nested string",
+        "nested_int": 42
+    },
+    "nested_dataclass_enum_value": {
+        "D1": "value1",
+        "D2": "value2"
+    },
+    "dict_of_dataclass_value": {"a": {"nested_int": 42,
+                                      "nested_str": "nested string"},
+                                "b": {"nested_int": 42,
+                                      "nested_str": "nested string"}},
+    "dict_value": {"a": "a", "b": 2},
+    "dict2_value": {"a": 1, "b": 2},
+    "list_value": ["a", 2, None],
+    "list2_value": ["a", 2, None],
+    # JSON kennt keine Tupel
+    "tuple_value": [None, "a", 2],
+    "tuple2_value": [None, "a", 2],
+
+    "optional_str_value": "string_value",
+    "optional_float_value": 5.2,
+    "optional_int_value": 6,
+    "optional_enum_value": "value1",
+    "optional_nested_dataclass": {
+        "nested_str": "nested string",
+        "nested_int": 42
+    },
+    "optional_nested_dataclass_enum_value": {
+        "D1": "value1",
+        "D2": "value2"
+    },
+    "optional_dict_of_dataclass_value": {"a": {"nested_int": 42,
+                                               "nested_str": "nested string"},
+                                         "b": {"nested_int": 42,
+                                               "nested_str": "nested string"}},
+    "optional_dict_value": {"a": "a", "b": 2},
+    "optional_dict2_value": {"a": 1, "b": 2},
+    "optional_list_value": ["a", 2, None],
+    "optional_list2_value": ["a", 2, None],
+    # JSON kennt keine Tupel
+    "optional_tuple_value": [None, "a", 2],
+    "optional_tuple2_value": [None, "a", 2],
+
+    "none_str_value": None,
+    "none_float_value": None,
+    "none_int_value": None,
+    "none_enum_value": None,
+    "none_nested_dataclass": None,
+    "none_nested_dataclass_enum_value": None,
+    "none_dict_of_dataclass_value": None,
+    "none_dict_value": None,
+    "none_dict2_value": None,
+    "none_list_value": None,
+    "none_list2_value": None,
+    "none_tuple_value": None,
+    "none_tuple2_value": None,
+
+}
+
+
+def test_dataclass_as_dict():
+    # execution
+    actual_dict = asdict(MyDataclass())
+
+    # evaluation
+    assert actual_dict == MY_DATACLASS_AS_DICT

packages/dataclass_utils/_dataclass_from_dict.py

@@ -1,7 +1,6 @@
 from enum import Enum
 import inspect
 from inspect import FullArgSpec, isclass
-import typing
 from typing import TypeVar, Type, Union, get_args, get_origin
 
 T = TypeVar('T')
@@ -20,8 +19,9 @@ def dataclass_from_dict(cls: Type[T], args: Union[dict, T]) -> T:
         if isinstance(args, cls):
             return args
     elif get_origin(cls):
-        # Generische Typen wie Dict[int, float]
-        if isinstance(args, get_origin(cls)):
+        # Generische Typen wie Dict[int, float] - aber nicht Union, da isinstance mit Union fehlschlägt
+        origin = get_origin(cls)
+        if origin != Union and isinstance(args, origin):
             return args
     elif isinstance(args, type(cls)):
         return args
@@ -46,26 +46,27 @@ def _get_argument_value(arg_spec: FullArgSpec, index: int, parameters: dict):
 
 
 def _dataclass_from_dict_recurse(value, requested_type: Type[T]):
-    if get_origin(requested_type) == list:
+    # Handle Optional types (Union[X, None]) - extract the actual type
+    actual_type = requested_type
+    if get_origin(requested_type) == Union:
+        args = get_args(requested_type)
+        if len(args) == 2 and args[1].__name__ == 'NoneType':
+            if value is None:
+                return None
+            actual_type = args[0]  # Extract X from Optional[X]
+
+    if get_origin(actual_type) == list:
         # Extrahiere den generischen Typ der Liste
-        if get_args(requested_type):
-            generic_type = get_args(requested_type)[0]
+        if get_args(actual_type):
+            generic_type = get_args(actual_type)[0]
             # Konvertiere jedes Element der Liste in den generischen Typ
             return [_dataclass_from_dict_recurse(item, generic_type) for item in value]
 
-    if isinstance(value, dict) and not (
-            _is_optional_of_dict(requested_type) or
-            issubclass(requested_type if isclass(requested_type) else type(bool), dict)):
-        return dataclass_from_dict(requested_type, value)
-    if isinstance(requested_type, type) and issubclass(requested_type, Enum):
-        return requested_type(value)
-    return value
-
+    # Handle dict types (both direct and Optional[dict])
+    if isinstance(value, dict) and isclass(actual_type) and not issubclass(actual_type, dict):
+        return dataclass_from_dict(actual_type, value)
 
-def _is_optional_of_dict(requested_type):
-    # Optional[dict] is an alias for Union[dict, None]
-    if typing.get_origin(requested_type) == Union:
-        args = typing.get_args(requested_type)
-        if len(args) == 2:
-            return issubclass(args[0], dict) and issubclass(args[1], type(None))
-    return False
+    # Handle Enum types (both direct and Optional[Enum])
+    if isinstance(actual_type, type) and issubclass(actual_type, Enum):
+        return actual_type(value)
+    return value

packages/dataclass_utils/_dataclass_from_dict_test.py

@@ -3,6 +3,7 @@
 import pytest
 
 from dataclass_utils import dataclass_from_dict
+from dataclass_utils.conftest import MyDataclass
 
 T = TypeVar('T')
 
@@ -125,3 +126,67 @@ def test_from_dict_without_optional():
     # evaluation
     assert actual.a == "aValue"
     assert actual.o is None
+
+
+MY_DATACLASS_AS_DICT = {
+    "str_value": "string_value",
+    "float_value": 5.2,
+    "int_value": 6,
+    "enum_value": "value1",
+    "nested_dataclass": {
+        "nested_str": "nested string",
+        "nested_int": 42
+    },
+    "nested_dataclass_enum_value": {
+        "D1": "value1",
+        "D2": "value2"
+    },
+    "dict_value": {"a": "a", "b": 2},
+    "dict2_value": {"a": 1, "b": 2},
+    "list_value": ["a", 2, None],
+    "list2_value": ["a", 2, None],
+    "tuple_value": (None, "a", 2),
+    "tuple2_value": (None, "a", 2),
+
+    "optional_str_value": "string_value",
+    "optional_float_value": 5.2,
+    "optional_int_value": 6,
+    "optional_enum_value": "value1",
+    "optional_nested_dataclass": {
+        "nested_str": "nested string",
+        "nested_int": 42
+    },
+    "optional_nested_dataclass_enum_value": {
+        "D1": "value1",
+        "D2": "value2"
+    },
+    "optional_dict_value": {"a": "a", "b": 2},
+    "optional_dict2_value": {"a": 1, "b": 2},
+    "optional_list_value": ["a", 2, None],
+    "optional_list2_value": ["a", 2, None],
+    "optional_tuple_value": (None, "a", 2),
+    "optional_tuple2_value": (None, "a", 2),
+
+    "none_str_value": None,
+    "none_float_value": None,
+    "none_int_value": None,
+    "none_enum_value": None,
+    "none_nested_dataclass": None,
+    "none_nested_dataclass_enum_value": None,
+    "none_dict_of_dataclass_value": None,
+    "none_dict_value": None,
+    "none_dict2_value": None,
+    "none_list_value": None,
+    "none_list2_value": None,
+    "none_tuple_value": None,
+    "none_tuple2_value": None,
+
+}
+
+
+def test_dataclass_from_dict():
+    # execution
+    actual_dict = dataclass_from_dict(MyDataclass, MY_DATACLASS_AS_DICT)
+
+    # evaluation
+    assert vars(actual_dict) == vars(MyDataclass())