From 9a05ff2981c622a69f58cff22fd0f2249be00ed5 Mon Sep 17 00:00:00 2001 From: LKuemmel Date: Wed, 19 Aug 2026 15:54:10 +0200 Subject: [PATCH 1/7] use chargemode enums for chargepoints --- packages/control/chargelog/chargelog.py | 5 +- packages/control/chargepoint/chargepoint.py | 23 ++-- packages/control/ev/charge_template.py | 83 ++++++++------ packages/control/ev/charge_template_test.py | 114 +++++++++++--------- packages/control/ev/ev.py | 19 ++-- packages/helpermodules/subdata.py | 3 + 6 files changed, 137 insertions(+), 110 deletions(-) diff --git a/packages/control/chargelog/chargelog.py b/packages/control/chargelog/chargelog.py index dcfbbfeb3d..621dbb06d4 100644 --- a/packages/control/chargelog/chargelog.py +++ b/packages/control/chargelog/chargelog.py @@ -1,3 +1,4 @@ +from control.chargemode import Chargemode from helpermodules.utils.json_file_handler import write_and_check from helpermodules import timecheck import copy @@ -99,8 +100,8 @@ def collect_data(chargepoint): if get_value_or_default(lambda: log_data.timestamp_start_charging is None): log_data.timestamp_start_charging = now submode = get_value_or_default(lambda: chargepoint.data.control_parameter.submode, "") - if submode == "time_charging": - log_data.chargemode_log_entry = "time_charging" + if submode == Chargemode.TIME_CHARGING: + log_data.chargemode_log_entry = Chargemode.TIME_CHARGING.value else: log_data.chargemode_log_entry = get_value_or_default( lambda: chargepoint.data.control_parameter.chargemode.value) diff --git a/packages/control/chargepoint/chargepoint.py b/packages/control/chargepoint/chargepoint.py index 8fdf400905..5975ae32ab 100644 --- a/packages/control/chargepoint/chargepoint.py +++ b/packages/control/chargepoint/chargepoint.py @@ -218,7 +218,7 @@ def _process_charge_stop(self) -> None: def setup_values_at_start(self): self._set_values_at_start() - def set_control_parameter(self, submode: str): + def set_control_parameter(self, submode: Chargemode): """ setzt die Regel-Parameter, die der Algorithmus verwendet. Parameter @@ -227,8 +227,8 @@ def set_control_parameter(self, submode: str): neuer Lademodus, in dem geladen werden soll """ try: - self.data.control_parameter.submode = Chargemode(submode) - if submode == "time_charging": + self.data.control_parameter.submode = submode + if submode == Chargemode.TIME_CHARGING: self.data.control_parameter.chargemode = Chargemode.TIME_CHARGING else: self.data.control_parameter.chargemode = Chargemode( @@ -602,9 +602,10 @@ def set_timestamp_charge_start(self): elif self.data.set.current == 0: self.data.control_parameter.timestamp_charge_start = None - def set_chargemode_changed(self, submode: str) -> None: - if ((submode == "time_charging" and self.data.control_parameter.chargemode != "time_charging") or - (submode != "time_charging" and + def set_chargemode_changed(self, submode: Chargemode) -> None: + if ((submode == Chargemode.TIME_CHARGING and + self.data.control_parameter.chargemode != Chargemode.TIME_CHARGING) or + (submode != Chargemode.TIME_CHARGING and self.data.control_parameter.chargemode != self.data.set.charge_template.data.chargemode.selected)): self.chargemode_changed = True log.debug("Änderung des Lademodus") @@ -612,7 +613,7 @@ def set_chargemode_changed(self, submode: str) -> None: else: self.chargemode_changed = False - def set_submode_changed(self, submode: str) -> None: + def set_submode_changed(self, submode: Chargemode) -> None: self.submode_changed = (submode != self.data.control_parameter.submode) def update_ev(self, ev_list: Dict[str, Ev]) -> None: @@ -700,7 +701,7 @@ def update(self, ev_list: Dict[str, Ev]) -> None: f", mittlerer Ist-Strom: {get_medium_charging_current(self.data.get.currents)}") except Exception: log.exception("Fehler im Prepare-Modul für Ladepunkt "+str(self.num)) - self.data.control_parameter.submode = "stop" + self.data.control_parameter.submode = Chargemode.STOP else: self._process_charge_stop() if vehicle != -1: @@ -807,8 +808,8 @@ def _pub_connected_vehicle(self, vehicle: Ev): self.data.get.connected_vehicle.soc.range = vehicle.data.get.range self.data.get.connected_vehicle.info = ConnectedInfo(id=vehicle.num, name=vehicle.data.name) - if (self.data.set.charge_template.data.chargemode.selected == "time_charging" or - self.data.set.charge_template.data.chargemode.selected == "scheduled_charging"): + if (self.data.set.charge_template.data.chargemode.selected == Chargemode.TIME_CHARGING.value or + self.data.set.charge_template.data.chargemode.selected == Chargemode.SCHEDULED_CHARGING.value): current_plan = self.data.control_parameter.current_plan else: current_plan = None @@ -820,7 +821,7 @@ def _pub_connected_vehicle(self, vehicle: Ev): current_plan=current_plan, average_consumption=vehicle.ev_template.data.average_consump, time_charging_in_use=True if (self.data.control_parameter.submode == - "time_charging") else False) + Chargemode.TIME_CHARGING) else False) except Exception: log.exception("Fehler im Prepare-Modul") diff --git a/packages/control/ev/charge_template.py b/packages/control/ev/charge_template.py index 20f7ba88bb..fee8e20d78 100644 --- a/packages/control/ev/charge_template.py +++ b/packages/control/ev/charge_template.py @@ -5,6 +5,7 @@ from typing import List, Optional, Tuple from control import data +from control.chargemode import Chargemode as ChargemodeEnum from control.chargepoint.chargepoint_state import CHARGING_STATES from control.chargepoint.charging_type import ChargingType from control.chargepoint.control_parameter import ControlParameter @@ -91,7 +92,7 @@ def instant_charging_factory() -> InstantCharging: @dataclass class Chargemode: - selected: str = "instant_charging" + selected: str = ChargemodeEnum.INSTANT_CHARGING.value eco_charging: EcoCharging = field(default_factory=eco_charging_factory) pv_charging: PvCharging = field(default_factory=pv_charging_factory) scheduled_charging: ScheduledCharging = field(default_factory=scheduled_charging_factory) @@ -151,11 +152,11 @@ class ChargeTemplate: def time_charging(self, soc: Optional[float], used_amount_time_charging: float, - charging_type: str) -> Tuple[int, str, Optional[str], Optional[str], int]: + charging_type: str) -> Tuple[int, ChargemodeEnum, Optional[str], Optional[str], int]: """ prüft, ob ein Zeitfenster aktiv ist und setzt entsprechend den Ladestrom """ message = None - sub_mode = "stop" + sub_mode = ChargemodeEnum.STOP current = 0 id = None phases = None @@ -179,12 +180,12 @@ def time_charging(self, log.debug( "Zeitladen: minimaler Speicher-SoC überschritten, Laden mit Zeitladen möglich.") current = plan.current if charging_type == ChargingType.AC.value else plan.dc_current - sub_mode = "time_charging" + sub_mode = ChargemodeEnum.TIME_CHARGING else: message = self.TIME_CHARGING_CONFLICT_ACTIVE_BAT_CONTROL else: current = plan.current if charging_type == ChargingType.AC.value else plan.dc_current - sub_mode = "time_charging" + sub_mode = ChargemodeEnum.TIME_CHARGING else: message = self.TIME_CHARGING_NO_PLAN_ACTIVE else: @@ -192,7 +193,10 @@ def time_charging(self, return current, sub_mode, message, id, phases except Exception: log.exception("Fehler im ev-Modul "+str(self.data.id)) - return (0, "stop", "Keine Ladung, da da ein interner Fehler aufgetreten ist: "+traceback.format_exc(), None, + return (0, + ChargemodeEnum.STOP, + "Keine Ladung, da da ein interner Fehler aufgetreten ist: "+traceback.format_exc(), + None, 0) SOC_REACHED = "Keine Ladung, da das Ladeziel bereits erreicht wurde." @@ -201,11 +205,11 @@ def time_charging(self, def instant_charging(self, soc: Optional[float], used_amount: float, - charging_type: str) -> Tuple[int, str, Optional[str], int]: + charging_type: str) -> Tuple[int, ChargemodeEnum, Optional[str], int]: """ prüft, ob die Lademengenbegrenzung erreicht wurde und setzt entsprechend den Ladestrom. """ message = None - sub_mode = "instant_charging" + sub_mode = ChargemodeEnum.INSTANT_CHARGING try: instant_charging = self.data.chargemode.instant_charging phases = instant_charging.phases_to_use @@ -216,17 +220,20 @@ def instant_charging(self, if instant_charging.limit.selected == "soc" and soc is not None and soc >= instant_charging.limit.soc: current = 0 - sub_mode = "stop" + sub_mode = ChargemodeEnum.STOP message = self.SOC_REACHED elif instant_charging.limit.selected == "amount": if used_amount >= self.data.chargemode.instant_charging.limit.amount: current = 0 - sub_mode = "stop" + sub_mode = ChargemodeEnum.STOP message = self.AMOUNT_REACHED return current, sub_mode, message, phases except Exception: log.exception("Fehler im ev-Modul "+str(self.data.id)) - return 0, "stop", "Keine Ladung, da da ein interner Fehler aufgetreten ist: "+traceback.format_exc(), 0 + return (0, + ChargemodeEnum.STOP, + "Keine Ladung, da da ein interner Fehler aufgetreten ist: "+traceback.format_exc(), + 0) PV_CHARGING_SOC_CHARGING = ("Ladung evtl. auch ohne PV-Überschuss, da der Mindest-SoC des Fahrzeugs noch nicht " "erreicht wurde.") @@ -236,11 +243,11 @@ def pv_charging(self, soc: Optional[float], min_current: int, charging_type: str, - used_amount: float) -> Tuple[int, str, Optional[str], int]: + used_amount: float) -> Tuple[int, ChargemodeEnum, Optional[str], int]: """ prüft, ob Min-oder Max-Soc erreicht wurden und setzt entsprechend den Ladestrom. """ message = None - sub_mode = "pv_charging" + sub_mode = ChargemodeEnum.PV_CHARGING try: pv_charging = self.data.chargemode.pv_charging phases = pv_charging.phases_to_use @@ -248,11 +255,11 @@ def pv_charging(self, else pv_charging.dc_min_current) if pv_charging.limit.selected == "soc" and soc is not None and soc >= pv_charging.limit.soc: current = 0 - sub_mode = "stop" + sub_mode = ChargemodeEnum.STOP message = self.SOC_REACHED elif pv_charging.limit.selected == "amount" and used_amount >= pv_charging.limit.amount: current = 0 - sub_mode = "stop" + sub_mode = ChargemodeEnum.STOP message = self.AMOUNT_REACHED else: if pv_charging.min_soc != 0 and soc is not None and soc < pv_charging.min_soc: @@ -260,33 +267,36 @@ def pv_charging(self, current = pv_charging.min_soc_current else: current = pv_charging.dc_min_soc_current - sub_mode = "instant_charging" + sub_mode = ChargemodeEnum.INSTANT_CHARGING message = self.PV_CHARGING_SOC_CHARGING phases = pv_charging.phases_to_use_min_soc elif min_pv_current == 0: # nur PV; Ampere darf nicht 0 sein, wenn geladen werden soll current = min_current - sub_mode = "pv_charging" + sub_mode = ChargemodeEnum.PV_CHARGING else: # Min PV current = min_pv_current - sub_mode = "instant_charging" + sub_mode = ChargemodeEnum.INSTANT_CHARGING message = self.PV_CHARGING_MIN_CURRENT_CHARGING return current, sub_mode, message, phases except Exception: log.exception("Fehler im ev-Modul "+str(self.data.id)) - return 0, "stop", "Keine Ladung, da ein interner Fehler aufgetreten ist: "+traceback.format_exc(), 1 + return (0, + ChargemodeEnum.STOP, + "Keine Ladung, da ein interner Fehler aufgetreten ist: "+traceback.format_exc(), + 1) def eco_charging(self, soc: Optional[float], control_parameter: ControlParameter, charging_type: str, used_amount: float, - max_phases_hw: int) -> Tuple[int, str, Optional[str], int]: + max_phases_hw: int) -> Tuple[int, ChargemodeEnum, Optional[str], int]: """ prüft, ob Min-oder Max-Soc erreicht wurden und setzt entsprechend den Ladestrom. """ message = None - sub_mode = "pv_charging" + sub_mode = ChargemodeEnum.PV_CHARGING try: eco_charging = self.data.chargemode.eco_charging phases = eco_charging.phases_to_use @@ -294,16 +304,16 @@ def eco_charging(self, if eco_charging.limit.selected == "soc" and soc is not None and soc >= eco_charging.limit.soc: current = 0 - sub_mode = "stop" + sub_mode = ChargemodeEnum.STOP message = self.SOC_REACHED elif (eco_charging.limit.selected == "amount" and used_amount >= self.data.chargemode.instant_charging.limit.amount): current = 0 - sub_mode = "stop" + sub_mode = ChargemodeEnum.STOP message = self.AMOUNT_REACHED elif data.data.optional_data.data.electricity_pricing.configured: if data.data.optional_data.ep_is_charging_allowed_price_threshold(eco_charging.max_price): - sub_mode = "instant_charging" + sub_mode = ChargemodeEnum.INSTANT_CHARGING message = self.CHARGING_PRICE_LOW phases = max_phases_hw else: @@ -316,7 +326,10 @@ def eco_charging(self, return current, sub_mode, message, phases except Exception: log.exception("Fehler im ev-Modul "+str(self.data.id)) - return 0, "stop", "Keine Ladung, da ein interner Fehler aufgetreten ist: "+traceback.format_exc(), 0 + return (0, + ChargemodeEnum.STOP, + "Keine Ladung, da ein interner Fehler aufgetreten ist: "+traceback.format_exc(), + 0) BUFFER_AFTER_END_TIME = -1200 # nach mehr als 20 Min Überschreitung wird der Termin als verpasst angesehen BUFFER_START_EARLIER = 600 # 10 Min vor dem geplanten Start kann begonnen werden @@ -553,9 +566,9 @@ def scheduled_charging_calc_current(self, charging_type: str, ev_template: EvTemplate, bidi_state: BidiState, - charge_state: bool) -> Tuple[float, str, str, int]: + charge_state: bool) -> Tuple[float, ChargemodeEnum, str, int]: current = 0 - submode = "stop" + submode = ChargemodeEnum.STOP if selected_plan is None: if len(self.data.chargemode.scheduled_charging.plans) == 0: return current, submode, self.SCHEDULED_CHARGING_NO_PLANS_CONFIGURED, control_parameter_phases @@ -585,14 +598,14 @@ def scheduled_charging_calc_current(self, if plan.bidi_charging_enabled and bidi_state == BidiState.BIDI_CAPABLE: message = self.SCHEDULED_CHARGING_BIDI current = min_current - submode = "bidi_charging" + submode = ChargemodeEnum.BIDI_CHARGING phases = control_parameter_phases else: message = self.SCHEDULED_CHARGING_REACHED_SCHEDULED_SOC if plan.bidi_charging_enabled and bidi_state != BidiState.BIDI_CAPABLE: message += bidi_state.value current = min_current - submode = "pv_charging" + submode = ChargemodeEnum.PV_CHARGING # bei Überschuss-Laden mit der Phasenzahl aus den control_parameter laden, # um die Umschaltung zu berücksichtigen. phases = plan.phases_to_use_pv @@ -609,7 +622,7 @@ def scheduled_charging_calc_current(self, limit_string = self.SCHEDULED_CHARGING_LIMITED_BY_AMOUNT.format(limit.amount/1000) message = self.SCHEDULED_CHARGING_IN_TIME.format(plan_current, limit_string, plan.time) current = plan_current - submode = "instant_charging" + submode = ChargemodeEnum.INSTANT_CHARGING # weniger als die berechnete Zeit verfügbar elif selected_plan.remaining_time <= 0 - soc_request_interval_offset: if selected_plan.duration + selected_plan.remaining_time < 0: @@ -618,7 +631,7 @@ def scheduled_charging_calc_current(self, current = min(selected_plan.missing_amount/((selected_plan.duration + selected_plan.remaining_time) / 3600)/(phases*230), max_current) message = self.SCHEDULED_CHARGING_MAX_CURRENT.format(round(current, 2)) - submode = "instant_charging" + submode = ChargemodeEnum.INSTANT_CHARGING else: # Wenn dynamische Tarife aktiv sind, prüfen, ob jetzt ein günstiger Zeitpunkt zum Laden # ist. @@ -671,12 +684,12 @@ def convert_loading_hours_to_string(hour_list: List[int]) -> str: message = self.SCHEDULED_CHARGING_CHEAP_HOUR.format(get_hours_message()) current = plan_current phases = max_phases_hw if plan.phases_to_use == 0 else plan.phases_to_use - submode = "instant_charging" + submode = ChargemodeEnum.INSTANT_CHARGING elif ((limit.selected == "soc" and soc <= limit.soc_limit) or (limit.selected == "amount" and used_amount < limit.amount)): message = self.SCHEDULED_CHARGING_EXPENSIVE_HOUR.format(get_hours_message()) current = min_current - submode = "pv_charging" + submode = ChargemodeEnum.PV_CHARGING phases = plan.phases_to_use_pv else: message = self.SCHEDULED_CHARGING_EXPENSIVE_HOUR_REACHED_MAX_SOC.format(get_hours_message()) @@ -694,12 +707,12 @@ def convert_loading_hours_to_string(hour_list: List[int]) -> str: message = self.SCHEDULED_CHARGING_USE_PV.format( f"am {start_time.strftime('%d.%m')} um {start_time.strftime('%-H:%M')} Uhr") current = min_current - submode = "pv_charging" + submode = ChargemodeEnum.PV_CHARGING phases = plan.phases_to_use_pv return current, submode, message, phases def stop(self) -> Tuple[int, str, str]: - return 0, "stop", "Keine Ladung, da der Lademodus Stop aktiv ist." + return 0, ChargemodeEnum.STOP, "Keine Ladung, da der Lademodus Stop aktiv ist." def bidi_charging_allowed(self, selected_plan: int, soc: float): # Wenn über den Limit-SoC geladen wurde, darf nur noch bidirektional entladen werden. diff --git a/packages/control/ev/charge_template_test.py b/packages/control/ev/charge_template_test.py index 5fc859e02b..652da3e6ea 100644 --- a/packages/control/ev/charge_template_test.py +++ b/packages/control/ev/charge_template_test.py @@ -7,6 +7,7 @@ from control import data from control import optional from control.bat_all import BatAll +from control.chargemode import Chargemode from control.chargepoint.control_parameter import ControlParameter from control.ev.charge_template import SelectedPlan from control.chargepoint.charging_type import ChargingType @@ -28,38 +29,39 @@ def data_module() -> None: @pytest.mark.parametrize( "plans, soc, used_amount_time_charging, plan_found, expected", [ - pytest.param({}, 0, 0, None, (0, "stop", ChargeTemplate.TIME_CHARGING_NO_PLAN_CONFIGURED, None, None), + pytest.param({}, 0, 0, None, (0, Chargemode.STOP, ChargeTemplate.TIME_CHARGING_NO_PLAN_CONFIGURED, None, None), id="no plan defined"), pytest.param({"0": TimeChargingPlan(id=0)}, 0, 0, None, - (0, "stop", ChargeTemplate.TIME_CHARGING_NO_PLAN_ACTIVE, None, None), id="no plan active"), + (0, Chargemode.STOP, ChargeTemplate.TIME_CHARGING_NO_PLAN_ACTIVE, None, None), + id="no plan active"), pytest.param({"0": TimeChargingPlan(id=0)}, 0, 0, TimeChargingPlan(id=0), - (16, "time_charging", None, 0, 1), id="plan active"), + (16, Chargemode.TIME_CHARGING, None, 0, 1), id="plan active"), pytest.param({"0": TimeChargingPlan(id=0, limit=Limit(selected="soc"))}, 100, 0, TimeChargingPlan(id=0, limit=Limit(selected="soc")), - (0, "stop", ChargeTemplate.TIME_CHARGING_SOC_REACHED, 0, 1), + (0, Chargemode.STOP, ChargeTemplate.TIME_CHARGING_SOC_REACHED, 0, 1), id="plan active, soc is reached"), pytest.param({"0": TimeChargingPlan(id=0, limit=Limit(selected="soc"))}, 40, 0, TimeChargingPlan(id=0, limit=Limit(selected="soc")), - (16, "time_charging", None, 0, 1), id="plan active, soc is not reached"), + (16, Chargemode.TIME_CHARGING, None, 0, 1), id="plan active, soc is not reached"), pytest.param({"0": TimeChargingPlan(id=0, limit=Limit(selected="soc"))}, None, 0, TimeChargingPlan(id=0, limit=Limit(selected="soc")), - (16, "time_charging", None, 0, 1), id="plan active, soc is not defined"), + (16, Chargemode.TIME_CHARGING, None, 0, 1), id="plan active, soc is not defined"), pytest.param({"0": TimeChargingPlan(id=0, limit=Limit(selected="amount", amount=1000))}, 0, 1500, TimeChargingPlan(id=0, limit=Limit(selected="amount", amount=1000)), - (0, "stop", ChargeTemplate.TIME_CHARGING_AMOUNT_REACHED, 0, 1), + (0, Chargemode.STOP, ChargeTemplate.TIME_CHARGING_AMOUNT_REACHED, 0, 1), id="plan active, used_amount_time_charging is reached"), pytest.param({"0": TimeChargingPlan(id=0, limit=Limit(selected="amount", amount=1000))}, 0, 500, TimeChargingPlan(id=0, limit=Limit(selected="amount", amount=1000)), - (16, "time_charging", None, 0, 1), + (16, Chargemode.TIME_CHARGING, None, 0, 1), id="plan active, used_amount_time_charging is not reached"), pytest.param({"0": TimeChargingPlan(id=0)}, 0, 0, None, - (0, "stop", ChargeTemplate.TIME_CHARGING_NO_PLAN_ACTIVE, None, None), + (0, Chargemode.STOP, ChargeTemplate.TIME_CHARGING_NO_PLAN_ACTIVE, None, None), id="plan defined but not found"), ] ) def test_time_charging(plans: Dict[int, TimeChargingPlan], soc: float, used_amount_time_charging: float, plan_found: TimeChargingPlan, - expected: Tuple[int, str, Optional[str], Optional[str]], + expected: Tuple[int, Chargemode, Optional[str], Optional[str], Optional[int]], monkeypatch): # setup ct = ChargeTemplate() @@ -77,19 +79,20 @@ def test_time_charging(plans: Dict[int, TimeChargingPlan], soc: float, used_amou @pytest.mark.parametrize( "min_bat_soc, charging_allowed_mock, soc, expected", [ - pytest.param(80, False, 81, (0, "stop", ChargeTemplate.TIME_CHARGING_CONFLICT_ACTIVE_BAT_CONTROL, 0, 1), + pytest.param(80, False, 81, + (0, Chargemode.STOP, ChargeTemplate.TIME_CHARGING_CONFLICT_ACTIVE_BAT_CONTROL, 0, 1), id="Konflikt mit aktiver Speichersteuerung -> nicht laden"), - pytest.param(80, True, 79, (0, "stop", ChargeTemplate.TIME_CHARGING_MIN_BAT_SOC_REACHED, 0, 1), + pytest.param(80, True, 79, (0, Chargemode.STOP, ChargeTemplate.TIME_CHARGING_MIN_BAT_SOC_REACHED, 0, 1), id="Mindest-SoC des Speichers unterschritten -> nicht laden"), - pytest.param(80, True, 80, (16, "time_charging", None, 0, 1), id="laden erlaubt"), - pytest.param(None, True, 80, (16, "time_charging", None, 0, 1), + pytest.param(80, True, 80, (16, Chargemode.TIME_CHARGING, None, 0, 1), id="laden erlaubt"), + pytest.param(None, True, 80, (16, Chargemode.TIME_CHARGING, None, 0, 1), id="Mindest-SoC-Beachtung nicht konfiguriert, laden erlaubt"), ] ) def test_time_charging_min_bat_soc(min_bat_soc: Optional[int], charging_allowed_mock: bool, soc: float, - expected: Tuple[int, str, Optional[str], Optional[str], int], + expected: Tuple[int, Chargemode, Optional[str], Optional[str], int], monkeypatch: pytest.MonkeyPatch): # setup ct = ChargeTemplate() @@ -115,17 +118,18 @@ def test_time_charging_min_bat_soc(min_bat_soc: Optional[int], @pytest.mark.parametrize( "selected, current_soc, used_amount, expected", [ - pytest.param("none", 0, 0, (16, "instant_charging", None, 3), id="without limit"), - pytest.param("soc", None, 0, (16, "instant_charging", None, 3), id="limit soc: soc not defined"), - pytest.param("soc", 49, 0, (16, "instant_charging", None, 3), id="limit soc: soc not reached"), - pytest.param("soc", 50, 0, (0, "stop", ChargeTemplate.SOC_REACHED, 3), + pytest.param("none", 0, 0, (16, Chargemode.INSTANT_CHARGING, None, 3), id="without limit"), + pytest.param("soc", None, 0, (16, Chargemode.INSTANT_CHARGING, None, 3), id="limit soc: soc not defined"), + pytest.param("soc", 49, 0, (16, Chargemode.INSTANT_CHARGING, None, 3), id="limit soc: soc not reached"), + pytest.param("soc", 50, 0, (0, Chargemode.STOP, ChargeTemplate.SOC_REACHED, 3), id="limit soc: soc reached"), - pytest.param("amount", 0, 999, (16, "instant_charging", None, 3), id="limit amount: amount not reached"), - pytest.param("amount", 0, 1000, (0, "stop", ChargeTemplate.AMOUNT_REACHED, 3), + pytest.param("amount", 0, 999, (16, Chargemode.INSTANT_CHARGING, + None, 3), id="limit amount: amount not reached"), + pytest.param("amount", 0, 1000, (0, Chargemode.STOP, ChargeTemplate.AMOUNT_REACHED, 3), id="limit amount: amount reached"), ]) def test_instant_charging(selected: str, current_soc: float, used_amount: float, - expected: Tuple[int, str, Optional[str]]): + expected: Tuple[int, Chargemode, Optional[str], int]): # setup ct = ChargeTemplate() ct.data.chargemode.instant_charging.limit.selected = selected @@ -141,21 +145,24 @@ def test_instant_charging(selected: str, current_soc: float, used_amount: float, @pytest.mark.parametrize( "min_soc, min_current, limit_selected, current_soc, used_amount, expected", [ - pytest.param(0, 0, "amount", 14, 1500, (0, "stop", ChargeTemplate.AMOUNT_REACHED, 0), id="max amount reached"), - pytest.param(0, 0, "soc", 100, 900, (0, "stop", ChargeTemplate.SOC_REACHED, 0), id="max soc reached"), - pytest.param(15, 0, None, 14, 900, (10, "instant_charging", ChargeTemplate.PV_CHARGING_SOC_CHARGING, 3), + pytest.param(0, 0, "amount", 14, 1500, (0, Chargemode.STOP, + ChargeTemplate.AMOUNT_REACHED, 0), id="max amount reached"), + pytest.param(0, 0, "soc", 100, 900, (0, Chargemode.STOP, ChargeTemplate.SOC_REACHED, 0), id="max soc reached"), + pytest.param(15, 0, None, 14, 900, + (10, Chargemode.INSTANT_CHARGING, ChargeTemplate.PV_CHARGING_SOC_CHARGING, 3), id="min soc not reached"), - pytest.param(15, 0, None, None, 900, (6, "pv_charging", None, 0), id="soc not defined"), - pytest.param(15, 8, None, 15, 900, (8, "instant_charging", ChargeTemplate.PV_CHARGING_MIN_CURRENT_CHARGING, 0), + pytest.param(15, 0, None, None, 900, (6, Chargemode.PV_CHARGING, None, 0), id="soc not defined"), + pytest.param(15, 8, None, 15, 900, + (8, Chargemode.INSTANT_CHARGING, ChargeTemplate.PV_CHARGING_MIN_CURRENT_CHARGING, 0), id="min current configured"), - pytest.param(15, 0, None, 15, 900, (6, "pv_charging", None, 0), id="bare pv charging"), + pytest.param(15, 0, None, 15, 900, (6, Chargemode.PV_CHARGING, None, 0), id="bare pv charging"), ]) def test_pv_charging(min_soc: int, min_current: int, limit_selected: str, current_soc: float, used_amount: float, - expected: Tuple[int, str, Optional[str], int]): + expected: Tuple[int, Chargemode, Optional[str], int]): # setup ct = ChargeTemplate() ct.data.chargemode.pv_charging.min_soc = min_soc @@ -267,34 +274,35 @@ def test_scheduled_charging_recent_plan(end_time_mock, @pytest.mark.parametrize( "plan_data, soc, used_amount, selected, bidi_charging_enabled, expected", [ - pytest.param(None, 0, 0, "none", False, (0, "stop", + pytest.param(None, 0, 0, "none", False, (0, Chargemode.STOP, ChargeTemplate.SCHEDULED_CHARGING_NO_DATE_PENDING, 3), id="no date pending"), - pytest.param(SelectedPlan(duration=3600), 90, 0, "soc", False, (0, "stop", + pytest.param(SelectedPlan(duration=3600), 90, 0, "soc", False, (0, Chargemode.STOP, ChargeTemplate.SCHEDULED_CHARGING_REACHED_MAX_AND_LIMIT_SOC, 1), id="reached limit soc"), - pytest.param(SelectedPlan(duration=3600), 80, 0, "soc", False, (6, "pv_charging", + pytest.param(SelectedPlan(duration=3600), 80, 0, "soc", False, (6, Chargemode.PV_CHARGING, ChargeTemplate.SCHEDULED_CHARGING_REACHED_SCHEDULED_SOC, 0), id="reached scheduled soc"), - pytest.param(SelectedPlan(duration=3600), 80, 0, "soc", True, (6, "bidi_charging", + pytest.param(SelectedPlan(duration=3600), 80, 0, "soc", True, (6, Chargemode.BIDI_CHARGING, ChargeTemplate.SCHEDULED_CHARGING_BIDI, 3), id="reached scheduled soc, bidi"), - pytest.param(SelectedPlan(phases=3, duration=3600), 0, 1000, "amount", False, (0, "stop", + pytest.param(SelectedPlan(phases=3, duration=3600), 0, 1000, "amount", False, (0, Chargemode.STOP, ChargeTemplate.SCHEDULED_CHARGING_REACHED_AMOUNT, 3), id="reached amount"), pytest.param(SelectedPlan(remaining_time=299, duration=3600), 0, 999, "amount", - False, (14, "instant_charging", ChargeTemplate.SCHEDULED_CHARGING_IN_TIME.format( + False, (14, Chargemode.INSTANT_CHARGING, ChargeTemplate.SCHEDULED_CHARGING_IN_TIME.format( 14, ChargeTemplate.SCHEDULED_CHARGING_LIMITED_BY_AMOUNT.format(1.0), "07:00"), 1), id="in time, limited by amount"), pytest.param(SelectedPlan(remaining_time=299, duration=3600), 79, 0, "soc", - False, (14, "instant_charging", ChargeTemplate.SCHEDULED_CHARGING_IN_TIME.format( + False, (14, Chargemode.INSTANT_CHARGING, ChargeTemplate.SCHEDULED_CHARGING_IN_TIME.format( 14, ChargeTemplate.SCHEDULED_CHARGING_LIMITED_BY_SOC.format(80), "07:00"), 1), id="in time, limited by soc"), pytest.param(SelectedPlan(remaining_time=-500, duration=3600, missing_amount=9000, phases=3), 79, 0, "soc", - False, (15.147265077138847, "instant_charging", + False, (15.147265077138847, Chargemode.INSTANT_CHARGING, ChargeTemplate.SCHEDULED_CHARGING_MAX_CURRENT.format(15.15), 3), id="too late, but didn't miss for today"), pytest.param(SelectedPlan(remaining_time=-800, duration=780, missing_amount=4600, phases=3), 79, 0, "soc", - False, (16, "instant_charging", + False, (16, Chargemode.INSTANT_CHARGING, ChargeTemplate.SCHEDULED_CHARGING_MAX_CURRENT.format(16), 3), id="few minutes too late, but didn't miss for today"), pytest.param(SelectedPlan(remaining_time=601, duration=3600), 79, 0, "soc", - False, (6, "pv_charging", ChargeTemplate.SCHEDULED_CHARGING_USE_PV.format("um 8:50 Uhr"), 0), + False, + (6, Chargemode.PV_CHARGING, ChargeTemplate.SCHEDULED_CHARGING_USE_PV.format("um 8:50 Uhr"), 0), id="too early, use pv"), ]) def test_scheduled_charging_calc_current(plan_data: SelectedPlan, @@ -302,7 +310,7 @@ def test_scheduled_charging_calc_current(plan_data: SelectedPlan, used_amount: float, selected: str, bidi_charging_enabled: bool, - expected: Tuple[float, str, str, int]): + expected: Tuple[float, Chargemode, str, int]): # setup ct = ChargeTemplate() plan = ScheduledChargingPlan(active=True, id=0) @@ -330,28 +338,28 @@ def test_scheduled_charging_calc_current_no_plans(): None, 63, 5, 3, 3, 6, 0, ChargingType.AC.value, EvTemplate(), BidiState.BIDI_CAPABLE, True) # evaluation - assert ret == (0, "stop", ChargeTemplate.SCHEDULED_CHARGING_NO_PLANS_CONFIGURED, 3) + assert ret == (0, Chargemode.STOP, ChargeTemplate.SCHEDULED_CHARGING_NO_PLANS_CONFIGURED, 3) @pytest.mark.parametrize( "selected_limit, charge_state, expected", [ pytest.param("soc", False, - (6, "pv_charging", ChargeTemplate.SCHEDULED_CHARGING_USE_PV.format("um 8:50 Uhr"), 0), + (6, Chargemode.PV_CHARGING, ChargeTemplate.SCHEDULED_CHARGING_USE_PV.format("um 8:50 Uhr"), 0), id="soc request interval not considered, use pv"), pytest.param("soc", True, - (14, "instant_charging", + (14, Chargemode.INSTANT_CHARGING, ChargeTemplate.SCHEDULED_CHARGING_IN_TIME.format(14, 'einen SoC von 80%', "07:00"), 3), id="soc request interval considered"), pytest.param("amount", True, - (6, "pv_charging", ChargeTemplate.SCHEDULED_CHARGING_USE_PV.format("um 8:50 Uhr"), 0), + (6, Chargemode.PV_CHARGING, ChargeTemplate.SCHEDULED_CHARGING_USE_PV.format("um 8:50 Uhr"), 0), id="amount, charging, soc request interval not considered, no soc limit configured"), pytest.param("amount", False, - (6, "pv_charging", ChargeTemplate.SCHEDULED_CHARGING_USE_PV.format("um 8:50 Uhr"), 0), + (6, Chargemode.PV_CHARGING, ChargeTemplate.SCHEDULED_CHARGING_USE_PV.format("um 8:50 Uhr"), 0), id="amount, not charging, soc request interval not considered, no soc limit configured"), ]) def test_scheduled_charging_calc_current_consider_soc_request_interval_offset( - selected_limit: str, charge_state: bool, expected: Tuple[float, str, str, int]): + selected_limit: str, charge_state: bool, expected: Tuple[float, Chargemode, str, int]): # setup ct = ChargeTemplate() plan = ScheduledChargingPlan() @@ -379,7 +387,7 @@ def test_scheduled_charging_calc_current_consider_soc_request_interval_offset( pytest.param(True, 79, 80, 90, LOADING_HOURS_TODAY + LOADING_HOURS_TOMORROW, ( 14, - "instant_charging", + Chargemode.INSTANT_CHARGING, ChargeTemplate.SCHEDULED_CHARGING_CHEAP_HOUR.format( "Geladen wird jetzt sowie morgen 8:00."), 3), @@ -387,7 +395,7 @@ def test_scheduled_charging_calc_current_consider_soc_request_interval_offset( pytest.param(True, 79, 80, 70, LOADING_HOURS_TODAY, ( 14, - "instant_charging", + Chargemode.INSTANT_CHARGING, ChargeTemplate.SCHEDULED_CHARGING_CHEAP_HOUR.format( "Geladen wird jetzt."), 3), @@ -395,7 +403,7 @@ def test_scheduled_charging_calc_current_consider_soc_request_interval_offset( pytest.param(False, 79, 80, 90, LOADING_HOURS_TODAY, ( 6, - "pv_charging", + Chargemode.PV_CHARGING, ChargeTemplate.SCHEDULED_CHARGING_EXPENSIVE_HOUR.format( "Geladen wird heute 8:00."), 0), @@ -403,7 +411,7 @@ def test_scheduled_charging_calc_current_consider_soc_request_interval_offset( pytest.param(False, 79, 80, 70, LOADING_HOURS_TODAY, ( 0, - "stop", + Chargemode.STOP, ChargeTemplate.SCHEDULED_CHARGING_EXPENSIVE_HOUR_REACHED_MAX_SOC.format( "Geladen wird heute 8:00."), 3), @@ -411,7 +419,7 @@ def test_scheduled_charging_calc_current_consider_soc_request_interval_offset( pytest.param(False, 79, 80, 70, LOADING_HOURS_TODAY + LOADING_HOURS_TOMORROW, ( 0, - "stop", + Chargemode.STOP, ChargeTemplate.SCHEDULED_CHARGING_EXPENSIVE_HOUR_REACHED_MAX_SOC.format( "Geladen wird heute 8:00 sowie morgen 8:00."), 3), @@ -419,7 +427,7 @@ def test_scheduled_charging_calc_current_consider_soc_request_interval_offset( pytest.param(False, 79, 60, 80, LOADING_HOURS_TODAY, ( 6, - "pv_charging", + Chargemode.PV_CHARGING, ChargeTemplate.SCHEDULED_CHARGING_REACHED_SCHEDULED_SOC.format( ""), 0), @@ -427,7 +435,7 @@ def test_scheduled_charging_calc_current_consider_soc_request_interval_offset( pytest.param(False, 79, 60, 50, LOADING_HOURS_TODAY, ( 0, - "stop", + Chargemode.STOP, ChargeTemplate.SCHEDULED_CHARGING_REACHED_MAX_AND_LIMIT_SOC.format( ""), 3), diff --git a/packages/control/ev/ev.py b/packages/control/ev/ev.py index fab9167328..16af34971d 100644 --- a/packages/control/ev/ev.py +++ b/packages/control/ev/ev.py @@ -6,6 +6,7 @@ stärke wird auch geprüft, ob sich an diesen Parametern etwas geändert hat. Falls ja, muss das EV in der Regelung neu priorisiert werden und eine neue Zuteilung des Stroms erhalten. """ +from control.chargemode import Chargemode from modules.common.configurable_vehicle import ConfigurableVehicle from modules.common.abstract_vehicle import VehicleUpdateData from helpermodules.constants import DEFAULT_COLORS, NO_ERROR @@ -125,7 +126,7 @@ def get_required_current(self, charging_type: str, imported_since_plugged: float, bidi: BidiState, - charge_state: bool) -> Tuple[bool, Optional[str], str, float, int]: + charge_state: bool) -> Tuple[bool, Optional[str], Chargemode, float, int]: """ ermittelt, ob und mit welchem Strom das EV geladen werden soll (unabhängig vom Lastmanagement) Parameter @@ -150,7 +151,7 @@ def get_required_current(self, tmp_message = None state = True try: - if charge_template.data.chargemode.selected == "stop": + if charge_template.data.chargemode.selected == Chargemode.STOP.value: required_current, submode, message = charge_template.stop() phases = control_parameter.phases or max_phases_hw else: @@ -160,7 +161,7 @@ def get_required_current(self, soc_request_interval_offset = self.soc_module.general_config.request_interval_charging else: soc_request_interval_offset = 0 - if charge_template.data.chargemode.selected == "scheduled_charging": + if charge_template.data.chargemode.selected == Chargemode.SCHEDULED_CHARGING.value: required_current, submode, tmp_message, phases = charge_template.scheduled_charging( self.data.get.soc, self.ev_template, @@ -175,7 +176,7 @@ def get_required_current(self, message = f"{tmp_message or ''}".strip() # Wenn Zielladen auf Überschuss wartet, prüfen, ob Zeitladen aktiv ist. - if (submode != "instant_charging" and + if (submode != Chargemode.INSTANT_CHARGING and charge_template.data.time_charging.active): tmp_current, tmp_submode, tmp_message, plan_id, tmp_phases = charge_template.time_charging( self.data.get.soc, @@ -190,21 +191,21 @@ def get_required_current(self, submode = tmp_submode phases = tmp_phases if (required_current == 0) or (required_current is None): - if charge_template.data.chargemode.selected == "instant_charging": + if charge_template.data.chargemode.selected == Chargemode.INSTANT_CHARGING.value: required_current, submode, tmp_message, phases = charge_template.instant_charging( self.data.get.soc, imported_since_plugged, charging_type) - elif charge_template.data.chargemode.selected == "pv_charging": + elif charge_template.data.chargemode.selected == Chargemode.PV_CHARGING.value: required_current, submode, tmp_message, phases = charge_template.pv_charging( self.data.get.soc, control_parameter.min_current, charging_type, imported_since_plugged) - elif charge_template.data.chargemode.selected == "eco_charging": + elif charge_template.data.chargemode.selected == Chargemode.ECO_CHARGING.value: required_current, submode, tmp_message, phases = charge_template.eco_charging( self.data.get.soc, control_parameter, charging_type, imported_since_plugged, max_phases_hw) else: tmp_message = None message = f"{message or ''} {tmp_message or ''}".strip() - if submode == "stop" or (charge_template.data.chargemode.selected == "stop"): + if submode == Chargemode.STOP or (charge_template.data.chargemode.selected == Chargemode.STOP.value): state = False if phases is None: log.debug("Keine Phasenvorgabe durch Lademodus. Behalte Phasenzahl bei.") @@ -212,7 +213,7 @@ def get_required_current(self, return state, message, submode, required_current, phases except Exception as e: log.exception("Fehler im ev-Modul "+str(self.num)) - return (False, f"Kein Ladevorgang, da ein Fehler aufgetreten ist: {' '.join(e.args)}", "stop", 0, + return (False, f"Kein Ladevorgang, da ein Fehler aufgetreten ist: {' '.join(e.args)}", Chargemode.STOP, 0, control_parameter.phases) def check_min_max_current(self, diff --git a/packages/helpermodules/subdata.py b/packages/helpermodules/subdata.py index 884ad3e0f4..0f03416ed7 100644 --- a/packages/helpermodules/subdata.py +++ b/packages/helpermodules/subdata.py @@ -1,5 +1,6 @@ """ Modul, um die Daten vom Broker zu erhalten. """ +from enum import Enum import importlib import logging from pathlib import Path @@ -246,6 +247,8 @@ def set_json_payload_class(self, class_obj: Dict, msg: mqtt.MQTTMessage): if isinstance(payload, Dict): for key, value in payload.items(): setattr(class_obj, key, value) + elif isinstance(getattr(class_obj, key, None), Enum): + setattr(class_obj, key, type(getattr(class_obj, key))(payload)) else: setattr(class_obj, key, decode_payload(msg.payload)) else: From b26677ab64eb4282d13b6e47a109429bacf64a59 Mon Sep 17 00:00:00 2001 From: LKuemmel Date: Wed, 19 Aug 2026 16:06:16 +0200 Subject: [PATCH 2/7] fix --- packages/control/chargepoint/chargepoint.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/packages/control/chargepoint/chargepoint.py b/packages/control/chargepoint/chargepoint.py index 5975ae32ab..6a4beff437 100644 --- a/packages/control/chargepoint/chargepoint.py +++ b/packages/control/chargepoint/chargepoint.py @@ -606,7 +606,7 @@ def set_chargemode_changed(self, submode: Chargemode) -> None: if ((submode == Chargemode.TIME_CHARGING and self.data.control_parameter.chargemode != Chargemode.TIME_CHARGING) or (submode != Chargemode.TIME_CHARGING and - self.data.control_parameter.chargemode != self.data.set.charge_template.data.chargemode.selected)): + self.data.control_parameter.chargemode != Chargemode(self.data.set.charge_template.data.chargemode.selected))): self.chargemode_changed = True log.debug("Änderung des Lademodus") self.data.control_parameter.timestamp_chargemode_changed = create_timestamp() From fa1e001d5c0ceffccfc44b85c1afc0f6f779fab6 Mon Sep 17 00:00:00 2001 From: LKuemmel Date: Wed, 19 Aug 2026 16:10:21 +0200 Subject: [PATCH 3/7] flake8 --- packages/control/chargepoint/chargepoint.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/packages/control/chargepoint/chargepoint.py b/packages/control/chargepoint/chargepoint.py index 6a4beff437..014fd87042 100644 --- a/packages/control/chargepoint/chargepoint.py +++ b/packages/control/chargepoint/chargepoint.py @@ -606,7 +606,8 @@ def set_chargemode_changed(self, submode: Chargemode) -> None: if ((submode == Chargemode.TIME_CHARGING and self.data.control_parameter.chargemode != Chargemode.TIME_CHARGING) or (submode != Chargemode.TIME_CHARGING and - self.data.control_parameter.chargemode != Chargemode(self.data.set.charge_template.data.chargemode.selected))): + self.data.control_parameter.chargemode != Chargemode( + self.data.set.charge_template.data.chargemode.selected))): self.chargemode_changed = True log.debug("Änderung des Lademodus") self.data.control_parameter.timestamp_chargemode_changed = create_timestamp() From b2c867db446a147089eb48498672e53e0810da69 Mon Sep 17 00:00:00 2001 From: LKuemmel Date: Wed, 19 Aug 2026 16:10:35 +0200 Subject: [PATCH 4/7] flake8 --- packages/control/chargepoint/chargepoint.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/packages/control/chargepoint/chargepoint.py b/packages/control/chargepoint/chargepoint.py index 014fd87042..0ba184310e 100644 --- a/packages/control/chargepoint/chargepoint.py +++ b/packages/control/chargepoint/chargepoint.py @@ -603,11 +603,11 @@ def set_timestamp_charge_start(self): self.data.control_parameter.timestamp_charge_start = None def set_chargemode_changed(self, submode: Chargemode) -> None: + selected_chargemode = Chargemode(self.data.set.charge_template.data.chargemode.selected) if ((submode == Chargemode.TIME_CHARGING and self.data.control_parameter.chargemode != Chargemode.TIME_CHARGING) or (submode != Chargemode.TIME_CHARGING and - self.data.control_parameter.chargemode != Chargemode( - self.data.set.charge_template.data.chargemode.selected))): + self.data.control_parameter.chargemode != selected_chargemode)): self.chargemode_changed = True log.debug("Änderung des Lademodus") self.data.control_parameter.timestamp_chargemode_changed = create_timestamp() From 0e9aca32df017fb8f6e77326a05080069e8c3862 Mon Sep 17 00:00:00 2001 From: LKuemmel Date: Thu, 20 Aug 2026 09:38:02 +0200 Subject: [PATCH 5/7] fix amount --- packages/control/ev/charge_template.py | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/packages/control/ev/charge_template.py b/packages/control/ev/charge_template.py index fee8e20d78..435f809fe3 100644 --- a/packages/control/ev/charge_template.py +++ b/packages/control/ev/charge_template.py @@ -306,8 +306,7 @@ def eco_charging(self, current = 0 sub_mode = ChargemodeEnum.STOP message = self.SOC_REACHED - elif (eco_charging.limit.selected == "amount" and - used_amount >= self.data.chargemode.instant_charging.limit.amount): + elif eco_charging.limit.selected == "amount" and used_amount >= eco_charging.limit.amount: current = 0 sub_mode = ChargemodeEnum.STOP message = self.AMOUNT_REACHED From 448212de06ddc747bab9e73515eb2bee5d28174d Mon Sep 17 00:00:00 2001 From: LKuemmel Date: Thu, 20 Aug 2026 10:08:58 +0200 Subject: [PATCH 6/7] use enum also for chargemode selected --- packages/control/chargepoint/chargepoint.py | 10 ++++------ packages/control/chargepoint/chargepoint_data.py | 3 ++- packages/control/ev/charge_template.py | 2 +- packages/control/ev/ev.py | 12 ++++++------ 4 files changed, 13 insertions(+), 14 deletions(-) diff --git a/packages/control/chargepoint/chargepoint.py b/packages/control/chargepoint/chargepoint.py index 0ba184310e..9f90092234 100644 --- a/packages/control/chargepoint/chargepoint.py +++ b/packages/control/chargepoint/chargepoint.py @@ -231,8 +231,7 @@ def set_control_parameter(self, submode: Chargemode): if submode == Chargemode.TIME_CHARGING: self.data.control_parameter.chargemode = Chargemode.TIME_CHARGING else: - self.data.control_parameter.chargemode = Chargemode( - self.data.set.charge_template.data.chargemode.selected) + self.data.control_parameter.chargemode = self.data.set.charge_template.data.chargemode.selected self.data.control_parameter.prio = self.data.set.charge_template.data.prio if self.template.data.charging_type == ChargingType.AC.value: self.data.control_parameter.min_current = self.data.set.charging_ev_data.ev_template.data.min_current @@ -603,11 +602,10 @@ def set_timestamp_charge_start(self): self.data.control_parameter.timestamp_charge_start = None def set_chargemode_changed(self, submode: Chargemode) -> None: - selected_chargemode = Chargemode(self.data.set.charge_template.data.chargemode.selected) if ((submode == Chargemode.TIME_CHARGING and self.data.control_parameter.chargemode != Chargemode.TIME_CHARGING) or (submode != Chargemode.TIME_CHARGING and - self.data.control_parameter.chargemode != selected_chargemode)): + self.data.control_parameter.chargemode != self.data.set.charge_template.data.chargemode.selected)): self.chargemode_changed = True log.debug("Änderung des Lademodus") self.data.control_parameter.timestamp_chargemode_changed = create_timestamp() @@ -809,8 +807,8 @@ def _pub_connected_vehicle(self, vehicle: Ev): self.data.get.connected_vehicle.soc.range = vehicle.data.get.range self.data.get.connected_vehicle.info = ConnectedInfo(id=vehicle.num, name=vehicle.data.name) - if (self.data.set.charge_template.data.chargemode.selected == Chargemode.TIME_CHARGING.value or - self.data.set.charge_template.data.chargemode.selected == Chargemode.SCHEDULED_CHARGING.value): + if (self.data.set.charge_template.data.chargemode.selected == Chargemode.TIME_CHARGING or + self.data.set.charge_template.data.chargemode.selected == Chargemode.SCHEDULED_CHARGING): current_plan = self.data.control_parameter.current_plan else: current_plan = None diff --git a/packages/control/chargepoint/chargepoint_data.py b/packages/control/chargepoint/chargepoint_data.py index 0f41753cfb..429d20bdc3 100644 --- a/packages/control/chargepoint/chargepoint_data.py +++ b/packages/control/chargepoint/chargepoint_data.py @@ -1,6 +1,7 @@ from dataclasses import dataclass, field from threading import Event from typing import Dict, List, Optional, Protocol +from control.chargemode import Chargemode from control.chargepoint.chargepoint_template import CpTemplate from control.chargepoint.control_parameter import ControlParameter, control_parameter_factory @@ -38,7 +39,7 @@ class ConnectedInfo: class ConnectedConfig: average_consumption: float = 17 charge_template: int = 0 - chargemode: str = "stop" + chargemode: Chargemode = Chargemode.STOP current_plan: Optional[int] = 0 ev_template: int = 0 priority: bool = False diff --git a/packages/control/ev/charge_template.py b/packages/control/ev/charge_template.py index 435f809fe3..6c53430dce 100644 --- a/packages/control/ev/charge_template.py +++ b/packages/control/ev/charge_template.py @@ -92,7 +92,7 @@ def instant_charging_factory() -> InstantCharging: @dataclass class Chargemode: - selected: str = ChargemodeEnum.INSTANT_CHARGING.value + selected: ChargemodeEnum = ChargemodeEnum.INSTANT_CHARGING eco_charging: EcoCharging = field(default_factory=eco_charging_factory) pv_charging: PvCharging = field(default_factory=pv_charging_factory) scheduled_charging: ScheduledCharging = field(default_factory=scheduled_charging_factory) diff --git a/packages/control/ev/ev.py b/packages/control/ev/ev.py index 16af34971d..c206812cd6 100644 --- a/packages/control/ev/ev.py +++ b/packages/control/ev/ev.py @@ -151,7 +151,7 @@ def get_required_current(self, tmp_message = None state = True try: - if charge_template.data.chargemode.selected == Chargemode.STOP.value: + if charge_template.data.chargemode.selected == Chargemode.STOP: required_current, submode, message = charge_template.stop() phases = control_parameter.phases or max_phases_hw else: @@ -161,7 +161,7 @@ def get_required_current(self, soc_request_interval_offset = self.soc_module.general_config.request_interval_charging else: soc_request_interval_offset = 0 - if charge_template.data.chargemode.selected == Chargemode.SCHEDULED_CHARGING.value: + if charge_template.data.chargemode.selected == Chargemode.SCHEDULED_CHARGING: required_current, submode, tmp_message, phases = charge_template.scheduled_charging( self.data.get.soc, self.ev_template, @@ -191,21 +191,21 @@ def get_required_current(self, submode = tmp_submode phases = tmp_phases if (required_current == 0) or (required_current is None): - if charge_template.data.chargemode.selected == Chargemode.INSTANT_CHARGING.value: + if charge_template.data.chargemode.selected == Chargemode.INSTANT_CHARGING: required_current, submode, tmp_message, phases = charge_template.instant_charging( self.data.get.soc, imported_since_plugged, charging_type) - elif charge_template.data.chargemode.selected == Chargemode.PV_CHARGING.value: + elif charge_template.data.chargemode.selected == Chargemode.PV_CHARGING: required_current, submode, tmp_message, phases = charge_template.pv_charging( self.data.get.soc, control_parameter.min_current, charging_type, imported_since_plugged) - elif charge_template.data.chargemode.selected == Chargemode.ECO_CHARGING.value: + elif charge_template.data.chargemode.selected == Chargemode.ECO_CHARGING: required_current, submode, tmp_message, phases = charge_template.eco_charging( self.data.get.soc, control_parameter, charging_type, imported_since_plugged, max_phases_hw) else: tmp_message = None message = f"{message or ''} {tmp_message or ''}".strip() - if submode == Chargemode.STOP or (charge_template.data.chargemode.selected == Chargemode.STOP.value): + if submode == Chargemode.STOP or (charge_template.data.chargemode.selected == Chargemode.STOP): state = False if phases is None: log.debug("Keine Phasenvorgabe durch Lademodus. Behalte Phasenzahl bei.") From e2b69866602c208a6df4faade4ffdf7498d36af6 Mon Sep 17 00:00:00 2001 From: LKuemmel <76958050+LKuemmel@users.noreply.github.com> Date: Mon, 24 Aug 2026 13:55:08 +0200 Subject: [PATCH 7/7] fix asdict import (#3847) * use chargemode enums for chargepoints * fix * flake8 * flake8 * fix amount * use enum also for chargemode selected * fix asdict import * review --- packages/control/chargepoint/chargepoint.py | 6 ++-- .../chargepoint/chargepoint_template.py | 3 +- packages/control/ev/charge_template.py | 3 +- packages/dataclass_utils/_dataclass_asdict.py | 14 ++++++--- .../helpermodules/changed_values_handler.py | 2 +- .../changed_values_handler_test.py | 3 +- packages/helpermodules/command.py | 31 +++++++++---------- packages/helpermodules/setdata.py | 3 +- packages/helpermodules/update_config.py | 24 +++++++------- .../backup_clouds/nextcloud/backup_cloud.py | 2 +- packages/modules/devices/avm/avm/device.py | 2 +- packages/modules/io_devices/eebus/api.py | 2 +- 12 files changed, 48 insertions(+), 47 deletions(-) diff --git a/packages/control/chargepoint/chargepoint.py b/packages/control/chargepoint/chargepoint.py index 9f90092234..4b0252c254 100644 --- a/packages/control/chargepoint/chargepoint.py +++ b/packages/control/chargepoint/chargepoint.py @@ -1,5 +1,3 @@ -from dataclasses import asdict -import dataclasses import logging from threading import Thread, Event import traceback @@ -21,6 +19,7 @@ from control.chargepoint.chargepoint_state import CHARGING_STATES, ChargepointState from control.limiting_value import loadmanagement_limit_factory from control.text import BidiState +from dataclass_utils import asdict from helpermodules.constants import DEFAULT_COLORS from helpermodules.phase_handling import convert_single_evu_phase_to_cp_phase from helpermodules.pub import Pub @@ -781,8 +780,7 @@ def _get_charging_ev(self, vehicle: int, ev_list: Dict[str, Ev]) -> Ev: def update_charge_template(self, charge_template: ChargeTemplate) -> None: # Prüfen, ob ein temporäres Ladeprofil aktiv ist und dieses übernehmen self.data.set.charge_template = charge_template - Pub().pub(f"openWB/set/chargepoint/{self.num}/set/charge_template", - dataclasses.asdict(charge_template.data)) + Pub().pub(f"openWB/set/chargepoint/{self.num}/set/charge_template", asdict(charge_template.data)) def _pub_connected_vehicle(self, vehicle: Ev): """ published die Daten, die zur Anzeige auf der Hauptseite benötigt werden. diff --git a/packages/control/chargepoint/chargepoint_template.py b/packages/control/chargepoint/chargepoint_template.py index b8cf05ba40..fd10a30e85 100644 --- a/packages/control/chargepoint/chargepoint_template.py +++ b/packages/control/chargepoint/chargepoint_template.py @@ -1,4 +1,4 @@ -from dataclasses import asdict, dataclass, field +from dataclasses import dataclass, field import logging import traceback from typing import List, Tuple @@ -6,6 +6,7 @@ from control import data from control.ev import ev as ev_module from control.chargepoint.charging_type import ChargingType +from dataclass_utils import asdict from dataclass_utils.factories import empty_list_factory from helpermodules.abstract_plans import AutolockPlan from helpermodules import timecheck diff --git a/packages/control/ev/charge_template.py b/packages/control/ev/charge_template.py index 6c53430dce..15ccf6af15 100644 --- a/packages/control/ev/charge_template.py +++ b/packages/control/ev/charge_template.py @@ -1,4 +1,4 @@ -from dataclasses import asdict, dataclass, field +from dataclasses import dataclass, field import datetime import logging import traceback @@ -11,6 +11,7 @@ from control.chargepoint.control_parameter import ControlParameter from control.ev.ev_template import EvTemplate from control.text import BidiState +from dataclass_utils import asdict from dataclass_utils.factories import empty_list_factory from helpermodules.abstract_plans import Limit, limit_factory, ScheduledChargingPlan, TimeChargingPlan from helpermodules import timecheck diff --git a/packages/dataclass_utils/_dataclass_asdict.py b/packages/dataclass_utils/_dataclass_asdict.py index 5d43362065..af13a567ca 100644 --- a/packages/dataclass_utils/_dataclass_asdict.py +++ b/packages/dataclass_utils/_dataclass_asdict.py @@ -1,11 +1,15 @@ from enum import Enum import logging +from typing import Any, Dict, List, Union, cast log = logging.getLogger(__name__) -def asdict(value): +AsDictValue = Union[None, bool, int, float, str, List["AsDictValue"], Dict[Any, "AsDictValue"]] + + +def asdict(value: Any) -> AsDictValue: """Converts an object to a dict This function is a simple replacement for the `dataclasses.asdict` function introduced in Python 3.7. This function @@ -17,7 +21,9 @@ def asdict(value): if isinstance(value, Enum): return value.value if isinstance(value, (list, tuple)): - return [None if v is None else asdict(v) for v in value] + sequence = cast(List[Any], value) + return [None if item is None else asdict(item) for item in sequence] if not isinstance(value, dict): - value = vars(value) - return {key: None if value is None else asdict(value) for key, value in value.items()} + value = vars(cast(object, value)) + mapping = cast(Dict[Any, Any], value) + return {key: None if item is None else asdict(item) for key, item in mapping.items()} diff --git a/packages/helpermodules/changed_values_handler.py b/packages/helpermodules/changed_values_handler.py index e191f5851d..5692cf5762 100644 --- a/packages/helpermodules/changed_values_handler.py +++ b/packages/helpermodules/changed_values_handler.py @@ -6,7 +6,7 @@ from control import data from control.data import Data -from dataclass_utils._dataclass_asdict import asdict +from dataclass_utils import asdict from helpermodules.pub import Pub diff --git a/packages/helpermodules/changed_values_handler_test.py b/packages/helpermodules/changed_values_handler_test.py index 189912e814..48c13a4ba3 100644 --- a/packages/helpermodules/changed_values_handler_test.py +++ b/packages/helpermodules/changed_values_handler_test.py @@ -1,10 +1,11 @@ -from dataclasses import asdict, dataclass, field +from dataclasses import dataclass, field from enum import IntEnum from typing import Dict, List, Optional, Tuple from unittest.mock import Mock import pytest +from dataclass_utils import asdict from dataclass_utils.factories import currents_list_factory from helpermodules.changed_values_handler import ChangedValuesHandler diff --git a/packages/helpermodules/command.py b/packages/helpermodules/command.py index 2b429f135e..e14fd561cc 100644 --- a/packages/helpermodules/command.py +++ b/packages/helpermodules/command.py @@ -1,5 +1,4 @@ import copy -from dataclasses import asdict import importlib import json import logging @@ -13,12 +12,14 @@ from pathlib import Path import paho.mqtt.client as mqtt +from control import bat, bridge, counter, counter_all, pv from control.chargelog.process_chargelog import get_log_data from control.chargepoint import chargepoint from control.chargepoint.chargepoint_template import get_chargepoint_template_default - +from control.ev import ev from control.ev.charge_template import ChargeTemplate, get_new_charge_template from control.ev.ev_template import EvTemplateData +from dataclass_utils import asdict from helpermodules import pub from helpermodules.abstract_plans import AutolockPlan, ScheduledChargingPlan, TimeChargingPlan from helpermodules.utils.run_command import run_command @@ -39,12 +40,9 @@ from helpermodules.pub import Pub, pub_single from helpermodules.subdata import SubData from helpermodules.utils.topic_parser import decode_payload, get_index -from control import bat, bridge, counter, counter_all, pv -from control.ev import ev from modules.chargepoints.internal_openwb.chargepoint_module import ChargepointModule from modules.chargepoints.internal_openwb.config import InternalChargepointMode from modules.common.component_type import ComponentType, special_to_general_type_mapping, type_to_topic_mapping -import dataclass_utils from modules.common.configurable_vehicle import GeneralVehicleConfig @@ -176,7 +174,7 @@ def addDevice(self, connection_id: str, payload: dict) -> None: dev = importlib.import_module(f'.devices.{payload["data"]["vendor"]}' f'.{payload["data"]["type"]}.device', "modules") - device_default = dataclass_utils.asdict(dev.device_descriptor.configuration_factory()) + device_default = asdict(dev.device_descriptor.configuration_factory()) device_default["id"] = new_id Pub().pub(f'openWB/set/system/device/{new_id}/config', device_default) self.max_id_device = self.max_id_device + 1 @@ -204,7 +202,7 @@ def addIoAction(self, connection_id: str, payload: dict) -> None: dev = importlib.import_module(f".io_actions.{'.'.join(payload['data']['type'])}.api", "modules") descriptor = dev.device_descriptor.configuration_factory() - device_default = dataclass_utils.asdict(descriptor) + device_default = asdict(descriptor) device_default["id"] = new_id Pub().pub(f'openWB/set/io/action/{new_id}/config', device_default) self.max_id_io_action = new_id @@ -238,7 +236,7 @@ def addIoDevice(self, connection_id: str, payload: dict) -> None: new_id = self.max_id_io_device + 1 dev = importlib.import_module(".io_devices."+payload["data"]["type"]+".api", "modules") descriptor = dev.device_descriptor.configuration_factory() - device_default = dataclass_utils.asdict(descriptor) + device_default = asdict(descriptor) device_default["id"] = new_id Pub().pub(f'openWB/set/system/io/{new_id}/config', device_default) self.max_id_io_device = new_id @@ -279,7 +277,7 @@ def setup_added_chargepoint(): Pub().pub(f'openWB/chargepoint/{new_id}/config', chargepoint_config) {Pub().pub(f"openWB/chargepoint/{new_id}/get/"+k, v) for (k, v) in asdict(chargepoint.Get()).items()} charge_template = SubData.ev_charge_template_data[f"ct{SubData.ev_data['ev0'].data.charge_template}"] - charge_template = dataclass_utils.asdict(charge_template.data) + charge_template = asdict(charge_template.data) Pub().pub(f'openWB/chargepoint/{new_id}/set/charge_template', charge_template) Pub().pub(f'openWB/chargepoint/{new_id}/set/manual_lock', False) self.max_id_hierarchy = self.max_id_hierarchy + 1 @@ -300,7 +298,7 @@ def setup_added_chargepoint(): chargepoint_config = chargepoint.get_chargepoint_config_default() # chargepoint_default["id"] = new_id module = importlib.import_module(".chargepoints." + payload["data"]["type"] + ".chargepoint_module", "modules") - chargepoint_config.update(dataclass_utils.asdict( + chargepoint_config.update(asdict( module.chargepoint_descriptor.configuration_factory()).items()) check_num_msg = self._check_max_num_of_internal_chargepoints(chargepoint_config) if check_num_msg is not None: @@ -491,7 +489,7 @@ def removeAutolockPlan(self, connection_id: str, payload: dict) -> None: break Pub().pub( f'openWB/set/chargepoint/template/{payload["data"]["template"]}', - dataclass_utils.asdict(SubData.cp_template_data[f'cpt{payload["data"]["template"]}'].data)) + asdict(SubData.cp_template_data[f'cpt{payload["data"]["template"]}'].data)) pub_user_message( payload, connection_id, f'Plan für Sperren nach Uhrzeit mit ID \'{payload["data"]["plan"]}\' vom Profil ' @@ -563,11 +561,11 @@ def _pub_charge_template_to_source(self, payload: dict, charge_template: ChargeT if payload["data"]["changed_in_theme"]: Pub().pub( f'openWB/set/chargepoint/{payload["data"]["chargepoint"]}/set/charge_template', - dataclass_utils.asdict(charge_template.data)) + asdict(charge_template.data)) else: Pub().pub( f'openWB/set/vehicle/template/charge_template/{payload["data"]["template"]}', - dataclass_utils.asdict(charge_template.data)) + asdict(charge_template.data)) def addChargeTemplateSchedulePlan(self, connection_id: str, payload: dict) -> None: """ sendet das Topic, zu dem ein neuer Zielladen-Plan erstellt werden soll. @@ -667,7 +665,7 @@ def set_default(topic: str, defaults: dict): component = importlib.import_module(f'.devices.{payload["data"]["deviceVendor"]}' f'.{payload["data"]["deviceType"]}.{payload["data"]["type"]}', "modules") - component_default = dataclass_utils.asdict(component.component_descriptor.configuration_factory()) + component_default = asdict(component.component_descriptor.configuration_factory()) component_default["id"] = new_id general_type = special_to_general_type_mapping(payload["data"]["type"]) try: @@ -729,7 +727,7 @@ def addEvTemplate(self, connection_id: str, payload: dict) -> None: new_ev_template = asdict(SubData.ev_template_data[f"et{payload['data']['copy']}"].data).copy() new_ev_template["name"] = f'Kopie von {new_ev_template["name"]}' else: - new_ev_template = dataclass_utils.asdict(EvTemplateData()) + new_ev_template = asdict(EvTemplateData()) new_ev_template["name"] = f'{new_ev_template["name"]} {new_id}' new_ev_template["id"] = new_id self.max_id_ev_template = new_id @@ -763,8 +761,7 @@ def addVehicle(self, connection_id: str, payload: dict) -> None: for default in vehicle_default: Pub().pub(f"openWB/set/vehicle/{new_id}/{default}", vehicle_default[default]) Pub().pub(f"openWB/set/vehicle/{new_id}/soc_module/config", {"type": None, "configuration": {}}) - Pub().pub(f"openWB/set/vehicle/{new_id}/soc_module/general_config", - dataclass_utils.asdict(GeneralVehicleConfig())) + Pub().pub(f"openWB/set/vehicle/{new_id}/soc_module/general_config", asdict(GeneralVehicleConfig())) self.max_id_vehicle = self.max_id_vehicle + 1 Pub().pub("openWB/set/command/max_id/vehicle", self.max_id_vehicle) # Default-Mäßig werden die Profile 0 zugewiesen, wenn diese noch nicht existieren -> anlegen diff --git a/packages/helpermodules/setdata.py b/packages/helpermodules/setdata.py index 7c03072896..09fe9a97a6 100644 --- a/packages/helpermodules/setdata.py +++ b/packages/helpermodules/setdata.py @@ -2,7 +2,6 @@ """ import copy -import dataclasses from pathlib import Path from threading import Event from typing import List, Optional, Tuple @@ -170,7 +169,7 @@ def _validate_value(self, msg: mqtt.MQTTMessage, data_type, ranges=[], collectio elif re.search("^openWB/set/chargepoint/[1-9][0-9]*/config.*$", msg.topic) is not None: event = self.event_cp_config if "cp"+str(index) in subdata.SubData.cp_data: - template = dataclasses.asdict( + template = dataclass_utils.asdict( subdata.SubData.cp_data["cp"+str(index)].chargepoint.data.config) else: template = {} diff --git a/packages/helpermodules/update_config.py b/packages/helpermodules/update_config.py index c35ba3b974..a8f1fb7d7e 100644 --- a/packages/helpermodules/update_config.py +++ b/packages/helpermodules/update_config.py @@ -1,6 +1,5 @@ from concurrent.futures import ProcessPoolExecutor import copy -from dataclasses import asdict import datetime import glob import importlib @@ -13,7 +12,6 @@ from paho.mqtt.client import Client as MqttClient, MQTTMessage from control.limiting_value import LoadmanagementLimit -import dataclass_utils from control.chargepoint.chargepoint_template import get_chargepoint_template_default from helpermodules import timecheck @@ -43,6 +41,7 @@ from control.ev.ev_template import EvTemplateData from control.general import Prices, PvCharging from control.optional_data import OcppConfig +from dataclass_utils import asdict from modules.common.abstract_vehicle import GeneralVehicleConfig from modules.common.component_type import ComponentType from modules.devices.sungrow.sungrow_sh.version import Version @@ -594,7 +593,7 @@ class UpdateConfig: ("openWB/vehicle/0/info", {"manufacturer": None, "model": None}), ("openWB/vehicle/0/charge_template", ev.Ev(0).charge_template.data.id), ("openWB/vehicle/0/soc_module/config", NO_MODULE), - ("openWB/vehicle/0/soc_module/general_config", dataclass_utils.asdict(GeneralVehicleConfig())), + ("openWB/vehicle/0/soc_module/general_config", asdict(GeneralVehicleConfig())), ("openWB/vehicle/0/ev_template", ev.Ev(0).ev_template.data.id), ("openWB/vehicle/0/tag_id", ev.Ev(0).data.tag_id), ("openWB/vehicle/0/get/soc", ev.Ev(0).data.get.soc), @@ -632,7 +631,7 @@ class UpdateConfig: ("openWB/general/prices/pv", Prices().pv), ("openWB/general/range_unit", "km"), ("openWB/general/temporary_charge_templates_active", False), - ("openWB/general/web_theme", dataclass_utils.asdict(KoalaWebTheme())), + ("openWB/general/web_theme", asdict(KoalaWebTheme())), ("openWB/graph/config/duration", 120), ("openWB/internal_chargepoint/0/data/parent_cp", None), ("openWB/internal_chargepoint/1/data/parent_cp", None), @@ -645,10 +644,10 @@ class UpdateConfig: ("openWB/optional/int_display/pin_code", "0000"), ("openWB/optional/int_display/standby", 60), ("openWB/optional/int_display/rotation", 0), - ("openWB/optional/int_display/theme", dataclass_utils.asdict(CardsDisplayTheme())), + ("openWB/optional/int_display/theme", asdict(CardsDisplayTheme())), ("openWB/optional/int_display/only_local_charge_points", False), ("openWB/optional/monitoring/config", NO_MODULE), - ("openWB/optional/ocpp/config", dataclass_utils.asdict(OcppConfig())), + ("openWB/optional/ocpp/config", asdict(OcppConfig())), ("openWB/optional/rfid/active", False), ("openWB/pv/config/configured", False), ("openWB/system/backup_password", None), @@ -1095,8 +1094,7 @@ def upgrade(topic: str, payload) -> Optional[dict]: if re.search("openWB/vehicle/[0-9]+/soc_module/config", topic) is not None: payload = decode_payload(payload) index = get_index(topic) - return {f"openWB/set/vehicle/{index}/soc_module/interval_config": - dataclass_utils.asdict(GeneralVehicleConfig())} + return {f"openWB/set/vehicle/{index}/soc_module/interval_config": asdict(GeneralVehicleConfig())} self._loop_all_received_topics(upgrade) self._append_datastore_version(12) @@ -2179,15 +2177,15 @@ def upgrade(topic: str, payload) -> Optional[dict]: {"value": 0, "input_matrix": {"RSE1": True, "RSE2": False}}, {"value": 0, "input_matrix": {"RSE1": True, "RSE2": True}}] - return {'openWB/system/io/0/config': dataclass_utils.asdict(io_device), - 'openWB/io/action/0/config': dataclass_utils.asdict(action)} + return {'openWB/system/io/0/config': asdict(io_device), + 'openWB/io/action/0/config': asdict(action)} self._loop_all_received_topics(upgrade) self._append_datastore_version(75) def upgrade_datastore_76(self) -> None: def upgrade(topic: str, payload) -> Optional[dict]: if re.search("openWB/chargepoint/[0-9]+/control_parameter/limit", topic) is not None: - return {topic: dataclass_utils.asdict(LoadmanagementLimit(None, None))} + return {topic: asdict(LoadmanagementLimit(None, None))} self._loop_all_received_topics(upgrade) self._append_datastore_version(76) @@ -2299,7 +2297,7 @@ def get_new_phases_to_use(topic) -> int: payload = decode_payload(payload) charge_template = copy.deepcopy(payload) - charge_template["chargemode"]["eco_charging"] = dataclass_utils.asdict(EcoCharging()) + charge_template["chargemode"]["eco_charging"] = asdict(EcoCharging()) if payload["chargemode"]["selected"] == "standby": charge_template["chargemode"]["selected"] = "stop" if payload["et"]["active"] is True: @@ -2315,7 +2313,7 @@ def get_new_phases_to_use(topic) -> int: charge_template["chargemode"]["pv_charging"]["phases_to_use"] = get_new_phases_to_use( "openWB/general/chargemode_config/pv_charging/phases_to_use") charge_template["chargemode"]["pv_charging"]["phases_to_use_min_soc"] = min(max_phases_ev, 3) - charge_template["chargemode"]["pv_charging"]["limit"] = dataclass_utils.asdict(Limit()) + charge_template["chargemode"]["pv_charging"]["limit"] = asdict(Limit()) if payload["chargemode"]["pv_charging"]["max_soc"] == 101: charge_template["chargemode"]["pv_charging"]["limit"]["selected"] = "none" else: diff --git a/packages/modules/backup_clouds/nextcloud/backup_cloud.py b/packages/modules/backup_clouds/nextcloud/backup_cloud.py index 9a8af0693b..f42be2a166 100644 --- a/packages/modules/backup_clouds/nextcloud/backup_cloud.py +++ b/packages/modules/backup_clouds/nextcloud/backup_cloud.py @@ -3,7 +3,7 @@ import re from typing import Callable, List, Tuple -from dataclass_utils._dataclass_asdict import asdict +from dataclass_utils import asdict from helpermodules.pub import Pub from modules.backup_clouds.nextcloud.config import NextcloudBackupCloud, NextcloudBackupCloudConfiguration from modules.common import req diff --git a/packages/modules/devices/avm/avm/device.py b/packages/modules/devices/avm/avm/device.py index db1fb7c51a..4c370a9ac0 100644 --- a/packages/modules/devices/avm/avm/device.py +++ b/packages/modules/devices/avm/avm/device.py @@ -5,7 +5,7 @@ from typing import Iterable import xml.etree.ElementTree as ET -from dataclass_utils._dataclass_asdict import asdict +from dataclass_utils import asdict from helpermodules.pub import Pub from modules.common import req from modules.common.abstract_device import DeviceDescriptor diff --git a/packages/modules/io_devices/eebus/api.py b/packages/modules/io_devices/eebus/api.py index 12b8a68b8d..f212d6bc30 100644 --- a/packages/modules/io_devices/eebus/api.py +++ b/packages/modules/io_devices/eebus/api.py @@ -7,7 +7,7 @@ from typing import Optional from control import data -from dataclass_utils._dataclass_asdict import asdict +from dataclass_utils import asdict from helpermodules import timecheck from helpermodules.broker import BrokerClient from helpermodules.pub import Pub