Feature integrated charging plans

packages/control/chargepoint/chargepoint.py

@@ -14,7 +14,6 @@
 Tag-Liste: Tags, mit denen der Ladepunkt freigeschaltet werden kann. Ist diese leer, kann mit jedem Tag der Ladepunkt
 freigeschaltet werden.
 """
-import copy
 from dataclasses import asdict
 import dataclasses
 import logging
@@ -35,7 +34,6 @@
 from control.ev.ev import Ev
 from control import phase_switch
 from control.chargepoint.chargepoint_state import CHARGING_STATES, ChargepointState
-from helpermodules.abstract_plans import ScheduledChargingPlan, TimeChargingPlan
 from helpermodules.broker import BrokerClient
 from helpermodules.phase_mapping import convert_single_evu_phase_to_cp_phase
 from helpermodules.pub import Pub
@@ -800,41 +798,8 @@ def __get_payload(client, userdata, msg):
             Pub().pub(topic, "")
 
     def update_charge_template(self, charge_template: ChargeTemplate) -> None:
-        self._clear_template_topics(f'openWB/chargepoint/{self.num}/set/charge_template/#')
-        self.data.set.charge_template = copy.deepcopy(charge_template)
-        pub_template = copy.deepcopy(self.data.set.charge_template.data)
-        pub_template = dataclasses.asdict(pub_template)
-        pub_template["chargemode"]["scheduled_charging"]["plans"].clear()
-        pub_template["time_charging"]["plans"].clear()
-        Pub().pub(f"openWB/set/chargepoint/{self.num}/set/charge_template", pub_template)
-        for id, plan in self.data.set.charge_template.data.time_charging.plans.items():
-            Pub().pub(f"openWB/set/chargepoint/{self.num}/set/charge_template/time_charging/plans/{id}",
-                      dataclasses.asdict(plan))
-        for id, plan in self.data.set.charge_template.data.chargemode.scheduled_charging.plans.items():
-            Pub().pub(f"openWB/set/chargepoint/{self.num}/set/charge_template/chargemode/scheduled_charging/plans/{id}",
-                      dataclasses.asdict(plan))
-
-    def update_bare_charge_template(self, charge_template: ChargeTemplate) -> None:
-        self._clear_template_topics(f"openWB/chargepoint/{self.num}/set/charge_template")
-        self.data.set.charge_template = copy.deepcopy(charge_template)
-        pub_template = copy.deepcopy(self.data.set.charge_template.data)
-        pub_template = dataclasses.asdict(pub_template)
-        pub_template["chargemode"]["scheduled_charging"]["plans"].clear()
-        pub_template["time_charging"]["plans"].clear()
-        Pub().pub(f"openWB/set/chargepoint/{self.num}/set/charge_template", pub_template)
-
-    def update_charge_template_scheduled_plan(self, plan: ScheduledChargingPlan) -> None:
-        self._clear_template_topics(
-            f"openWB/chargepoint/{self.num}/set/charge_template/chargemode/scheduled_charging/plans/{plan.id}")
-        Pub().pub(f"openWB/set/chargepoint/{self.num}"
-                  f"/set/charge_template/chargemode/scheduled_charging/plans/{plan.id}",
-                  dataclasses.asdict(plan))
-
-    def update_charge_template_time_plan(self, plan: TimeChargingPlan) -> None:
-        self._clear_template_topics(
-            f"openWB/chargepoint/{self.num}/set/charge_template/time_charging/plans/{plan.id}")
-        Pub().pub(f"openWB/set/chargepoint/{self.num}/set/charge_template/time_charging/plans/{plan.id}",
-                  dataclasses.asdict(plan))
+        Pub().pub(f"openWB/set/chargepoint/{self.num}/set/charge_template",
+                  dataclasses.asdict(charge_template.data))
 
     def _pub_connected_vehicle(self, vehicle: Ev):
         """ published die Daten, die zur Anzeige auf der Hauptseite benötigt werden.
