fix(car): stop the car charge start time walking forward (#269)

apps/predbat/gateway.py

@@ -16,7 +16,7 @@
 import time
 import uuid
 import traceback
-from utils import calc_percent_limit, export_mode_of, export_target_of, export_power_of
+from utils import calc_percent_limit, export_mode_of, export_target_of, export_power_of, parse_car_plan_windows, in_car_plan_window
 from const import EXPORT_MODE_FREEZE, EXPORT_MODE_IDLE, MINUTE_WATT
 import pytz as _pytz
 
@@ -505,41 +505,17 @@ def _refresh_ev_windows(self):
         with datetime strings in ``"%m-%d %H:%M:%S"`` format (produced by output.py).
 
         Datetimes are localized using ``self.local_tz`` so comparisons respect the component
-        timezone and DST transitions.  Year boundaries are handled: if a parsed start is more
-        than 23 hours in the past (i.e. the plan was built on Dec 31 and contains Jan 1
-        windows), the year is bumped forward.  Similarly, if end falls before start after
-        localization the end year is incremented to handle windows that straddle midnight
-        on New Year's Eve.
+        timezone and DST transitions.  The plan carries no year; ``utils.parse_car_plan_windows()``
+        (shared with the other charger components) rebuilds it around now, so a window that straddles
+        New Year is anchored correctly whichever side of midnight it is read (#269).
         """
         planned = self.get_state_wrapper(f"binary_sensor.{self.prefix}_car_charging_slot", attribute="planned") or []
         now = datetime.datetime.now(self.local_tz)
-        current_year = now.year
-        windows = []
-        for w in planned:
-            try:
-                start_naive = datetime.datetime.strptime(w["start"], "%m-%d %H:%M:%S").replace(year=current_year)
-                end_naive = datetime.datetime.strptime(w["end"], "%m-%d %H:%M:%S").replace(year=current_year)
-                start_dt = self.local_tz.localize(start_naive)
-                end_dt = self.local_tz.localize(end_naive)
-                # If start is far in the past the plan crossed a year boundary (Dec 31 → Jan 1)
-                if start_dt < now - datetime.timedelta(hours=23):
-                    start_dt = start_dt.replace(year=start_dt.year + 1)
-                    end_dt = end_dt.replace(year=end_dt.year + 1)
-                elif end_dt < start_dt:
-                    # end crossed into the new year but start did not (e.g. 23:30 → 00:30)
-                    end_dt = end_dt.replace(year=end_dt.year + 1)
-                windows.append((start_dt, end_dt))
-            except (KeyError, ValueError):
-                continue
-        self._ev_windows = windows
+        self._ev_windows = parse_car_plan_windows(planned, now, self.local_tz)
 
     def _should_ev_charge_now(self):
         """Return True if the current local time falls inside any planned charge window."""
-        now = datetime.datetime.now(self.local_tz)
-        for start_dt, end_dt in self._ev_windows:
-            if start_dt <= now < end_dt:
-                return True
-        return False
+        return in_car_plan_window(self._ev_windows, datetime.datetime.now(self.local_tz))
 
     async def _apply_ev_charging_state(self):
         """Start or stop EVC charging when the window state transitions.

apps/predbat/ohme.py

@@ -28,6 +28,7 @@
 from const import TIME_FORMAT_HA, DISPATCH_SOURCE_CHARGER_SCHEDULE
 from component_base import ComponentBase
 from predbat_metrics import record_api_call
+from utils import parse_car_plan_windows, in_car_plan_window
 
 GOOGLE_API_KEY = "AIzaSyC8ZeZngm33tpOXLpbXeKfwtyZ1WrkbdBY"  # cspell:disable-line
 VERSION = "1.5.1"
@@ -95,9 +96,6 @@
     "octopus_charge_limit": {"entity_id": "number.predbat_ohme_target_percent", "domain": "number", "access": "rw"},
 }
 
-# Format Predbat writes its planned car charging windows in - see PredBat.time_abs_str()
-PLAN_TIME_FORMAT = "%m-%d %H:%M:%S"
-
 # Longest gap between power readings we will still integrate over. The charge session is polled
 # every 120 seconds, so a longer gap means Predbat stalled or was restarted and we have no evidence
 # of what the charger did meanwhile - under-counting is safe, inventing energy is not
@@ -391,30 +389,16 @@ def refresh_car_windows(self):
         if planned is None:
             return False
 
-        now = self.now_utc_exact
-        windows = []
-        for window in planned:
-            try:
-                start = self.local_tz.localize(datetime.datetime.strptime(window["start"], PLAN_TIME_FORMAT).replace(year=now.year))
-                end = self.local_tz.localize(datetime.datetime.strptime(window["end"], PLAN_TIME_FORMAT).replace(year=now.year))
-            except (KeyError, TypeError, ValueError):
-                continue
-            # The plan carries no year, so rebuild it around now for windows that cross New Year
-            if start < now - timedelta(hours=23):
-                start = start.replace(year=start.year + 1)
-                end = end.replace(year=end.year + 1)
-            elif end < start:
-                end = end.replace(year=end.year + 1)
-            windows.append((start, end))
-        self.control_windows = windows
+        # The plan carries no year - parse_car_plan_windows() (shared with myenergi and GivEnergy
+        # EVC) rebuilds it around now, whichever side of New Year it is read (#269)
+        self.control_windows = parse_car_plan_windows(planned, self.now_utc_exact, self.local_tz)
         return True
 
     def should_charge_now(self):
         """
         Is now inside one of Predbat's planned charging windows.
         """
-        now = self.now_utc_exact
-        return any(start <= now < end for start, end in self.control_windows)
+        return in_car_plan_window(self.control_windows, self.now_utc_exact)
 
     def control_drifted(self, should_charge):
         """

apps/predbat/output.py

@@ -163,10 +163,16 @@ def publish_car_plan(self):
                 else:
                     slot = False
 
+                # Show when the window really began, not the planner's clamped start, so the
+                # displayed start time stops walking forward once charging is underway (#269).
+                # Octopus slots carry no start_orig and show their own start - which the Octopus
+                # component trims to now once a dispatch is underway, so those still move.
+                window_start = window.get("start_orig", window["start"])
+
                 time_format_time = "%H:%M:%S"
-                car_startt = self.midnight_utc + timedelta(minutes=window["start"])
+                car_startt = self.midnight_utc + timedelta(minutes=window_start)
                 car_start_time_str = car_startt.strftime(time_format_time)
-                minutes_to = max(window["start"] - self.minutes_now, 0)
+                minutes_to = max(window_start - self.minutes_now, 0)
                 self.dashboard_item(
                     self.prefix + ".car_charging_start" + postfix,
                     state=car_start_time_str,
@@ -184,7 +190,7 @@ def publish_car_plan(self):
                 total_kwh = 0
                 total_cost = 0
                 for window in self.car_charging_slots[car_n]:
-                    start = self.time_abs_str(window["start"])
+                    start = self.time_abs_str(window.get("start_orig", window["start"]))
                     end = self.time_abs_str(window["end"])
                     kwh = dp2(window["kwh"])
                     average = dp2(window["average"])

apps/predbat/plan.py

@@ -5978,6 +5978,11 @@ def plan_car_charging(self, car_n, low_rates):
                 new_slot = {}
                 new_slot["start"] = start
                 new_slot["end"] = end
+                # start is clamped to minutes_now above so the kWh maths only counts time the car
+                # can still charge for, but that makes an in-progress window's start walk forward
+                # by plan_interval_minutes every cycle. Keep the window's real start for display
+                # and for anything driving a charger off the published plan (issue #269).
+                new_slot["start_orig"] = window["start"]
                 new_slot["kwh"] = dp3(kwh)
                 new_slot["average"] = window["average"]
                 new_slot["cost"] = dp2(new_slot["average"] * kwh)

apps/predbat/tests/test_car_charging_smart.py

@@ -7,6 +7,7 @@
 # pylint: disable=consider-using-f-string
 # pylint: disable=line-too-long
 # pylint: disable=attribute-defined-outside-init
+import copy
 from tests.test_infra import reset_rates2, reset_inverter
 from config import CONFIG_ITEMS
 
@@ -157,4 +158,108 @@ def run_car_charging_smart_tests(my_predbat):
     failed |= run_car_charging_slot_integer_test("smart_int2_issue3911", my_predbat, battery_size=11.9, soc=0.0, limit=11.9, rate=7.4, loss=1.0, plan_time="07:00:00")
     failed |= run_car_charging_slot_integer_test("smart_int3_issue3911", my_predbat, battery_size=77.0, soc=74.3, limit=77.0, rate=11.0, loss=0.9, plan_time="07:00:00")
 
+    failed |= run_car_charging_start_stable_test(my_predbat)
+
+    return failed
+
+
+def run_car_charging_start_stable_test(my_predbat):
+    """
+    The published car charge start time must not walk forward once a slot is underway.
+
+    Regression test for issue #269: plan_car_charging() clamps an in-progress window's start
+    to minutes_now so the kWh maths only counts time the car can still charge for. That
+    clamped value was also published as predbat.car_charging_start, so the displayed start
+    time advanced by plan_interval_minutes on every cycle while the car was charging.
+    """
+    failed = False
+    print("**** Running Test: car_charging_start_stable_issue269 ****")
+
+    saved_battery_size = my_predbat.car_charging_battery_size
+    saved_limit = my_predbat.car_charging_limit
+    saved_soc = my_predbat.car_charging_soc
+    saved_soc_next = my_predbat.car_charging_soc_next
+    saved_rate = my_predbat.car_charging_rate
+    saved_loss = my_predbat.car_charging_loss
+    saved_max_price = my_predbat.car_charging_plan_max_price
+    saved_smart = my_predbat.car_charging_plan_smart
+    saved_plan_time = my_predbat.car_charging_plan_time
+    saved_num_cars = my_predbat.num_cars
+    saved_minutes_now = my_predbat.minutes_now
+    saved_slots = my_predbat.car_charging_slots
+
+    # publish_car_plan() writes the car sensors into the shared dashboard state
+    saved_dashboard = copy.deepcopy(my_predbat.dashboard_values)
+
+    try:
+        my_predbat.car_charging_battery_size = [100.0]
+        my_predbat.car_charging_limit = [100.0]
+        my_predbat.car_charging_soc = [0]
+        my_predbat.car_charging_soc_next = [None]
+        my_predbat.car_charging_rate = [1.0]
+        my_predbat.car_charging_loss = 1.0
+        my_predbat.car_charging_plan_max_price = [99]
+        my_predbat.car_charging_plan_smart = [False]
+        my_predbat.car_charging_plan_time = ["23:00:00"]
+        my_predbat.num_cars = 1
+        my_predbat.car_charging_slots = [[]]
+
+        # Plan once, then replan from part way through the first window as the real 5 minute
+        # cycle would, and check the published start time did not move.
+        start_times = []
+        plan_starts = []
+        first_window_start = None
+        for step in range(3):
+            if step == 0:
+                my_predbat.minutes_now = saved_minutes_now
+            else:
+                # Step forward a cycle at a time while staying inside the first planned window,
+                # which is what a car charging through the window actually sees
+                my_predbat.minutes_now = first_window_start + step * 5
+
+            my_predbat.car_charging_slots[0] = my_predbat.plan_car_charging(0, my_predbat.low_rates)
+            if not my_predbat.car_charging_slots[0]:
+                print("ERROR: no car charging slots planned at step {}".format(step))
+                failed = True
+                break
+            if first_window_start is None:
+                first_window_start = my_predbat.car_charging_slots[0][0]["start"]
+                first_window_end = my_predbat.car_charging_slots[0][0]["end"]
+                # The drift only shows while one window stays current, so fail loudly rather than
+                # silently passing if the fixture's windows ever get shorter than the steps below
+                if first_window_end - first_window_start < 3 * 5:
+                    print("ERROR: test needs a first window at least 15 minutes long, got {}-{}".format(first_window_start, first_window_end))
+                    failed = True
+                    break
+
+            my_predbat.publish_car_plan()
+            start_times.append(my_predbat.dashboard_values["predbat.car_charging_start"]["state"])
+            plan_starts.append(my_predbat.dashboard_values["binary_sensor.predbat_car_charging_slot"]["attributes"]["planned"][0]["start"])
+
+        if not failed:
+            if len(set(start_times)) != 1:
+                print("ERROR: car_charging_start moved while the slot was underway: {} (issue #269)".format(start_times))
+                failed = True
+            else:
+                print("OK: car_charging_start held at {} across the slot (issue #269)".format(start_times[0]))
+            if len(set(plan_starts)) != 1:
+                print("ERROR: planned window start moved while the slot was underway: {} (issue #269)".format(plan_starts))
+                failed = True
+    finally:
+        my_predbat.car_charging_battery_size = saved_battery_size
+        my_predbat.car_charging_limit = saved_limit
+        my_predbat.car_charging_soc = saved_soc
+        my_predbat.car_charging_soc_next = saved_soc_next
+        my_predbat.car_charging_rate = saved_rate
+        my_predbat.car_charging_loss = saved_loss
+        my_predbat.car_charging_plan_max_price = saved_max_price
+        my_predbat.car_charging_plan_smart = saved_smart
+        my_predbat.car_charging_plan_time = saved_plan_time
+        my_predbat.num_cars = saved_num_cars
+        my_predbat.minutes_now = saved_minutes_now
+        my_predbat.car_charging_slots = saved_slots
+
+        my_predbat.dashboard_values.clear()
+        my_predbat.dashboard_values.update(saved_dashboard)
+
     return failed

apps/predbat/tests/test_car_plan_windows.py

@@ -0,0 +1,72 @@
+# -----------------------------------------------------------------------------
+# Predbat Home Battery System
+# Copyright Trefor Southwell 2026 - All Rights Reserved
+# This application maybe used for personal use only and not for commercial use
+# -----------------------------------------------------------------------------
+# fmt off
+# pylint: disable=consider-using-f-string
+# pylint: disable=line-too-long
+# pylint: disable=attribute-defined-outside-init
+
+from datetime import datetime
+import pytz
+from utils import parse_car_plan_windows, in_car_plan_window
+
+LONDON = pytz.timezone("Europe/London")
+
+
+def _window(start, end):
+    """A planned-attribute entry as output.py publishes it."""
+    return {"start": start, "end": end, "kwh": 5.0, "average": 20.0, "cost": 1.0}
+
+
+def test_car_plan_windows(my_predbat=None):
+    """
+    utils.parse_car_plan_windows() rebuilds the year on the car charging plan (published as
+    "%m-%d %H:%M:%S") around now. Shared by every charger that follows the plan - myenergi,
+    GivEnergy EVC, Ohme and the gateway EVC (#269, #5120 review).
+    """
+    failed = False
+    print("**** Testing parse_car_plan_windows ****")
+
+    cases = [
+        # (name, now, start, end, expected start year, expected end year, now inside)
+        ("ordinary window", datetime(2026, 6, 15, 14, 0), "06-15 13:00:00", "06-15 15:00:00", 2026, 2026, True),
+        ("New Year window read before midnight", datetime(2026, 12, 31, 23, 45), "12-31 23:30:00", "01-01 01:30:00", 2026, 2027, True),
+        # Once now itself is in January the Dec 31 start belongs to the previous year, or the window
+        # lands a year in the future and a charging car is stopped mid-window
+        ("New Year window read after midnight", datetime(2027, 1, 1, 0, 15), "12-31 23:30:00", "01-01 01:30:00", 2026, 2027, True),
+        # A window wholly in January, read on 31 December, parses into January of the year ending and
+        # has to move forward a year
+        ("January window read on 31 December", datetime(2026, 12, 31, 23, 30), "01-01 00:30:00", "01-01 02:00:00", 2027, 2027, False),
+        # A long window whose start is days old is still the current one, not a year rollover (#269)
+        ("long active window", datetime(2026, 6, 16, 23, 5), "06-14 00:10:00", "06-17 02:00:00", 2026, 2026, True),
+        ("midnight-crossing window", datetime(2026, 6, 15, 23, 30), "06-15 23:00:00", "06-16 00:30:00", 2026, 2026, True),
+        # strptime with no year assumes 1900, which has no 29 February - the window used to be dropped
+        ("leap day window", datetime(2028, 2, 29, 1, 0), "02-29 00:30:00", "02-29 05:00:00", 2028, 2028, True),
+        ("window starting on leap day, read on 1 March", datetime(2028, 3, 1, 0, 30), "02-29 23:30:00", "03-01 02:00:00", 2028, 2028, True),
+    ]
+    for name, now_naive, start, end, start_year, end_year, inside in cases:
+        now = LONDON.localize(now_naive)
+        windows = parse_car_plan_windows([_window(start, end)], now, LONDON)
+        if len(windows) != 1:
+            print("ERROR: {}: expected one window, got {}".format(name, windows))
+            failed = True
+            continue
+        start_dt, end_dt = windows[0]
+        if (start_dt.year, end_dt.year) != (start_year, end_year) or in_car_plan_window(windows, now) != inside:
+            print("ERROR: {}: got start={} end={} (now {}, inside {})".format(name, start_dt, end_dt, now, in_car_plan_window(windows, now)))
+            failed = True
+
+    # A malformed entry, or a stale 29 February one that has no counterpart in the neighbouring year,
+    # is skipped without costing the rest of the plan
+    now = LONDON.localize(datetime(2028, 9, 1, 12, 0))
+    planned = [_window("not-a-date", "09-01 13:00:00"), {"start": None, "end": "09-01 13:00:00"}, {"end": "09-01 13:00:00"}, _window("02-29 01:00:00", "02-29 02:00:00"), _window("09-01 11:00:00", "09-01 13:00:00")]
+    windows = parse_car_plan_windows(planned, now, LONDON)
+    if len(windows) != 1 or not in_car_plan_window(windows, now):
+        print("ERROR: malformed entries should be skipped and the valid window kept, got {}".format(windows))
+        failed = True
+
+    if not failed:
+        print("**** All parse_car_plan_windows tests PASSED ****")
+    return failed