increase robustness of dataclass from dict; add missing electricity pricing update config

packages/dataclass_utils/_dataclass_from_dict.py

@@ -1,4 +1,5 @@
 from enum import Enum
+from dataclasses import MISSING, fields, is_dataclass
 import inspect
 from inspect import FullArgSpec, isclass
 from typing import TypeVar, Type, Union, get_args, get_origin
@@ -26,22 +27,39 @@ def dataclass_from_dict(cls: Type[T], args: Union[dict, T]) -> T:
     elif isinstance(args, type(cls)):
         return args
     arg_spec = inspect.getfullargspec(cls.__init__)
-    return cls(*[_get_argument_value(arg_spec, index, args) for index in range(1, len(arg_spec.args))])
+    return cls(*[_get_argument_value(cls, arg_spec, index, args) for index in range(1, len(arg_spec.args))])
 
 
-def _get_argument_value(arg_spec: FullArgSpec, index: int, parameters: dict):
+def _get_argument_value(cls: Type[T], arg_spec: FullArgSpec, index: int, parameters: dict):
     argument_name = arg_spec.args[index]
     try:
         value = parameters[argument_name]
     except KeyError:
-        try:
-            value = arg_spec.defaults[-len(arg_spec.args) + index]
-        except (IndexError, TypeError):
-            # If none of the parameters have a default value, then `arg_spec.defaults` is None and we get a `TypeError`.
-            # If there are parameters with default value, but not the one requested, we get an `IndexError`.
-            raise Exception(
-                "Cannot determine value for parameter %s: not given in %s and no default value specified" % (
-                    argument_name, parameters))
+        if is_dataclass(cls):
+            dataclass_field = next((field for field in fields(cls) if field.name == argument_name), None)
+            if dataclass_field is not None:
+                if dataclass_field.default_factory is not MISSING:
+                    value = dataclass_field.default_factory()
+                elif dataclass_field.default is not MISSING:
+                    value = dataclass_field.default
+                else:
+                    raise Exception(
+                        "Cannot determine value for parameter %s: not given in %s and no default value specified" % (
+                            argument_name, parameters))
+            else:
+                raise Exception(
+                    "Cannot determine value for parameter %s: not given in %s and no dataclass field found" % (
+                        argument_name, parameters))
+        else:
+            try:
+                value = arg_spec.defaults[-len(arg_spec.args) + index]
+            except (IndexError, TypeError):
+                # If none of the parameters have a default value, then `arg_spec.defaults` is None and we get a
+                # `TypeError`. If there are parameters with default value, but not the one requested, we get an
+                # `IndexError`.
+                raise Exception(
+                    "Cannot determine value for parameter %s: not given in %s and no default value specified" % (
+                        argument_name, parameters))
     return _dataclass_from_dict_recurse(value, arg_spec.annotations.get(argument_name))
 
 

packages/dataclass_utils/_dataclass_from_dict_test.py

@@ -1,3 +1,4 @@
+from dataclasses import dataclass, field
 from typing import Dict, Generic, Optional, Type, TypeVar
 
 import pytest
@@ -42,6 +43,12 @@ def __init__(self, a: str, o: Dict[int, float] = None):
         self.o = o
 
 
+@dataclass
+class DataclassWithDefaultFactoryList:
+    values: list[int] = field(default_factory=lambda: [14])
+    name: str = "default-name"
+
+
 def test_from_dict_simple():
     # execution
     actual = dataclass_from_dict(SimpleSample, {"b": "bValue", "a": "aValue"})
@@ -128,6 +135,17 @@ def test_from_dict_without_optional():
     assert actual.o is None
 
 
+def test_from_dict_dataclass_uses_default_factory_if_key_missing():
+    # execution
+    actual = dataclass_from_dict(DataclassWithDefaultFactoryList, {"name": "configured"})
+    default_instance = DataclassWithDefaultFactoryList()
+
+    # evaluation
+    assert actual.name == "configured"
+    assert actual.values == default_instance.values  # vergleicht die Werte
+    assert actual.values is not default_instance.values  # vergleicht die Instanzen
+
+
 MY_DATACLASS_AS_DICT = {
     "str_value": "string_value",
     "float_value": 5.2,

packages/helpermodules/update_config.py

@@ -57,7 +57,7 @@
 
 class UpdateConfig:
 
-    DATASTORE_VERSION = 128
+    DATASTORE_VERSION = 129
 
     valid_topic = [
         "^openWB/bat/config/bat_control_activated$",
@@ -3264,3 +3264,27 @@ def upgrade_component(topic: str, payload) -> Optional[dict]:
             )
 
         self._append_datastore_version(128)
+
+    def upgrade_datastore_129(self) -> None:
+        def upgrade(topic: str, payload) -> None:
+            if "openWB/optional/ep/flexible_tariff/provider" == topic:
+                provider = decode_payload(payload)
+                if provider["type"] == "fixed_hours":
+                    if provider["configuration"].get("update_hours") is None:
+                        provider["configuration"]["update_hours"] = list(range(24))
+                        return {topic: provider}
+                elif provider["type"] == "ostrom":
+                    if provider["configuration"].get("update_hours") is None:
+                        provider["configuration"]["update_hours"] = [2, 8, 14, 20]
+                        return {topic: provider}
+                elif provider["type"] == "tibber":
+                    if (provider["configuration"].get("update_hours") is None or
+                            provider["configuration"].get("includes_grid_fee") is None):
+                        if provider["configuration"].get("update_hours") is None:
+                            provider["configuration"]["update_hours"] = [14]
+                        if provider["configuration"].get("includes_grid_fee") is None:
+                            provider["configuration"]["includes_grid_fee"] = True
+                        return {topic: provider}
+            return None
+        self._loop_all_received_topics(upgrade)
+        self._append_datastore_version(129)