@@ -903,11 +868,15 @@ def chargemode_support_phase_switch(self) -> bool:
                            control_parameter.chargemode == Chargemode.ECO_CHARGING) and
                           control_parameter.submode == Chargemode.PV_CHARGING and
                           self.data.set.charge_template.data.chargemode.pv_charging.phases_to_use == 0)
+        for p in self.data.set.charge_template.data.chargemode.scheduled_charging.plans:
+            if p.id == self.data.control_parameter.current_plan:
+                phases_to_use_pv = p.phases_to_use_pv
+        else:
+            phases_to_use_pv = 1
         scheduled_auto_switch = (
             control_parameter.chargemode == Chargemode.SCHEDULED_CHARGING and
             control_parameter.submode == Chargemode.PV_CHARGING and
-            self.data.set.charge_template.data.chargemode.scheduled_charging.plans[
-                str(self.data.control_parameter.current_plan)].phases_to_use_pv == 0)
+            phases_to_use_pv == 0)
         return (pv_auto_switch or scheduled_auto_switch)
 
     def failed_phase_switches_reached(self) -> bool:

packages/control/chargepoint/chargepoint_template.py

@@ -1,12 +1,12 @@
 from dataclasses import asdict, dataclass, field
 import logging
 import traceback
-from typing import Dict, List
+from typing import List
 
 from control import data
 from control.ev import ev as ev_module
 from control.chargepoint.charging_type import ChargingType
-from dataclass_utils.factories import empty_dict_factory, empty_list_factory
+from dataclass_utils.factories import empty_list_factory
 from helpermodules.abstract_plans import AutolockPlan
 from helpermodules import timecheck
 
@@ -16,18 +16,13 @@
 
 def get_chargepoint_template_default():
     default = asdict(CpTemplateData())
-    default["autolock"].pop("plans")
     return default
 
 
-def get_autolock_plan_default():
-    return asdict(AutolockPlan())
-
-
 @dataclass
 class Autolock:
     active: bool = False
-    plans: Dict[int, AutolockPlan] = field(default_factory=empty_dict_factory)
+    plans: List[AutolockPlan] = field(default_factory=empty_list_factory)
     wait_for_charging_end: bool = False
 
 

packages/control/ev/charge_template.py

@@ -2,43 +2,39 @@
 import datetime
 import logging
 import traceback
-from typing import Optional, Tuple
+from typing import List, Optional, Tuple
 
 from control import data
 from control.chargepoint.chargepoint_state import CHARGING_STATES
 from control.chargepoint.charging_type import ChargingType
 from control.chargepoint.control_parameter import ControlParameter
 from control.ev.ev_template import EvTemplate
-from dataclass_utils.factories import empty_dict_factory
-from helpermodules.abstract_plans import Limit, limit_factory, ScheduledChargingPlan
+from dataclass_utils.factories import empty_list_factory
+from helpermodules.abstract_plans import Limit, TimeChargingPlan, limit_factory, ScheduledChargingPlan
 from helpermodules import timecheck
 log = logging.getLogger(__name__)
 
 
 def get_new_charge_template() -> dict:
     ct_default = asdict(ChargeTemplateData())
-    ct_default["chargemode"]["scheduled_charging"].pop("plans")
-    ct_default["time_charging"].pop("plans")
     return ct_default
 
 
 def get_charge_template_default() -> dict:
     ct_default = asdict(ChargeTemplateData(name="Standard-Lade-Profil"))
-    ct_default["chargemode"]["scheduled_charging"].pop("plans")
-    ct_default["time_charging"].pop("plans")
     return ct_default
 
 
 @dataclass
 class ScheduledCharging:
