glenn schrooyenandClaude Opus 5.5
0b6b34f563
test_maint10.py part 10: at two inverters both packs get the site zone, the real Controller is identical for 11 UTC and 13 Brussels, and two due packs across 24/25 Oct 2026 and 27/28 Mar 2027 start at the local hour on successive days; a pack freed in the repeated hour does not start there. Part 3 adds 28 Mar 2027. DOCS: an existing install keeps its saved hour, now read as local time; an hour set to make up for UTC goes back to 19. DECISIONS: the SAFETY-83 gate is met; section on two inverters. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
529 lines
22 KiB
Python
529 lines
22 KiB
Python
"""MAINT-10: the maintenance start hour is LOCAL time. `python3 test_maint10.py`
|
|
|
|
Graded here: the slot on ordinary days and on both DST change days, the zone
|
|
source and its fallbacks, the 19:00 default, the startup wiring, and that the
|
|
controller's commands are unchanged once the start instant is the same.
|
|
See DECISIONS/MAINT-10-local-start-hour.md.
|
|
|
|
Mutating what this file grades? Read MUTATION.md first (stale bytecode).
|
|
|
|
⚠️ The CI image has no tzdata, so every DST check runs on tariff.py's encoded
|
|
EU rule and, where zoneinfo works, on Europe/Brussels from tzdata as well.
|
|
"""
|
|
|
|
import asyncio
|
|
import functools
|
|
import logging
|
|
import os
|
|
import re
|
|
import sys
|
|
from datetime import datetime, timedelta, timezone
|
|
from pathlib import Path
|
|
|
|
sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), "tools"))
|
|
|
|
import app.main as M # noqa: E402
|
|
from app import maintenance as MT # noqa: E402
|
|
from app.maintenance import ( # noqa: E402
|
|
DRAIN, IDLE, MAINT_OPTION_DEFAULTS, MaintConfig, Maintenance,
|
|
in_start_slot, next_slot_start, site_zone,
|
|
)
|
|
from app.tariff import _EUTimezone # noqa: E402
|
|
|
|
HERE = Path(__file__).resolve().parent
|
|
UTC = timezone.utc
|
|
fails = []
|
|
|
|
|
|
def check(name, cond):
|
|
print(f" {'ok ' if cond else 'FAIL'} {name}")
|
|
if not cond:
|
|
fails.append(name)
|
|
|
|
|
|
ZONES = [("encoded EU rule", _EUTimezone())]
|
|
try:
|
|
from zoneinfo import ZoneInfo
|
|
ZONES.append(("zoneinfo Europe/Brussels", ZoneInfo("Europe/Brussels")))
|
|
except Exception as err: # noqa: BLE001
|
|
print(f" zoneinfo unavailable here ({err}); graded on the EU rule only")
|
|
|
|
SPRING = datetime(2026, 3, 29, tzinfo=UTC) # 02:00 CET -> 03:00 CEST
|
|
SPRING27 = datetime(2027, 3, 28, tzinfo=UTC) # the next one, last Sunday of March
|
|
AUTUMN = datetime(2026, 10, 25, tzinfo=UTC) # 03:00 CEST -> 02:00 CET
|
|
SUMMER = datetime(2026, 7, 15, tzinfo=UTC)
|
|
WINTER = datetime(2026, 1, 15, tzinfo=UTC)
|
|
|
|
|
|
def slot_minutes(day_utc, hour, tz):
|
|
"""UTC minutes in the slot, over the LOCAL day of that date."""
|
|
t, out = day_utc - timedelta(hours=3), []
|
|
while t < day_utc + timedelta(hours=27):
|
|
if (t.astimezone(tz).date() == day_utc.date()
|
|
and in_start_slot(t, hour, tz)):
|
|
out.append(t)
|
|
t += timedelta(minutes=1)
|
|
return out
|
|
|
|
|
|
def one_block(mins):
|
|
"""`(start, minutes)` when the minutes form one contiguous run, else None."""
|
|
if not mins:
|
|
return None
|
|
for a, b in zip(mins, mins[1:]):
|
|
if b - a != timedelta(minutes=1):
|
|
return None
|
|
return mins[0], len(mins)
|
|
|
|
|
|
# =========================================================================== #
|
|
print("1. UTC stays what it was: the slot IS `now.hour == start_hour`")
|
|
t, same = datetime(2026, 1, 1, tzinfo=UTC), True
|
|
while t < datetime(2027, 1, 1, tzinfo=UTC):
|
|
for h in range(24):
|
|
if in_start_slot(t, h, UTC) != (t.hour == h):
|
|
same = False
|
|
t += timedelta(minutes=37)
|
|
check("over a year of instants, every hour 0-23: identical to the old test",
|
|
same)
|
|
check("a naive `now` is read as UTC, as before",
|
|
in_start_slot(datetime(2026, 7, 1, 19, 5), 19, UTC)
|
|
and not in_start_slot(datetime(2026, 7, 1, 18, 5), 19, UTC))
|
|
|
|
# =========================================================================== #
|
|
print("2. local time: 19 means 19:00 on the site's clock")
|
|
for label, tz in ZONES:
|
|
s = one_block(slot_minutes(SUMMER, 19, tz))
|
|
w = one_block(slot_minutes(WINTER, 19, tz))
|
|
check(f"[{label}] summer: 17:00-18:00 UTC (CEST), one hour",
|
|
s == (SUMMER + timedelta(hours=17), 60))
|
|
check(f"[{label}] winter: 18:00-19:00 UTC (CET), one hour",
|
|
w == (WINTER + timedelta(hours=18), 60))
|
|
check(f"[{label}] the old reading is gone: 19:00 UTC is not the slot "
|
|
f"in summer", not in_start_slot(SUMMER + timedelta(hours=19, minutes=5),
|
|
19, tz))
|
|
|
|
# =========================================================================== #
|
|
print("3. the DST change days, every start hour")
|
|
for label, tz in ZONES:
|
|
for name, day in (("spring", SPRING), ("spring 2027", SPRING27),
|
|
("autumn", AUTUMN)):
|
|
bad = []
|
|
for h in range(24):
|
|
b = one_block(slot_minutes(day, h, tz))
|
|
# the local day runs 23:00 UTC the day before to 22:00/23:00 UTC
|
|
if b is None or b[1] != 60:
|
|
bad.append((h, b))
|
|
check(f"[{label}] {name}: every hour 0-23 gives ONE slot of exactly "
|
|
f"60 minutes that day {bad or ''}", not bad)
|
|
sp2 = one_block(slot_minutes(SPRING, 2, tz))
|
|
check(f"[{label}] spring, hour 2 does not exist: it starts at 03:00 CEST "
|
|
f"(01:00 UTC), once", sp2 == (SPRING + timedelta(hours=1), 60))
|
|
check(f"[{label}] spring 2027 (28 March), hour 2: 01:00 UTC as well",
|
|
one_block(slot_minutes(SPRING27, 2, tz))
|
|
== (SPRING27 + timedelta(hours=1), 60))
|
|
check(f"[{label}] the evening before spring 2027, hour 19: 18:00 UTC "
|
|
f"(CET); that day, 17:00 UTC (CEST)",
|
|
one_block(slot_minutes(SPRING27 - timedelta(days=1), 19, tz))
|
|
== (SPRING27 - timedelta(hours=6), 60)
|
|
and one_block(slot_minutes(SPRING27, 19, tz))
|
|
== (SPRING27 + timedelta(hours=17), 60))
|
|
au2 = one_block(slot_minutes(AUTUMN, 2, tz))
|
|
check(f"[{label}] autumn, hour 2 happens twice: only the first, 02:00 CEST "
|
|
f"(00:00 UTC)", au2 == (AUTUMN, 60))
|
|
check(f"[{label}] autumn: the second 02:xx (CET) is not a slot",
|
|
not in_start_slot(AUTUMN + timedelta(hours=1, minutes=30), 2, tz))
|
|
check(f"[{label}] autumn, hour 19 still 18:00 UTC (CET)",
|
|
one_block(slot_minutes(AUTUMN, 19, tz))
|
|
== (AUTUMN + timedelta(hours=18), 60))
|
|
check(f"[{label}] spring, hour 19 already 17:00 UTC (CEST)",
|
|
one_block(slot_minutes(SPRING, 19, tz))
|
|
== (SPRING + timedelta(hours=17), 60))
|
|
|
|
|
|
# =========================================================================== #
|
|
class FakeStore:
|
|
def __init__(self):
|
|
self.data = {"phase": IDLE, "phase_started": None,
|
|
"last_completed": None, "last_start_attempt": None}
|
|
|
|
def set(self, k, v):
|
|
self.data[k] = v
|
|
|
|
def set_time(self, k, when):
|
|
self.data[k] = when
|
|
|
|
def get_time(self, k):
|
|
return self.data.get(k)
|
|
|
|
def get(self, k, default=None):
|
|
return self.data.get(k, default)
|
|
|
|
def save(self):
|
|
pass
|
|
|
|
|
|
def starts(day, hour, tz, *, abort_after=None, blocked=None):
|
|
"""Tick a due pack each minute around a day; the UTC instants in that
|
|
LOCAL day at which it entered DRAIN. `abort_after` aborts each drain that many minutes in (interlock);
|
|
`blocked(t)` defers it."""
|
|
m = Maintenance(MaintConfig(enabled=True, start_hour=hour), FakeStore(),
|
|
tz=tz)
|
|
out, since = [], None
|
|
t = day - timedelta(hours=2)
|
|
while t < day + timedelta(hours=26):
|
|
why = ""
|
|
if since is not None and abort_after is not None \
|
|
and t - since >= timedelta(minutes=abort_after):
|
|
why = "a test interlock"
|
|
if blocked is not None and blocked(t):
|
|
why = "the other battery is cycling"
|
|
was = m.phase
|
|
m.tick(t, soc=80, blocked_by=why)
|
|
if m.phase == DRAIN and was != DRAIN:
|
|
out.append(t)
|
|
since = t
|
|
if m.phase == IDLE:
|
|
since = None
|
|
t += timedelta(minutes=1)
|
|
return [x for x in out if x.astimezone(tz).date() == day.date()]
|
|
|
|
|
|
print("4. the state machine: one start per day, DST days included")
|
|
for label, tz in ZONES:
|
|
check(f"[{label}] spring, hour 2: one start, at 01:00 UTC (03:00 CEST)",
|
|
starts(SPRING, 2, tz, abort_after=5)
|
|
== [SPRING + timedelta(hours=1)])
|
|
# An aborted start waits RESTART_COOLDOWN_MIN (90). The repeated hour is
|
|
# 120 minutes long, so without the first-pass rule it would start again.
|
|
check(f"[{label}] autumn, hour 2, aborted after 5 min: NO second start "
|
|
f"in the repeated hour",
|
|
starts(AUTUMN, 2, tz, abort_after=5) == [AUTUMN])
|
|
check(f"[{label}] autumn, hour 2, deferred for the whole first pass: it "
|
|
f"does not start in the second pass either",
|
|
starts(AUTUMN, 2, tz, blocked=lambda t: t < AUTUMN + timedelta(hours=1))
|
|
== [])
|
|
check(f"[{label}] summer, hour 19: one start at 17:00 UTC",
|
|
starts(SUMMER, 19, tz, abort_after=5)
|
|
== [SUMMER + timedelta(hours=17)])
|
|
|
|
# =========================================================================== #
|
|
print("5. where the start is shown (homeview) is where it happens")
|
|
for label, tz in ZONES:
|
|
ok = True
|
|
for day in (SPRING, AUTUMN, SUMMER):
|
|
for h in (0, 2, 3, 10, 19, 23):
|
|
first = next(iter(slot_minutes(day, h, tz)), None)
|
|
if first is None:
|
|
continue
|
|
for probe in (first - timedelta(hours=5), first,
|
|
first + timedelta(minutes=59)):
|
|
if next_slot_start(probe, h, tz) != first:
|
|
ok = False
|
|
check(f"[{label}] next_slot_start() lands on the slot's first minute, "
|
|
f"from before it and from inside it", ok)
|
|
|
|
# =========================================================================== #
|
|
print("6. which zone: Home Assistant, else the container's TZ, else UTC")
|
|
_FAKE = {"Europe/Brussels": ZONES[-1][1], "America/New_York": object(),
|
|
"UTC": UTC}
|
|
_real_load = MT._load_zone
|
|
MT._load_zone = lambda n: ((_FAKE[n], "") if n in _FAKE
|
|
else (None, f"{n!r} is not a known zone"
|
|
if n else "none given"))
|
|
try:
|
|
tz, name, src, _ = site_zone("Europe/Brussels", "", "America/New_York")
|
|
check("Home Assistant's zone wins over the container's TZ",
|
|
name == "Europe/Brussels" and src == "ha"
|
|
and tz is _FAKE["Europe/Brussels"])
|
|
tz, name, src, why = site_zone(None, "ClientConnectorError",
|
|
"America/New_York")
|
|
check("Home Assistant unavailable: the container's TZ, and why",
|
|
name == "America/New_York" and src == "tz"
|
|
and why == "ClientConnectorError")
|
|
tz, name, src, why = site_zone("Mars/Base", "", "Europe/Brussels")
|
|
check("an unknown Home Assistant zone falls back too",
|
|
name == "Europe/Brussels" and src == "tz" and "Mars/Base" in why)
|
|
tz, name, src, why = site_zone(None, "no answer", None)
|
|
check("neither: UTC", tz is UTC and src == "utc" and "no answer" in why)
|
|
finally:
|
|
MT._load_zone = _real_load
|
|
if len(ZONES) > 1:
|
|
check("the real loader resolves a real name and refuses a made-up one",
|
|
MT._load_zone("Europe/Brussels")[0] is not None
|
|
and MT._load_zone("Mars/Base")[0] is None)
|
|
check("the real loader refuses an empty name without raising",
|
|
MT._load_zone("")[0] is None and MT._load_zone(None)[0] is None)
|
|
|
|
# =========================================================================== #
|
|
print("7. the default is 19, an explicit value is honoured")
|
|
cfg_text = (HERE / "config.yaml").read_text()
|
|
opts_block = cfg_text.split("\nschema:")[0]
|
|
check("config.yaml options: ships 19",
|
|
re.search(r"^ maintenance_start_hour: 19$", opts_block, re.M) is not None)
|
|
check("config.yaml schema: still int(0,23)",
|
|
re.search(r"^ maintenance_start_hour: int\(0,23\)$", cfg_text, re.M)
|
|
is not None)
|
|
check("the in-code mirror says 19",
|
|
MAINT_OPTION_DEFAULTS["maintenance_start_hour"] == 19
|
|
and MaintConfig().start_hour == 19
|
|
and MaintConfig.from_options({}).start_hour == 19)
|
|
check("an explicit 10 stays 10 (a stored option is never rewritten)",
|
|
MaintConfig.from_options({"maintenance_start_hour": 10}).start_hour == 10)
|
|
_eu = ZONES[0][1]
|
|
_m10 = Maintenance(MaintConfig.from_options(
|
|
{"maintenance_enabled": True, "maintenance_start_hour": 10}),
|
|
FakeStore(), tz=_eu)
|
|
check("...and now means 10:00 local: 08:00 UTC in summer, not 10:00 UTC",
|
|
_m10.wants_start(SUMMER + timedelta(hours=8, minutes=1))
|
|
and not _m10.wants_start(SUMMER + timedelta(hours=10, minutes=1)))
|
|
check("a Maintenance built with no zone reads UTC, as before",
|
|
Maintenance(MaintConfig(), FakeStore()).tz is UTC)
|
|
|
|
|
|
# =========================================================================== #
|
|
print("8. startup: Home Assistant's zone is asked for and reaches the machines")
|
|
|
|
|
|
class _Resp:
|
|
def __init__(self, body, status=200):
|
|
self.body, self.status = body, status
|
|
|
|
async def __aenter__(self):
|
|
return self
|
|
|
|
async def __aexit__(self, *a):
|
|
return False
|
|
|
|
async def json(self, content_type=None):
|
|
if isinstance(self.body, Exception):
|
|
raise self.body
|
|
return self.body
|
|
|
|
async def text(self):
|
|
return "{}"
|
|
|
|
|
|
class _Session:
|
|
def __init__(self, zone_body):
|
|
self.zone_body, self.asked = zone_body, []
|
|
|
|
async def __aenter__(self):
|
|
return self
|
|
|
|
async def __aexit__(self, *a):
|
|
return False
|
|
|
|
def get(self, url, **kw):
|
|
self.asked.append(url)
|
|
if url.endswith("/core/api/config"):
|
|
return _Resp(self.zone_body)
|
|
return _Resp({"data": {}})
|
|
|
|
|
|
s = _Session({"time_zone": "Europe/Brussels"})
|
|
check("ha_time_zone() reads /core/api/config's time_zone",
|
|
asyncio.run(M.ha_time_zone(s, "tok")) == ("Europe/Brussels", "")
|
|
and s.asked == [M.CORE_API + "/config"])
|
|
check("...and a failure is None plus the reason, never a raise",
|
|
asyncio.run(M.ha_time_zone(_Session(OSError("down")), "tok"))
|
|
== (None, "OSError: down")
|
|
and asyncio.run(M.ha_time_zone(_Session({}), "tok"))[0] is None)
|
|
|
|
|
|
class _Built(Exception):
|
|
pass
|
|
|
|
|
|
def amain_zone(zone_body, env_tz):
|
|
"""Run the real amain() up to the Controller; the tz its machines get."""
|
|
got, logs = {}, []
|
|
|
|
class _Capture(logging.Handler):
|
|
def emit(self, rec):
|
|
logs.append((rec.levelname, rec.getMessage()))
|
|
|
|
class _Hass:
|
|
async def limits(self, e):
|
|
return (-5000.0, 5000.0, 1.0)
|
|
|
|
async def mqtt_service(self):
|
|
raise RuntimeError("no broker in this test")
|
|
|
|
real = {k: getattr(M, k) for k in ("load_options", "HomeAssistant",
|
|
"validate_startup_config",
|
|
"mode_announcement", "Controller")}
|
|
real_session, real_store = M.aiohttp.ClientSession, None
|
|
import app.store
|
|
real_store = app.store.Store
|
|
|
|
def _controller(opts, hass, store, pub, *, stop, site_tz=None, **kw):
|
|
got["tz"] = site_tz
|
|
raise _Built()
|
|
|
|
M.load_options = lambda: {"setpoint_entity": "number.sp"}
|
|
M.HomeAssistant = lambda session, **kw: _Hass()
|
|
M.validate_startup_config = lambda *a, **k: []
|
|
M.mode_announcement = lambda opts, store: ((), ())
|
|
M.Controller = _controller
|
|
M.aiohttp.ClientSession = lambda *a, **k: _Session(zone_body)
|
|
app.store.Store = lambda *a, **k: object()
|
|
old_tz = os.environ.pop("TZ", None)
|
|
if env_tz:
|
|
os.environ["TZ"] = env_tz
|
|
h = _Capture()
|
|
logging.getLogger("goodwe").addHandler(h)
|
|
try:
|
|
asyncio.run(M.amain())
|
|
except _Built:
|
|
pass
|
|
finally:
|
|
logging.getLogger("goodwe").removeHandler(h)
|
|
for k, v in real.items():
|
|
setattr(M, k, v)
|
|
M.aiohttp.ClientSession = real_session
|
|
app.store.Store = real_store
|
|
os.environ.pop("TZ", None)
|
|
if old_tz is not None:
|
|
os.environ["TZ"] = old_tz
|
|
return got.get("tz"), [m for lvl, m in logs if "Maintenance start" in m], logs
|
|
|
|
|
|
MT._load_zone = lambda n: ((_FAKE[n], "") if n in _FAKE
|
|
else (None, "none given"))
|
|
try:
|
|
tz, lines, _ = amain_zone({"time_zone": "Europe/Brussels"}, "America/New_York")
|
|
check("amain(): Home Assistant's zone reaches the Controller",
|
|
tz is _FAKE["Europe/Brussels"])
|
|
check("...logged once, naming the zone and the source",
|
|
lines == ["Maintenance start hour is local time in Europe/Brussels, "
|
|
"from Home Assistant."])
|
|
tz, lines, _ = amain_zone(OSError("down"), "America/New_York")
|
|
check("amain(), Home Assistant unavailable: the container's TZ, said "
|
|
"plainly", tz is _FAKE["America/New_York"]
|
|
and lines == ["Maintenance start hour is local time in "
|
|
"America/New_York, from the add-on's time zone. Home "
|
|
"Assistant gave none: OSError: down."])
|
|
tz, lines, logs = amain_zone(OSError("down"), None)
|
|
check("amain(), neither: UTC, with a WARNING line",
|
|
tz is UTC and any(lvl == "WARNING" and "uses UTC" in m
|
|
for lvl, m in logs))
|
|
finally:
|
|
MT._load_zone = _real_load
|
|
|
|
|
|
# =========================================================================== #
|
|
print("9. nothing else in control moves: the same start instant, the same "
|
|
"commands")
|
|
import opt11_identity as rig # noqa: E402
|
|
import safety78_identity as s78 # noqa: E402
|
|
|
|
# rig's clock starts 2026-09-20 11:00 UTC = 13:00 CEST. The same drain,
|
|
# asked for as UTC hour 11 (the old reading) and as local hour 13 (the new).
|
|
|
|
|
|
def rows_with(start_hour, tz, n=1, both=False):
|
|
"""The rig's rows; `both=True` also returns the Controller."""
|
|
real_drain, real_ctl = s78.DRAIN, rig.Controller
|
|
s78.DRAIN = {"maintenance_enabled": True,
|
|
"maintenance_start_hour": start_hour}
|
|
rig.Controller = functools.partial(real_ctl, site_tz=tz)
|
|
try:
|
|
# s78.run() wraps rig.Controller once more for its own tariff db.
|
|
rows, c = s78.run(n, dur=600.0)
|
|
return (rows, c) if both else rows
|
|
finally:
|
|
s78.DRAIN, rig.Controller = real_drain, real_ctl
|
|
|
|
|
|
_utc11 = rows_with(11, None)
|
|
_loc13 = rows_with(13, _eu)
|
|
check("precondition: the drain really starts in the UTC-11 run",
|
|
any("drain" in r[-1][4] for r in _utc11))
|
|
check(f"start hour 11 UTC and 13 Europe/Brussels: every command, claim, "
|
|
f"reason and log line identical over {len(_utc11)} cycles",
|
|
_utc11 == _loc13)
|
|
_utc12 = rows_with(12, None)
|
|
_loc14 = rows_with(14, _eu)
|
|
check("precondition: no drain in the not-yet-due-hour run",
|
|
not any("drain" in r[-1][4] for r in _utc12))
|
|
check("outside the start hour (12 UTC / 14 local): identical as well",
|
|
_utc12 == _loc14)
|
|
check("...and the zone matters: hour 11 read in Brussels does NOT start",
|
|
not any("drain" in r[-1][4] for r in rows_with(11, _eu)))
|
|
|
|
# =========================================================================== #
|
|
print("10. two inverters: every pack reads the start hour in local time")
|
|
_r2u, _c2u = rows_with(11, None, n=2, both=True)
|
|
_r2l, _c2l = rows_with(13, _eu, n=2, both=True)
|
|
check("N=2: both packs get the site's zone, not just the first",
|
|
len(_c2l.maints) == 2 and all(mm.tz is _eu for mm in _c2l.maints))
|
|
check("N=2 precondition: a drain starts in the UTC-11 run",
|
|
any(DRAIN in r[-1][4] for r in _r2u))
|
|
check(f"N=2: start hour 11 UTC and 13 Europe/Brussels identical over "
|
|
f"{len(_r2u)} cycles", _r2u == _r2l)
|
|
check("N=2: hour 11 read in Brussels does NOT start either pack",
|
|
not any(DRAIN in r[-1][4] for r in rows_with(11, _eu, n=2)))
|
|
|
|
|
|
def pair_starts(day0, days, hour, tz, cycle_min=70):
|
|
"""Two due packs, ticked each minute through `arbitrate()` as main.py
|
|
does; a cycle takes about `cycle_min` minutes. `(pack, "DD HH:MM" UTC)`s."""
|
|
cfg = MaintConfig(enabled=True, start_hour=hour, hold_min=10)
|
|
ms = [Maintenance(cfg, FakeStore(), name=u, tz=tz)
|
|
for u in ("A", "B")]
|
|
ms[0].store.set_time("last_completed", day0 - timedelta(days=60))
|
|
ms[1].store.set_time("last_completed", day0 - timedelta(days=40))
|
|
half = (cycle_min - 10) // 2
|
|
soc, out = [80.0, 80.0], []
|
|
t = day0 - timedelta(hours=3)
|
|
while t < day0 + timedelta(days=days):
|
|
blocks = MT.arbitrate(ms, t)
|
|
for i, mm in enumerate(ms):
|
|
was = mm.phase
|
|
mm.tick(t, soc=soc[i], blocked_by=blocks[i])
|
|
if mm.phase == DRAIN:
|
|
soc[i] -= (80.0 - cfg.soc_floor) / half
|
|
elif mm.phase == MT.CHARGE:
|
|
soc[i] += (cfg.soc_target - cfg.soc_floor) / half
|
|
if mm.phase == DRAIN and was != DRAIN:
|
|
out.append((mm.name, t))
|
|
t += timedelta(minutes=1)
|
|
return [(n, x.strftime("%d %H:%M")) for n, x in out]
|
|
|
|
|
|
def _at(*pairs):
|
|
return [(n, x.strftime("%d %H:%M")) for n, x in pairs]
|
|
|
|
|
|
for label, tz in ZONES:
|
|
# 24 Oct (CEST) and 25 Oct (CET): 19:00 local is 17:00 then 18:00 UTC.
|
|
got = pair_starts(AUTUMN - timedelta(days=1), 2, 19, tz)
|
|
check(f"[{label}] N=2 across the autumn change, hour 19: A at 19:00 "
|
|
f"local on the 24th, B at 19:00 local on the 25th {got}",
|
|
got == _at(("A", AUTUMN - timedelta(hours=7)),
|
|
("B", AUTUMN + timedelta(hours=18))))
|
|
got = pair_starts(SPRING27 - timedelta(days=1), 2, 19, tz)
|
|
check(f"[{label}] N=2 across the spring 2027 change, hour 19: 18:00 UTC "
|
|
f"then 17:00 UTC {got}",
|
|
got == _at(("A", SPRING27 - timedelta(hours=6)),
|
|
("B", SPRING27 + timedelta(hours=17))))
|
|
# Hour 2 on 25 Oct: A takes the first pass (00:00 UTC) and is done in its
|
|
# second pass. B must not start there - it waits for the next day.
|
|
got = pair_starts(AUTUMN, 2, 2, tz)
|
|
check(f"[{label}] N=2, autumn, hour 2: B does not start in the repeated "
|
|
f"hour after A is done; it starts at 02:00 CET on the 26th {got}",
|
|
got == _at(("A", AUTUMN),
|
|
("B", AUTUMN + timedelta(days=1, hours=1))))
|
|
got = pair_starts(SPRING27, 2, 2, tz)
|
|
check(f"[{label}] N=2, spring 2027, hour 2: A at 03:00 CEST (01:00 UTC), "
|
|
f"B the next night at 02:00 CEST (00:00 UTC) {got}",
|
|
got == _at(("A", SPRING27 + timedelta(hours=1)),
|
|
("B", SPRING27 + timedelta(days=1))))
|
|
|
|
print()
|
|
if fails:
|
|
print(f"{len(fails)} FAILED: {', '.join(fails)}")
|
|
sys.exit(1)
|
|
print("all checks passed")
|