From 4e1362f3efffea7e2d5adf8bac773557e2a8bf68 Mon Sep 17 00:00:00 2001 From: LKuemmel Date: Mon, 14 Sep 2026 09:49:15 +0200 Subject: [PATCH 1/3] =?UTF-8?q?Text=20f=C3=BCr=20Start=20des=20n=C3=A4chst?= =?UTF-8?q?en=20Zeitladen-Plans=20in=20Ladeprofil=20und=20Verbraucher?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- packages/control/consumer/consumer.py | 7 +- packages/control/consumer/consumer_test.py | 36 +++++- packages/control/ev/charge_template.py | 8 +- packages/control/ev/charge_template_test.py | 37 +++++- packages/control/text.py | 10 ++ packages/helpermodules/timecheck.py | 125 +++++++++++++++++--- packages/helpermodules/timecheck_test.py | 66 +++++++++++ 7 files changed, 263 insertions(+), 26 deletions(-) diff --git a/packages/control/consumer/consumer.py b/packages/control/consumer/consumer.py index f0ceebf415..4820255337 100644 --- a/packages/control/consumer/consumer.py +++ b/packages/control/consumer/consumer.py @@ -9,6 +9,7 @@ from control.chargepoint.chargepoint_state import CHARGING_STATES, ChargepointState from control.consumer.consumer_data import ConsumerData, ConsumerUsage, ResetModes, WaitForStartStates from control.load_protocol import Load +from control.text import format_next_time_charging_start from helpermodules import timecheck from helpermodules.abstract_plans import ScheduledPlanConsumer from helpermodules.phase_handling import convert_single_evu_phase_to_cp_phase, voltages_mean @@ -350,7 +351,11 @@ def time_charging(self) -> Tuple[float, Optional[str], Chargemode]: self._convert_power_to_current(self.data.config.max_power)) submode = Chargemode.TIME_CHARGING else: - message = self.TIME_CHARGING_NO_PLAN_ACTIVE + next_start = timecheck.get_next_timeframe_plan_start(self.data.usage.time_charging.plans) + if next_start is not None: + message = format_next_time_charging_start(next_start) + else: + message = self.TIME_CHARGING_NO_PLAN_ACTIVE else: message = self.TIME_CHARGING_NO_PLAN_CONFIGURED return required_current, message, submode diff --git a/packages/control/consumer/consumer_test.py b/packages/control/consumer/consumer_test.py index 49f3a129cc..a7176aa45e 100644 --- a/packages/control/consumer/consumer_test.py +++ b/packages/control/consumer/consumer_test.py @@ -7,7 +7,7 @@ from helpermodules.abstract_plans import (ScheduledChargingPlan, ScheduledPlanConsumer, TimeChargingPlan, TimeChargingPlanConsumer) from helpermodules import timecheck -from control import optional +from control import optional, text from control import data from control.general import General from control.bat_all import BatAll @@ -149,7 +149,8 @@ def test_wait_for_start_handler( pytest.param({}, None, (0, Consumer.TIME_CHARGING_NO_PLAN_CONFIGURED, Chargemode.STOP), id="no plan defined"), pytest.param({"0": TimeChargingPlanConsumer(id=0)}, None, - (0, Consumer.TIME_CHARGING_NO_PLAN_ACTIVE, Chargemode.STOP), id="no plan active"), + (0, Consumer.TIME_CHARGING_NO_PLAN_ACTIVE, Chargemode.STOP), + id="no plan active without next start"), pytest.param({"0": TimeChargingPlanConsumer(id=0)}, TimeChargingPlanConsumer(id=0), (10, None, Chargemode.TIME_CHARGING), id="plan active"), pytest.param({"0": TimeChargingPlanConsumer(id=0)}, None, @@ -166,6 +167,7 @@ def test_time_charging(plans: Dict[int, TimeChargingPlanConsumer], consumer.data.usage.time_charging.plans = plans check_plans_timeframe_mock = Mock(return_value=plan_found) monkeypatch.setattr(timecheck, "check_plans_timeframe", check_plans_timeframe_mock) + monkeypatch.setattr(timecheck, "get_next_timeframe_plan_start", Mock(return_value=None)) # execution ret = consumer.time_charging() @@ -174,6 +176,36 @@ def test_time_charging(plans: Dict[int, TimeChargingPlanConsumer], assert ret == expected +def test_time_charging_no_plan_active_includes_next_start( + consumer: Consumer, + monkeypatch: pytest.MonkeyPatch): + # setup + consumer.data.usage.time_charging.plans = [TimeChargingPlanConsumer(id=0)] + next_start = datetime.datetime(2026, 1, 2, 14, 30) + + monkeypatch.setattr(timecheck, "check_plans_timeframe", Mock(return_value=None)) + monkeypatch.setattr( + timecheck, + "get_next_timeframe_plan_start", + Mock(return_value=next_start), + ) + monkeypatch.setattr( + text, + "format_next_time_charging_start", + Mock(return_value="Nächster Zeitladen-Plan startet am 02.01. um 14:30 Uhr."), + ) + + # execution + ret = consumer.time_charging() + + # evaluation + assert ret == ( + 0, + text.format_next_time_charging_start(next_start), + Chargemode.STOP, + ) + + @pytest.mark.parametrize( "min_bat_soc, charging_allowed_mock, soc, expected", [ diff --git a/packages/control/ev/charge_template.py b/packages/control/ev/charge_template.py index daa02f3c5c..d6a309064f 100644 --- a/packages/control/ev/charge_template.py +++ b/packages/control/ev/charge_template.py @@ -10,7 +10,7 @@ from control.chargepoint.charging_type import ChargingType from control.chargepoint.control_parameter import ControlParameter from control.ev.ev_template import EvTemplate -from control.text import BidiState +from control.text import BidiState, format_next_time_charging_start from dataclass_utils import asdict from dataclass_utils.factories import empty_list_factory from helpermodules.abstract_plans import Limit, limit_factory, ScheduledChargingPlan, TimeChargingPlan @@ -187,7 +187,11 @@ def time_charging(self, current = plan.current if charging_type == ChargingType.AC.value else plan.dc_current sub_mode = ChargemodeEnum.TIME_CHARGING else: - message = self.TIME_CHARGING_NO_PLAN_ACTIVE + next_start = timecheck.get_next_timeframe_plan_start(self.data.time_charging.plans) + if next_start is not None: + message = format_next_time_charging_start(next_start) + else: + message = self.TIME_CHARGING_NO_PLAN_ACTIVE else: message = self.TIME_CHARGING_NO_PLAN_CONFIGURED return current, sub_mode, message, id, phases diff --git a/packages/control/ev/charge_template_test.py b/packages/control/ev/charge_template_test.py index 652da3e6ea..9136ba4ac5 100644 --- a/packages/control/ev/charge_template_test.py +++ b/packages/control/ev/charge_template_test.py @@ -6,6 +6,7 @@ from control import data from control import optional +from control import text from control.bat_all import BatAll from control.chargemode import Chargemode from control.chargepoint.control_parameter import ControlParameter @@ -14,7 +15,7 @@ from control.ev.charge_template import ChargeTemplate from control.ev.ev_template import EvTemplate, EvTemplateData from control.general import General -from control.text import BidiState +from control.text import BidiState, format_next_time_charging_start from helpermodules import timecheck from helpermodules.abstract_plans import Limit, ScheduledChargingPlan, TimeChargingPlan, ScheduledLimit @@ -62,12 +63,13 @@ def data_module() -> None: def test_time_charging(plans: Dict[int, TimeChargingPlan], soc: float, used_amount_time_charging: float, plan_found: TimeChargingPlan, expected: Tuple[int, Chargemode, Optional[str], Optional[str], Optional[int]], - monkeypatch): + monkeypatch: pytest.MonkeyPatch): # setup ct = ChargeTemplate() ct.data.time_charging.plans = plans check_plans_timeframe_mock = Mock(return_value=plan_found) monkeypatch.setattr(timecheck, "check_plans_timeframe", check_plans_timeframe_mock) + monkeypatch.setattr(timecheck, "get_next_timeframe_plan_start", Mock(return_value=None)) # execution ret = ct.time_charging(soc, used_amount_time_charging, ChargingType.AC.value) @@ -76,6 +78,37 @@ def test_time_charging(plans: Dict[int, TimeChargingPlan], soc: float, used_amou assert ret == expected +def test_time_charging_no_plan_active_includes_next_start(monkeypatch: pytest.MonkeyPatch): + # setup + ct = ChargeTemplate() + ct.data.time_charging.plans = [TimeChargingPlan(id=0)] + next_start = datetime.datetime(2026, 1, 2, 14, 30) + + monkeypatch.setattr(timecheck, "check_plans_timeframe", Mock(return_value=None)) + monkeypatch.setattr( + timecheck, + "get_next_timeframe_plan_start", + Mock(return_value=next_start), + ) + monkeypatch.setattr( + text, + "format_next_time_charging_start", + Mock(return_value="Nächster Zeitladen-Plan startet am 02.01. um 14:30 Uhr."), + ) + + # execution + ret = ct.time_charging(0, 0, ChargingType.AC.value) + + # evaluation + assert ret == ( + 0, + Chargemode.STOP, + format_next_time_charging_start(next_start), + None, + None, + ) + + @pytest.mark.parametrize( "min_bat_soc, charging_allowed_mock, soc, expected", [ diff --git a/packages/control/text.py b/packages/control/text.py index 290893882f..840a2ee6d2 100644 --- a/packages/control/text.py +++ b/packages/control/text.py @@ -1,3 +1,4 @@ +import datetime from enum import Enum @@ -6,3 +7,12 @@ class BidiState(Enum): CP_NOT_BIDI_CAPABLE = "Bidirektionales Laden ist nur mit einer openWB Pro oder Pro+ möglich. " CP_WRONG_PROTOCOL = "Bitte in den Einstellungen der openWB Pro/Pro+ die Charging Version auf 'HLC' stellen. " EV_NOT_BIDI_CAPABLE = "Das Fahrzeug unterstützt kein bidirektionales Laden. " + + +def format_next_time_charging_start(next_start: datetime.datetime) -> str: + today = datetime.datetime.today().date() + if next_start.date() == today: + return f"Nächster Zeitladen-Plan startet heute um {next_start.strftime('%-H:%M')} Uhr." + if next_start.date() == today + datetime.timedelta(days=1): + return f"Nächster Zeitladen-Plan startet morgen um {next_start.strftime('%-H:%M')} Uhr." + return f"Nächster Zeitladen-Plan startet am {next_start.strftime('%d.%m.')} um {next_start.strftime('%-H:%M')} Uhr." diff --git a/packages/helpermodules/timecheck.py b/packages/helpermodules/timecheck.py index c0a86edea6..e4720c4d7f 100644 --- a/packages/helpermodules/timecheck.py +++ b/packages/helpermodules/timecheck.py @@ -48,6 +48,19 @@ def is_now_in_locking_time(now: datetime.datetime, T = TypeVar("T", AutolockPlan, TimeChargingPlan, TimeChargingPlanConsumer) +def _parse_plan_time(time_value: str) -> Tuple[int, int]: + parsed_time = datetime.datetime.strptime(time_value, "%H:%M") + return parsed_time.hour, parsed_time.minute + + +def _is_timeframe_valid(now: datetime.datetime, begin: datetime.datetime, end: datetime.datetime) -> bool: + return (not now < begin) and now < end + + +def _is_weekday_enabled(weekly: List[bool], weekday: int) -> bool: + return weekly[weekday % 7] + + def check_plans_timeframe(plans: List[T]) -> Optional[T]: """ gibt den ersten aktiven Plan zurück. None, falls kein Plan aktiv ist. """ @@ -68,52 +81,126 @@ def check_plans_timeframe(plans: List[T]) -> Optional[T]: def check_timeframe(plan: Union[AutolockPlan, TimeChargingPlan]) -> bool: """ Returns: True -> Zeitfenster gültig, False -> Zeitfenster nicht gültig """ - def is_timeframe_valid(now: datetime.datetime, begin: datetime.datetime, end: datetime.datetime) -> bool: - return True if (not now < begin) and now < end else False - state = False try: now = datetime.datetime.today() - begin = datetime.datetime.strptime(plan.time[0], '%H:%M') - end = datetime.datetime.strptime(plan.time[1], '%H:%M') + begin_hour, begin_minute = _parse_plan_time(plan.time[0]) + end_hour, end_minute = _parse_plan_time(plan.time[1]) if plan.frequency.selected == "once": - beginDate = datetime.datetime.strptime(plan.frequency.once[0], "%Y-%m-%d") - begin = begin.replace(beginDate.year, beginDate.month, beginDate.day) - endDate = datetime.datetime.strptime(plan.frequency.once[1], "%Y-%m-%d") - end = end.replace(endDate.year, endDate.month, endDate.day) - state = is_timeframe_valid(now, begin, end) + begin_date = datetime.datetime.strptime(plan.frequency.once[0], "%Y-%m-%d") + begin = now.replace( + year=begin_date.year, + month=begin_date.month, + day=begin_date.day, + hour=begin_hour, + minute=begin_minute, + second=0, + microsecond=0, + ) + end_date = datetime.datetime.strptime(plan.frequency.once[1], "%Y-%m-%d") + end = now.replace( + year=end_date.year, + month=end_date.month, + day=end_date.day, + hour=end_hour, + minute=end_minute, + second=0, + microsecond=0, + ) + state = _is_timeframe_valid(now, begin, end) else: - begin = begin.replace(now.year, now.month, now.day) - end = end.replace(now.year, now.month, now.day) + begin = now.replace(hour=begin_hour, minute=begin_minute, second=0, microsecond=0) + end = now.replace(hour=end_hour, minute=end_minute, second=0, microsecond=0) day_change = begin > end if day_change: # Endzeit ist am nächsten Tag, in Zeitabschnitt vor und nach Mitternacht einteilen next_day = now + datetime.timedelta(days=1) next_day_midnight = next_day.replace(hour=0, minute=0) - state_after_midnight = is_timeframe_valid(now, begin, next_day_midnight) - state_before_midnight = is_timeframe_valid(now, now.replace(hour=0, minute=0), end) + state_after_midnight = _is_timeframe_valid(now, begin, next_day_midnight) + state_before_midnight = _is_timeframe_valid(now, now.replace(hour=0, minute=0), end) if plan.frequency.selected == "daily": if day_change: state = state_before_midnight or state_after_midnight else: - state = is_timeframe_valid(now, begin, end) + state = _is_timeframe_valid(now, begin, end) elif plan.frequency.selected == "weekly": if day_change: - state = ((state_after_midnight and plan.frequency.weekly[now.weekday()]) or - (state_before_midnight and plan.frequency.weekly[now.weekday() - 1])) + state = ((state_after_midnight and _is_weekday_enabled(plan.frequency.weekly, now.weekday())) or + (state_before_midnight and _is_weekday_enabled(plan.frequency.weekly, now.weekday() - 1))) else: - if plan.frequency.weekly[now.weekday()]: - state = is_timeframe_valid(now, begin, end) + if _is_weekday_enabled(plan.frequency.weekly, now.weekday()): + state = _is_timeframe_valid(now, begin, end) except Exception: log.exception("Fehler im System-Modul") finally: return state +def get_next_timeframe_plan_start(plans: List[T], + now: Optional[datetime.datetime] = None) -> Optional[datetime.datetime]: + if now is None: + now = datetime.datetime.today() + + next_starts: List[datetime.datetime] = [] + for plan in plans: + if not getattr(plan, "active", False): + continue + try: + next_start = _get_next_start_for_timeframe_plan(plan, now) + if next_start is not None: + next_starts.append(next_start) + except Exception: + log.exception("Fehler im System-Modul") + + if not next_starts: + return None + return min(next_starts) + + +def _get_next_start_for_timeframe_plan(plan: T, now: datetime.datetime) -> Optional[datetime.datetime]: + start_hour, start_minute = _parse_plan_time(plan.time[0]) + + if plan.frequency.selected == "once": + begin_date = datetime.datetime.strptime(plan.frequency.once[0], "%Y-%m-%d") + next_start = now.replace( + year=begin_date.year, + month=begin_date.month, + day=begin_date.day, + hour=start_hour, + minute=start_minute, + second=0, + microsecond=0, + ) + return next_start if next_start > now else None + + if plan.frequency.selected == "daily": + next_start = now.replace(hour=start_hour, minute=start_minute, second=0, microsecond=0) + if next_start <= now: + next_start += datetime.timedelta(days=1) + return next_start + + if plan.frequency.selected == "weekly": + if not any(plan.frequency.weekly): + return None + for day_offset in range(0, 8): + weekday = (now.weekday() + day_offset) % 7 + if not plan.frequency.weekly[weekday]: + continue + candidate = (now + datetime.timedelta(days=day_offset)).replace( + hour=start_hour, + minute=start_minute, + second=0, + microsecond=0, + ) + if candidate > now: + return candidate + return None + + def check_end_time(plan: Union[ScheduledPlanConsumer, ScheduledChargingPlan, TimeChargingPlanConsumer], buffer: Optional[float]) -> float: """ gibt die verbleibende Zeit in Sekunden zurück. diff --git a/packages/helpermodules/timecheck_test.py b/packages/helpermodules/timecheck_test.py index 38cb6d605e..143d010aac 100644 --- a/packages/helpermodules/timecheck_test.py +++ b/packages/helpermodules/timecheck_test.py @@ -118,6 +118,72 @@ def test_check_timeframe(plan: Union[AutolockPlan, TimeChargingPlan], now: str, assert state == expected_state +@pytest.mark.parametrize( + "plans, now, expected", + [ + pytest.param( + [TimeChargingPlan(active=True, time=["10:00", "12:00"], frequency=FrequencyPeriod(selected="daily"))], + datetime.datetime(2026, 1, 2, 9, 0), + datetime.datetime(2026, 1, 2, 10, 0), + id="daily-start-today", + ), + pytest.param( + [TimeChargingPlan(active=True, time=["10:00", "12:00"], frequency=FrequencyPeriod(selected="daily"))], + datetime.datetime(2026, 1, 2, 10, 30), + datetime.datetime(2026, 1, 3, 10, 0), + id="daily-start-next-day", + ), + pytest.param( + [ + TimeChargingPlan( + active=True, + time=["10:00", "12:00"], + frequency=FrequencyPeriod(selected="weekly", weekly=[ + False, False, False, False, True, False, False]), + ) + ], + datetime.datetime(2026, 1, 2, 9, 0), + datetime.datetime(2026, 1, 2, 10, 0), + id="weekly-start-same-day", + ), + pytest.param( + [ + TimeChargingPlan( + active=True, + time=["10:00", "12:00"], + frequency=FrequencyPeriod(selected="once", once=["2026-01-05", "2026-01-05"]), + ) + ], + datetime.datetime(2026, 1, 2, 9, 0), + datetime.datetime(2026, 1, 5, 10, 0), + id="once-start-future", + ), + pytest.param( + [ + TimeChargingPlan( + active=False, + time=["10:00", "12:00"], + frequency=FrequencyPeriod(selected="daily"), + ) + ], + datetime.datetime(2026, 1, 2, 9, 0), + None, + id="inactive-plan-ignored", + ), + ], +) +def test_get_next_timeframe_plan_start( + plans: List[TimeChargingPlan], + now: datetime.datetime, + expected: Optional[datetime.datetime], +): + # execution + result = timecheck.get_next_timeframe_plan_start(plans, now) + + # evaluation + assert result == expected + + @pytest.mark.parametrize("timestamp, expected", [ pytest.param(1652683202, "40 Sek."), From bb819f8328512ad038c7b0f014d504938ff1489b Mon Sep 17 00:00:00 2001 From: LKuemmel <76958050+LKuemmel@users.noreply.github.com> Date: Mon, 14 Sep 2026 10:00:40 +0200 Subject: [PATCH 2/3] Potential fix for pull request finding Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com> --- packages/control/consumer/consumer_test.py | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/packages/control/consumer/consumer_test.py b/packages/control/consumer/consumer_test.py index a7176aa45e..8d656dab88 100644 --- a/packages/control/consumer/consumer_test.py +++ b/packages/control/consumer/consumer_test.py @@ -190,8 +190,7 @@ def test_time_charging_no_plan_active_includes_next_start( Mock(return_value=next_start), ) monkeypatch.setattr( - text, - "format_next_time_charging_start", + "control.consumer.consumer.format_next_time_charging_start", Mock(return_value="Nächster Zeitladen-Plan startet am 02.01. um 14:30 Uhr."), ) @@ -201,7 +200,7 @@ def test_time_charging_no_plan_active_includes_next_start( # evaluation assert ret == ( 0, - text.format_next_time_charging_start(next_start), + "Nächster Zeitladen-Plan startet am 02.01. um 14:30 Uhr.", Chargemode.STOP, ) From 8f05f3b597d0ce2df48d415d84c36e7a816b496a Mon Sep 17 00:00:00 2001 From: LKuemmel Date: Mon, 14 Sep 2026 10:10:18 +0200 Subject: [PATCH 3/3] flake8 --- packages/control/consumer/consumer_test.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/packages/control/consumer/consumer_test.py b/packages/control/consumer/consumer_test.py index 8d656dab88..a54e62344a 100644 --- a/packages/control/consumer/consumer_test.py +++ b/packages/control/consumer/consumer_test.py @@ -7,7 +7,7 @@ from helpermodules.abstract_plans import (ScheduledChargingPlan, ScheduledPlanConsumer, TimeChargingPlan, TimeChargingPlanConsumer) from helpermodules import timecheck -from control import optional, text +from control import optional from control import data from control.general import General from control.bat_all import BatAll