-    plans: dict = field(default_factory=empty_dict_factory, metadata={
+    plans: List[ScheduledChargingPlan] = field(default_factory=empty_list_factory, metadata={
         "topic": ""})  # Dict[int,ScheduledChargingPlan] wird bei der dict to dataclass Konvertierung nicht unterstützt
 
 
 @dataclass
 class TimeCharging:
     active: bool = False
-    plans: dict = field(default_factory=empty_dict_factory, metadata={
+    plans: List[TimeChargingPlan] = field(default_factory=empty_list_factory, metadata={
         "topic": ""})  # Dict[int, TimeChargingPlan] wird bei der dict to dataclass Konvertierung nicht unterstützt
 
 
@@ -320,7 +316,7 @@ def scheduled_charging_recent_plan(self,
                                        control_parameter: ControlParameter,
                                        soc_request_interval_offset: int) -> Optional[SelectedPlan]:
         plans_diff_end_date = []
-        for p in self.data.chargemode.scheduled_charging.plans.values():
+        for p in self.data.chargemode.scheduled_charging.plans:
             if p.active:
                 if p.limit.selected == "soc" and soc is None:
                     raise ValueError("Um Zielladen mit SoC-Ziel nutzen zu können, bitte ein SoC-Modul konfigurieren "
@@ -347,7 +343,9 @@ def scheduled_charging_recent_plan(self,
                     plan_id = list(plan_dict.keys())[0]
                     plan_end_time = list(plan_dict.values())[0]
 
-                    plan = self.data.chargemode.scheduled_charging.plans[str(plan_id)]
+                    for p in self.data.chargemode.scheduled_charging.plans:
+                        if p.id == plan_id:
+                            plan = p
 
                     remaining_time, missing_amount, phases, duration = self._calc_remaining_time(
                         plan, plan_end_time, soc, ev_template, used_amount, max_hw_phases, phase_switch_supported,

packages/control/ev/charge_template_test.py

@@ -199,7 +199,7 @@ def test_scheduled_charging_recent_plan(end_time_mock,
     plan_mock_0 = Mock(spec=ScheduledChargingPlan, active=True, current=14, id=0, limit=Limit(selected="amount"))
     plan_mock_1 = Mock(spec=ScheduledChargingPlan, active=True, current=14, id=1, limit=Limit(selected="amount"))
     plan_mock_2 = Mock(spec=ScheduledChargingPlan, active=True, current=14, id=2, limit=Limit(selected="amount"))
-    ct.data.chargemode.scheduled_charging.plans = {"0": plan_mock_0, "1": plan_mock_1, "2": plan_mock_2}
+    ct.data.chargemode.scheduled_charging.plans = [plan_mock_0, plan_mock_1, plan_mock_2]
 
     # execution
     selected_plan = ct.scheduled_charging_recent_plan(
@@ -230,7 +230,7 @@ def test_scheduled_charging_recent_plan_fulfilled(end_time_mock, expected_plan_n
     plan_mock_0 = Mock(spec=ScheduledChargingPlan, active=True, current=14, id=0, limit=Limit(selected="amount"))
     plan_mock_1 = Mock(spec=ScheduledChargingPlan, active=True, current=14, id=1, limit=Limit(selected="amount"))
     plan_mock_2 = Mock(spec=ScheduledChargingPlan, active=True, current=14, id=2, limit=Limit(selected="amount"))
-    ct.data.chargemode.scheduled_charging.plans = {"0": plan_mock_0, "1": plan_mock_1, "2": plan_mock_2}
+    ct.data.chargemode.scheduled_charging.plans = [plan_mock_0, plan_mock_1, plan_mock_2]
 
     # execution
     selected_plan = ct.scheduled_charging_recent_plan(
@@ -284,7 +284,7 @@ def test_scheduled_charging_calc_current(plan_data: SelectedPlan,
     plan = ScheduledChargingPlan(active=True, id=0)
     plan.limit.selected = selected
     # json verwandelt Keys in strings
-    ct.data.chargemode.scheduled_charging.plans = {"0": plan}
+    ct.data.chargemode.scheduled_charging.plans = [plan]
     if plan_data:
         plan_data.plan = plan
 
@@ -318,9 +318,9 @@ def test_scheduled_charging_calc_current_electricity_tariff(loading_hour, expect
     # setup
     ct = ChargeTemplate()
     plan = ScheduledChargingPlan(active=True)
+    plan.et_active = True
     plan.limit.selected = "soc"
-    ct.data.chargemode.scheduled_charging.plans = {"0": plan}
-    ct.data.chargemode.scheduled_charging.plans["0"].et_active = True
+    ct.data.chargemode.scheduled_charging.plans = [plan]
     # für Github-Test keinen Zeitstempel verwenden
     mock_et_get_loading_hours = Mock(return_value=[datetime.datetime(
         year=2022, month=5, day=16, hour=8, minute=0).timestamp()])

packages/dataclass_utils/_dataclass_from_dict.py

@@ -2,7 +2,7 @@
 import inspect
 from inspect import FullArgSpec, isclass
 import typing
-from typing import TypeVar, Type, Union, get_origin
+from typing import TypeVar, Type, Union, get_args, get_origin
 
 T = TypeVar('T')
 
@@ -46,15 +46,20 @@ 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:
+        # Extrahiere den generischen Typ der Liste
+        if get_args(requested_type):
+            generic_type = get_args(requested_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)
-    else:
-        if isinstance(requested_type, type) and issubclass(requested_type, Enum):
-            return requested_type(value)
-        else:
-            return value
+    if isinstance(requested_type, type) and issubclass(requested_type, Enum):
+        return requested_type(value)
+    return value
 
 
 def _is_optional_of_dict(requested_type):

packages/helpermodules/abstract_plans.py

@@ -3,8 +3,12 @@
 from typing import List, Optional
 
 
-def once_factory() -> List:
-    return [datetime.datetime.today().strftime("%Y%m%d"), datetime.datetime.today().strftime("%Y%m%d")]
+def once_period_factory() -> List:
+    return [datetime.datetime.today().strftime("%Y-%m-%d"), datetime.datetime.today().strftime("%Y-%m-%d")]
+
+
+def once_date_factory() -> List:
+    return datetime.datetime.today().strftime("%Y-%m-%d")
 
 
 def weekly_factory() -> List:
@@ -27,14 +31,25 @@ def limit_factory() -> Limit:
 
 
 @dataclass
-class Frequency:
+class FrequencyPeriod:
+    selected: str = "daily"
+    once: List[str] = field(default_factory=once_period_factory)
+    weekly: List[bool] = field(default_factory=weekly_factory)
+
+
+def frequency_period_factory() -> FrequencyPeriod:
+    return FrequencyPeriod()
+
+
+@dataclass
+class FrequencyDate:
     selected: str = "daily"
-    once: List[str] = field(default_factory=once_factory)
+    once: str = field(default_factory=once_date_factory)
     weekly: List[bool] = field(default_factory=weekly_factory)
 
 
-def frequency_factory() -> Frequency:
-    return Frequency()
+def frequency_date_factory() -> FrequencyDate:
+    return FrequencyDate()
 
 
 @dataclass
@@ -52,19 +67,20 @@ def scheduled_limit_factory() -> ScheduledLimit:
 @dataclass
 class PlanBase:
     active: bool = True
-    frequency: Frequency = field(default_factory=frequency_factory)
 
 
 @dataclass
 class TimeframePlan(PlanBase):
     time: List[str] = field(default_factory=time_factory)  # ToDo: aktuelle Zeit verwenden + 1 Stunde
+    frequency: FrequencyPeriod = field(default_factory=frequency_period_factory)
 
 
 @dataclass
 class ScheduledChargingPlan(PlanBase):
     current: int = 14
     dc_current: float = 145
     et_active: bool = False
+    frequency: FrequencyDate = field(default_factory=frequency_date_factory)
     id: Optional[int] = None
     name: str = "neuer Zielladen-Plan"
     limit: ScheduledLimit = field(default_factory=scheduled_limit_factory)
@@ -85,4 +101,5 @@ class TimeChargingPlan(TimeframePlan):
 
 @dataclass
 class AutolockPlan(TimeframePlan):
+    id: Optional[int] = None
     name: str = "neuer Plan für Sperren nach Uhrzeit"