UX-54 round 5: give blocks end at the floor point, the tab on the clock-change days
checks / unit (pull_request) Failing after 50m11s
infra-pin / fresh (pull_request) Successful in 1s

daycurve.curve() puts a point where a giving battery reaches the floor
inside a step (_floor_cut(), as it already does for a grid charge reaching
its level), so _stops() ends a give block in the quarter control stops. A
walk where control moves the packs with its own model (not the curve's
levels) matches the tab quarter by quarter in 11 cases; 25/45 % with the
peak guard: control 17:00-20:15, block 17:00-20:06.

SolarToday keys an hour by its start taken back from the UTC instant, so
the two 02:00 hours of the autumn change stay apart. test_ux54_dst.py builds
the tab (forecast_payload()) at every quarter of 27-29 March and
30 October-1 November 2027, modes on and off, in the no-tzdata zone and,
where the host has tzdata, ZoneInfo, every call under a time limit raised
as a BaseException that forecast_payload() cannot swallow.

give_later: "No block shows while a battery has no energy to give." /
"Er verschijnt geen blok zolang een batterij niets kan leveren."
DECISIONS: what the walks prove, the modes-off charge drawn on a sunny day,
the clock-change days.

No decision, command, timing or option key changes.

Co-Authored-By: Claude <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Bdqr7PiaHcBnzPituu2yTC
This commit is contained in:
glenn schrooyen
2026-10-27 15:54:33 +01:00
co-authored by Claude
parent 00c8ae354c
commit 8dbf831854
6 changed files with 536 additions and 24 deletions
@@ -31,22 +31,28 @@ battery's expected level with its blocks. It follows the add-on, as Home does
block where the emptiest battery is at the floor by that count, and control
does not give there. A charge is read from the floor: a full battery covers
the house before the cheap hour, and control charges it back then. Walked
every 15 minutes on a cloudy day with the house (`test_ux54.py` part 7,
cases A-C), control charges and gives in exactly the quarters the tab has a
block.
every 15 minutes on a cloudy day, the packs moved by control and a 400 W
house (`test_ux54.py` part 7, cases A-D, one price quarter per leg),
control charges and gives in exactly the quarters the tab has a block.
- **A block lasts as long as control acts.** A cheap run is sized for all the
batteries charging together, so a slower one is still below
`modetrade.charge_to()` when it ends, and control plans the next cheap slots
(`Controller._price_runs()`). `dayview._levels()` asks the same functions,
`modetrade.energy()` and `priceruns.plan()`, with the packs at the curve's
level, and draws the grid block to where control stops. Each battery's block
then ends where the curve has it at its own level (the curve puts a point
there), and a give block where it reaches the floor (a step end, within the
0.1 % the tab shows). Control, walked every 15 minutes through tomorrow
(`test_ux54.py` part 7), charges and gives in exactly the quarters the tab
has a block. A GIVE role counts only where control's own discharge envelope
(`packshare.envelope()`) is open; the walk puts each pack's cut-out at the
floor.
then ends where the curve has it at its own level, and a give block where
it reaches the floor: the curve puts a point at both (`daycurve.curve()`
cuts the step there), and the block ends at the first point where the level
the tab shows (0.1 %) is there. What is proven: control walked every 15
minutes through tomorrow with its own model of the packs, not the curve's
levels (`_mode_roles()` each quarter; the house from the profile, the sun
at 100 % trust, a GIVE role counted only where control's own discharge
envelope `packshare.envelope()` is open, the cut-out at the floor), charges
and gives in exactly the quarters the tab has a block in 11 cases
(`test_ux54.py` part 7, `own_walk()`), e.g. 25/45 % with the peak guard:
control gives 17:00-20:15, the block is 17:00-20:06. With a cut-out under
the floor, or a house or sun that differs from the forecast and profile,
control differs (Limits).
- **The sun** is what `solarfc` reads. An hourly split (Solcast
`detailedForecast`, Open-Meteo `wh_period`) is used as given. A daily total
(Forecast.Solar, and any other state-only sensor, the owner's three "today
@@ -89,7 +95,7 @@ is `SolarToday.add()` after the solar read, in a `try`, in memory.
at its level (`charge_to()`) and giving at the floor.
- Today, a give run in the plan on a battery that is empty now has no block on
Home (no energy to give). The tab equals Home, and says so: "The add-on plans
to give power. No block shows while the battery has no energy to give." The
to give power. No block shows while a battery has no energy to give." The
curve gives in that run if the sun fills the battery first.
- With no forecast for tomorrow the curve stops at the first hour of the sun,
`solarfc.BELL_FROM_H` (06:00): `Forecast.between()` has no answer past it. The
@@ -103,11 +109,22 @@ is `SolarToday.add()` after the solar read, in a `try`, in memory.
hour, net of the solar the house used.
- Modes off: the count knows no sun and no house. A battery at the floor that
the sun fills before a give gets no block until then (the tab says "The
add-on plans to give power. No block shows while the battery has no energy
to give."), and a small charge may be used by the house before the give.
add-on plans to give power. No block shows while a battery has no energy to
give."), and a small charge may be used by the house before the give. The
other way round, a charge is drawn even when the sun fills the batteries
before it: at the ceiling control then takes none (a cheap quarter at 13:00
on a sunny day: the tab draws the charge, control charges nothing at 95 %
or 100 %).
A leg is the verdict's as it stands now, one price period, as on Home; with
several equal prices in a row control takes the next one as each passes.
- Measured solar starts when the add-on starts; the tab says from when.
- Clock changes: the tab walks local hours with `dayplan._hour_slices()`,
which steps in UTC (SAFETY-123), and keys measured solar by the start of
the local hour taken back from the UTC instant, so the spring day has 23
hours, the autumn day 25 (two of them 02:00), and a request always returns
(`test_ux54_dst.py`, every quarter of both change days, with and without
tzdata). The solar bell is spread over hours from local midnight, so on
those two days it is an hour off the clock (SAFETY-123, not changed here).
## Found, not changed
+50 -9
View File
@@ -31,6 +31,8 @@ from .dayplan import EFF, _hour_slices, _local
from .modetrade import CHARGE, GIVE, HOLD, HOUSE, REST
EPS = 1e-9
MIN_CUT = timedelta(seconds=1) # no step is cut closer to its ends
MAX_FLOOR_CUTS = 400 # a bound on the floor cuts of one curve
GAP_S = 60.0 # a longer gap between readings is not counted
MIN_COVER = 0.5 # share of the elapsed hour a reading must cover
@@ -88,10 +90,42 @@ def _spread(amount, caps, weights):
return got, max(0.0, left)
def _give_caps(roles, level, by, floor, hrs):
"""`(first, rest)`: what each battery that may give can give in a step
of `hrs` hours (its rating, or what it holds above the floor), those in a
give run first."""
caps = {n: min(by[n].rating_kw * hrs,
max(0.0, level[n] - floor) / 100.0
* by[n].capacity_kwh * EFF)
for n, r in roles.items() if r[2]}
first = {n: c for n, c in caps.items() if roles[n][0] == GIVE}
return first, {n: c for n, c in caps.items() if n not in first}
def _floor_cut(a, z, net, roles, level, by, floor, hrs):
"""Where in the step `[a, z)` the first battery that gives reaches the
floor, or None. A shortfall is spread over a step at a steady rate and
shared by what each can give, so a group that holds less than its part
of the shortfall, one of them short of its rating, is empty at
`held / part` of the step."""
left, cut = -net, None
for grp in _give_caps(roles, level, by, floor, hrs):
held = sum(grp.values())
if held > EPS and left > held + EPS and any(
c < by[n].rating_kw * hrs * (1.0 - 1e-9)
for n, c in grp.items()):
t = a + (z - a) * (held / left)
if a + MIN_CUT < t < z - MIN_CUT and (cut is None or t < cut):
cut = t
left = max(0.0, left - held)
return cut
def curve(batts, start, end, sun_kwh, use_kwh, *, floor_pct, ceiling_pct,
trust=1.0):
"""`{name: [(t, pct), ...]}` from `start` (each battery's level now) to
`end`, a point at every hour edge and every block edge. A battery's list
`end`, a point at every hour edge, every block edge and where a battery
reaches its grid level or, giving, the floor. A battery's list
stops where `sun_kwh(a, b)` is None (the sun is not known) or its
`stop_at` comes. `sun_kwh` and `use_kwh` give kWh for the span `[a, b)`.
Spare solar is `trust x max(0, sun - use)`, and what is stored loses
@@ -109,6 +143,7 @@ def curve(batts, start, end, sun_kwh, use_kwh, *, floor_pct, ceiling_pct,
level = {b.name: float(b.soc) for b in batts}
out = {b.name: [(start, level[b.name])] for b in batts}
alive = {b.name for b in batts}
floor_cuts = 0
i = 0
while i < len(ts) - 1:
a, z = ts[i], ts[i + 1]
@@ -147,6 +182,14 @@ def curve(batts, start, end, sun_kwh, use_kwh, *, floor_pct, ceiling_pct,
continue
roles = {b.name: _role(b, a, bool(want), b.name in want)
for b in batts if b.name in alive}
# A battery that gives all it holds inside the step gets a point
# where it reaches the floor, the same way.
cut = _floor_cut(a, z, net, roles, level, by, floor, hrs) \
if net < -EPS and floor_cuts < MAX_FLOOR_CUTS else None
if cut is not None:
floor_cuts += 1
ts.insert(i + 1, cut)
continue
for n, to in want.items(): # charging from the grid
b = by[n]
room = (to - level[n]) / 100.0 * b.capacity_kwh
@@ -165,12 +208,7 @@ def curve(batts, start, end, sun_kwh, use_kwh, *, floor_pct, ceiling_pct,
for n, v in got.items():
level[n] += v * EFF / by[n].capacity_kwh * 100.0
elif net < -EPS:
caps = {n: min(by[n].rating_kw * hrs,
max(0.0, level[n] - floor) / 100.0
* by[n].capacity_kwh * EFF)
for n, r in roles.items() if r[2]}
first = {n: c for n, c in caps.items() if roles[n][0] == GIVE}
rest = {n: c for n, c in caps.items() if n not in first}
first, rest = _give_caps(roles, level, by, floor, hrs)
left = -net
for grp in (first, rest):
got, left = _spread(left, grp, dict(grp))
@@ -234,6 +272,9 @@ class SolarToday:
def _hour_start(t: datetime) -> datetime:
"""The start of `t`'s local hour, in UTC. Taken back from the UTC instant,
never from the wall clock, so each of the two 02:00 hours of the autumn
change has its own (`dayplan._hour_slices()`, SAFETY-123)."""
lt = _local(t)
return lt.replace(minute=0, second=0, microsecond=0).astimezone(
timezone.utc)
return t.astimezone(timezone.utc) - timedelta(
minutes=lt.minute, seconds=lt.second, microseconds=lt.microsecond)
+1 -1
View File
@@ -456,7 +456,7 @@
"measured_from": "Measured solar is counted from {at}, when the add-on started.",
"tomorrow_total": "Solar expected tomorrow: {kwh}, we count on {count}.",
"no_blocks": "No charge or give blocks planned.",
"give_later": "The add-on plans to give power. No block shows while the battery has no energy to give.",
"give_later": "The add-on plans to give power. No block shows while a battery has no energy to give.",
"ahead": "The add-on plans about a day ahead. Blocks for later hours appear as they come near."
},
"region": {
+1 -1
View File
@@ -384,7 +384,7 @@
"measured_from": "Gemeten zon telt vanaf {at}, toen de add-on startte.",
"tomorrow_total": "Verwachte zon morgen: {kwh}, we rekenen op {count}.",
"no_blocks": "Geen laad- of leverblokken gepland.",
"give_later": "De add-on plant stroom te leveren. Er verschijnt geen blok zolang de batterij geen stroom heeft.",
"give_later": "De add-on plant stroom te leveren. Er verschijnt geen blok zolang een batterij niets kan leveren.",
"ahead": "De add-on plant ongeveer een dag vooruit. Blokken voor latere uren verschijnen als ze naderen."
},
"region": {"preset": "Heffingen en btw: {region}, cijfers van {year}.", "own": "Heffingen en btw: uw eigen bedrag."},
+160
View File
@@ -898,6 +898,153 @@ check("F2: the same with the batteries part full and nearly full at 22:00",
for x in o3))
# Control walked with its own model of the packs, not the curve's levels:
# every 15 minutes the real `_mode_roles()`; a battery in CHARGE charges at
# its rating up to `charge_to()`; spare sun goes to the batteries whose range
# may charge, by room, up to the ceiling; a shortfall is covered by those
# whose range may give and whose own discharge envelope is open, by what
# they hold above the cut-out (at the floor), each at most its rating. The
# house is the profile the tab reads, the sun the forecast at 100 % trust.
# Per battery and quarter tomorrow, control charges (gives) exactly where the
# tab has a grid (give) block, both ways, no tolerance.
def own_walk(soc, *, guard=None, sun=(0.0, 0.0)):
now0 = t22
c = S.make("dynamic", now0, soc=soc)
cut_out_at_floor(c)
floor_, ceil_ = arbitrage.soc_bounds(c.o)
c.tariff_plan = S.quarters(now0, night_evening, days=3)
c.tariff_at = now0
if guard is not None:
c.peak_guard = guard
c.solar_trust_pct = 100
c.solar_fc = fc = S.forecast(now0, sun[0], sun[1])
c._mode_roles(now0, None)
out = tab(c, now0)
house_w, base_w = c.house_profile.estimator()
d0, _d1 = homeview.local_day(now0)
seen = {inv.name: {} for inv in c.inverters}
t = now0
while t < d0 + timedelta(hours=48):
c.tariff_at = t
tick = c._mode_roles(t, None)
targets = c.plan_targets()
packs = {p.name: p for p in c._mode_packs()}
house = max(house_w(S.brussels(t).hour), base_w) / 1000.0 * 0.25
net = (dayview.sun_kwh(fc, now0, t, t + Q) or 0.0) - house
takers, givers = [], []
for inv in c.inverters:
role = tick.roles.get(inv.name)
lo, hi = tick.ranges.get(inv.name, (0.0, 0.0))
act = None
if role == modetrade.CHARGE:
to = modetrade.charge_to(packs[inv.name], ceil_, targets)
step = min(inv.max_w / 1000.0 * 0.25 * dayplan.EFF,
max(0.0, to - inv.soc) / 100.0 * inv.capacity_kwh)
act = "grid" if step > 1e-9 else None
inv.soc += step / inv.capacity_kwh * 100.0
else:
if lo < 0.0 and inv.soc < ceil_:
takers.append(inv)
if hi > 0.0 and may_give(inv):
givers.append(inv)
act = "give" if role == modetrade.GIVE else None
seen[inv.name][t] = act
if net > 0.0:
room = {i.name: (ceil_ - i.soc) / 100.0 * i.capacity_kwh
/ dayplan.EFF for i in takers}
tot = sum(room.values())
for i in takers:
part = min(net * room[i.name] / tot, room[i.name],
i.max_w / 1000.0 * 0.25) if tot > 0.0 else 0.0
i.soc += part * dayplan.EFF / i.capacity_kwh * 100.0
elif net < 0.0:
hold = {i.name: max(0.0, i.soc - floor_) / 100.0 * i.capacity_kwh
for i in givers}
tot = sum(hold.values())
for i in givers:
part = min(-net * hold[i.name] / tot,
i.max_w / 1000.0 * 0.25) if tot > 0.0 else 0.0
i.soc = max(floor_, i.soc - part / dayplan.EFF
/ i.capacity_kwh * 100.0)
t += Q
return out, seen, d0
def own_match(out, seen, d0):
"""`[(name, kind, hour, who)]`: the quarters tomorrow where control and
the tab disagree (who acts there alone)."""
bad = []
for i, name in enumerate(seen):
for kind in ("grid", "give"):
blocks = [r for r in out["days"][1]["batteries"][i]["blocks"]
if r["kind"] == kind]
for t, a in seen[name].items():
h = (t - d0).total_seconds() / 3600.0 - 24.0
if h < 0.0:
continue
inside = any(r["start_h"] - 1e-6 <= h < r["end_h"] - 1e-6
for r in blocks)
if (a == kind) != inside:
bad.append((name, kind, round(h, 2),
"control" if a == kind else "tab"))
return bad
def give_span(seen, d0, name):
hs = [(t - d0).total_seconds() / 3600.0 - 24.0
for t, a in seen[name].items()
if a == "give" and (t - d0).total_seconds() >= 86400]
return (round(min(hs), 2), round(max(hs) + 0.25, 2)) if hs else None
own = {}
for key, soc, g, sun in (("20/20", (20.0, 20.0), None, (0.0, 0.0)),
("55/40", (55.0, 40.0), None, (0.0, 0.0)),
("90/70", (90.0, 70.0), None, (0.0, 0.0)),
("30/80", (30.0, 80.0), None, (0.0, 0.0)),
("60/60 guard", (60.0, 60.0), guard, (0.0, 0.0)),
("20/20 guard", (20.0, 20.0), guard, (0.0, 0.0)),
("30/30 guard", (30.0, 30.0), guard, (0.0, 0.0)),
("25/45 guard", (25.0, 45.0), guard, (0.0, 0.0)),
("20/20 guard 14 kWh", (20.0, 20.0), guard,
(14.0, 14.0)),
("20/20 guard 6 kWh", (20.0, 20.0), guard,
(6.0, 6.0)),
("20/20 8 kWh", (20.0, 20.0), None, (8.0, 8.0))):
o_w, s_w, d_w = own_walk(soc, guard=g, sun=sun)
own[key] = (o_w, s_w, d_w, own_match(o_w, s_w, d_w))
print(f" own walk {key}: tab",
[[(r["kind"], r["from"], r["to"]) for r in b["blocks"]]
for b in o_w["days"][1]["batteries"]],
"control gives", {n: give_span(s_w, d_w, n) for n in s_w},
"mismatched", len(own[key][3]), own[key][3][:4])
check("F2, control walked with its own model of the packs (not the curve's "
"levels): in all 11 cases it charges and gives in exactly the quarters "
"the tab has a block of that kind, on both batteries",
all(not v[3] for v in own.values()))
o_g, s_g, d_g, _b = own["25/45 guard"]
check("...25/45 % with the peak guard: control gives 17:00-20:15 on both "
"(the packs empty in the 20:00 quarter), and the tab's give block "
"starts at 17:00 and ends inside that quarter, at the curve's floor "
"point",
all(give_span(s_g, d_g, n) == (17.0, 20.25) for n in s_g)
and all([(r["kind"], r["from"]) for r in b["blocks"]] == [("give",
"17:00")]
and 20.0 < b["blocks"][0]["end_h"] <= 20.25
for b in o_g["days"][1]["batteries"]))
o_6, s_6, d_6, _b = own["20/20 guard 6 kWh"]
check("...20/20 % with the guard and 6 kWh of sun: control gives 17:00-17:15, "
"and the tab's give block ends inside that quarter",
all(give_span(s_6, d_6, n) == (17.0, 17.25) for n in s_6)
and all(len(b["blocks"]) == 1 and 17.0 < b["blocks"][0]["end_h"] <= 17.25
for b in o_6["days"][1]["batteries"]))
fl_g = arbitrage.soc_bounds(S.make("dynamic", t22).o)[0]
check("...and the curve has a point exactly at the floor where the block "
"ends (no step end)",
all([v for h, v in b["levels"] if abs(h - b["blocks"][0]["end_h"])
< 1e-3] == [fl_g] for b in o_g["days"][1]["batteries"]))
def off_prices(h):
if 23.0 <= h < 24.0:
return 0.05
@@ -1065,6 +1212,19 @@ check("F2, modes off: one battery at the floor and one at 60 %: no give block "
all(not [r for r in b if r[0] == "give"] for b in kinds(od))
and all(not v for v in acts(sd, dd, "give").values())
and off_match(od, sd, dd))
c_dn = S.make("off", t22, soc=(20.0, 60.0))
c_dn.tariff_plan = S.quarters(t22, one_quarter(13.0, 7.0), days=3)
c_dn.tariff_at = t22
od_nl = tab(c_dn, t22, "nl")
check("...and B, which has energy, is not told it has none: the line names "
"'a battery' (the emptiest one holds the site back), EN and NL",
all(b["line"] == say("give_later") for b in od["days"][1]["batteries"])
and say("give_later").endswith(
"No block shows while a battery has no energy to give.")
and all(b["line"] == say("give_later", "nl")
for b in od_nl["days"][1]["batteries"])
and say("give_later", "nl").endswith(
"Er verschijnt geen blok zolang een batterij niets kan leveren."))
# C: full at 22:00 and a cheap quarter at 01:00: the house uses some first,
# so control charges then, and the tab has that block.
cc, oc, sc, dc = off_sweep(one_quarter(1.0, 17.0), t22, (95.0, 95.0))
+294
View File
@@ -0,0 +1,294 @@
"""UX-54: the Forecast tab on the two clock-change days.
`python3 test_ux54_dst.py`
The tab walks local hours (`dayplan._hour_slices()` in `dayview` and
`daycurve`, `SolarToday`'s hour keys). On the spring change a loop that steps
on the wall clock never ends (SAFETY-123), and a tab request runs on the
event loop control runs on. Graded here, every call under a hard time limit
that the add-on's own `except Exception` cannot swallow (`Hang` is a
`BaseException`), and a memory limit:
1. The limit itself: a tab build that never returns is stopped and reported,
through `forecast_payload()`'s `except Exception`.
2. `Controller.forecast_payload()` at every quarter of 27-29 March 2027 and
30 October-1 November 2027, modes on (Leftover solar and Cheap hours,
with a forecast and a house profile: the level curve) and modes off, in
the no-tzdata zone (`tariff._EUTimezone`) and, where the host has tzdata,
in ZoneInfo Europe/Brussels: it returns, `ok`, each day has 23 / 24 / 25
hours with the local hour labels of that day, every level inside floor
and ceiling, every block inside its day.
3. `SolarToday` keeps the two 02:00 hours of the autumn change apart.
"""
import logging
import os
import resource
import signal
import sys
from datetime import datetime, timedelta, timezone
HERE = os.path.dirname(os.path.abspath(__file__))
sys.path.insert(0, HERE)
sys.path.insert(0, os.path.join(HERE, "tools"))
logging.disable(logging.WARNING)
import ux54_scenarios as S # noqa: E402
import app.main as M # noqa: E402
from app import arbitrage, dayplan, dayview, homeview, solarfc # noqa: E402
from app import tariff # noqa: E402
from app.daycurve import SolarToday # noqa: E402
UTC = timezone.utc
fails = []
LIMIT_S = 10.0 # one call; a build takes well under a second
try: # a loop that appends without end must not take the host
resource.setrlimit(resource.RLIMIT_AS, (3 << 29, resource.getrlimit(
resource.RLIMIT_AS)[1]))
except (ValueError, OSError):
pass
def check(name, cond):
print(f" {'ok ' if cond else 'FAIL'} {name}")
if not cond:
fails.append(name)
class Hang(BaseException):
"""A call ran over LIMIT_S. Not an `Exception`: `forecast_payload()`
catches every `Exception` and would turn it into `{"ok": False}`."""
def _alarm(*_a):
raise Hang()
def limited(fn, *a, **k):
"""`fn(*a, **k)`, or `Hang` when it takes more than LIMIT_S seconds or
runs out of memory."""
signal.signal(signal.SIGALRM, _alarm)
signal.setitimer(signal.ITIMER_REAL, LIMIT_S)
try:
return fn(*a, **k)
except MemoryError:
raise Hang() from None
finally:
signal.setitimer(signal.ITIMER_REAL, 0)
# --------------------------------------------------------------------------- #
# The clock and the zone
# --------------------------------------------------------------------------- #
NOW = [datetime(2027, 1, 1, tzinfo=UTC)]
class FakeDT(datetime):
@classmethod
def now(cls, tz=None):
return NOW[0].astimezone(tz) if tz else NOW[0].replace(tzinfo=None)
for _mod in (M, homeview, dayview): # what reads the wall clock
_mod.datetime = FakeDT
TZ_MODULES = (tariff, dayplan, homeview, solarfc)
class zone:
"""`BILLING_TZ` is `tz` in every module that holds it, in the block."""
def __init__(self, tz):
self.tz = tz
def __enter__(self):
self.was = [m.BILLING_TZ for m in TZ_MODULES]
for m in TZ_MODULES:
m.BILLING_TZ = self.tz
def __exit__(self, *_a):
for m, tz in zip(TZ_MODULES, self.was):
m.BILLING_TZ = tz
ZONES = [("no tzdata (_EUTimezone)", tariff._EUTimezone())]
try:
from zoneinfo import ZoneInfo
_bru = ZoneInfo("Europe/Brussels")
if datetime(2026, 7, 1, tzinfo=UTC).astimezone(_bru).hour == 2:
ZONES.append(("ZoneInfo Europe/Brussels", _bru))
except Exception: # noqa: BLE001
pass
if len(ZONES) == 1:
print(" no tzdata on this host: ZoneInfo is not graded here")
# The local days (UTC midnight-to-midnight instants are not used: the days
# are known by their dates) and their hour labels.
SPRING, AUTUMN = (2027, 3, 28), (2027, 10, 31)
WINDOWS = (((2027, 3, 27), SPRING, (2027, 3, 29)),
((2027, 10, 30), AUTUMN, (2027, 11, 1)))
def labels_of(ymd):
if ymd == SPRING:
return ["00:00", "01:00"] + [f"{h:02d}:00" for h in range(3, 24)]
if ymd == AUTUMN:
return ["00:00", "01:00", "02:00", "02:00"] + [
f"{h:02d}:00" for h in range(3, 24)]
return [f"{h:02d}:00" for h in range(24)]
def next_ymd(ymd):
d = datetime(*ymd) + timedelta(days=1)
return (d.year, d.month, d.day)
def day_start_utc(ymd):
"""Local midnight of `ymd` in UTC (CET before the change, CEST after it;
midnight is never inside a change)."""
summer = (2027, 3, 28) < ymd < (2027, 10, 31) or ymd == (2027, 10, 31)
if ymd == SPRING:
summer = False
return datetime(*ymd, tzinfo=UTC) - timedelta(hours=2 if summer else 1)
# --------------------------------------------------------------------------- #
print("1. the time limit cannot be swallowed")
# --------------------------------------------------------------------------- #
check("`Hang` is a BaseException and not an Exception",
issubclass(Hang, BaseException) and not issubclass(Hang, Exception))
c0 = limited(S.make, "off",
datetime(2027, 1, 12, 11, 0, 5, tzinfo=UTC))
_real_build = dayview.build
def _spin(*_a, **_k):
while True:
pass
dayview.build = _spin
try:
NOW[0] = datetime(2027, 1, 12, 11, 0, 5, tzinfo=UTC)
got = limited(c0.forecast_payload, "en")
except Hang:
got = "stopped"
finally:
dayview.build = _real_build
check("a tab build that never returns is stopped by the limit and reported, "
f"not turned into a page answer by forecast_payload() ({got!r})",
got == "stopped")
# --------------------------------------------------------------------------- #
print("\n2. the tab at every quarter of the change days")
# --------------------------------------------------------------------------- #
Q = timedelta(minutes=15)
SCENARIOS = (("modes on", "dynamic", (55.0, 40.0)),
("modes off", "off", (55.0, 40.0)))
def wrong(out, ymd, floor_, ceil_):
"""What is wrong with one build, or None."""
if not out.get("ok") or len(out.get("days") or []) != 2:
return "not ok"
for k, day in enumerate(out["days"]):
exp = labels_of((ymd, next_ymd(ymd))[k])
hrs = day["hours"]
if day["length_h"] != len(exp):
return f"day {k}: length {day['length_h']}"
if [h["label"] for h in hrs] != exp:
return f"day {k}: labels"
if any(abs(h["end_h"] - h["start_h"] - 1.0) > 1e-6
or abs(h["start_h"] - j) > 1e-6 for j, h in enumerate(hrs)):
return f"day {k}: hour spans"
for b in day["batteries"]:
if any(not floor_ - 1e-6 <= v <= ceil_ + 1e-6
or not -1e-6 <= h <= len(exp) + 1e-6
for h, v in b["levels"]):
return f"day {k}: level {b['name']}"
if any(not 0.0 <= r["start_h"] <= r["end_h"] <= len(exp) + 1e-6
for r in b["blocks"]):
return f"day {k}: block {b['name']}"
return None
def walk(scn, soc):
"""`(builds, with a curve, stopped, wrong)` over every quarter of the
six local days; the first call stopped by the limit ends the walk."""
builds, curves, bad = 0, 0, []
for window in WINDOWS:
for ymd in window:
a = day_start_utc(ymd)
z = day_start_utc(next_ymd(ymd))
t = a + timedelta(seconds=5)
try:
NOW[0] = t
c = limited(S.make, scn, t, soc=soc)
floor_, ceil_ = arbitrage.soc_bounds(c.o)
while t < z:
NOW[0] = t
c.tariff_at = t
if c.modes_on:
limited(c._mode_roles, t, None)
limited(c.solar_today.add, t, 1500.0)
out = limited(c.forecast_payload, "en")
builds += 1
curves += any(len(b["levels"]) > 1
for d in out.get("days") or []
for b in d["batteries"])
why = wrong(out, ymd, floor_, ceil_)
if why:
bad.append((t.isoformat(), why))
t += Q
except Hang:
return builds, curves, t.isoformat(), bad
return builds, curves, None, bad
for zname, tz in ZONES:
with zone(tz):
for label, scn, soc in SCENARIOS:
builds, curves, stopped, bad = walk(scn, soc)
print(f" {zname}, {label}: {builds} builds ({curves} with a "
f"level curve), stopped at {stopped}, {len(bad)} wrong "
f"{bad[:3]}")
check(f"{zname}, {label}: the tab returns at every quarter of "
"27-29 March and 30 October-1 November 2027 (under "
f"{LIMIT_S:g} s each), and each day has its 23, 24 or 25 "
"hours, labelled, levels inside floor and ceiling, blocks "
"inside the day",
stopped is None and not bad
and builds == (96 + 92 + 96) + (96 + 100 + 96)
and (curves == builds if scn == "dynamic" else not curves))
# --------------------------------------------------------------------------- #
print("\n3. measured solar keeps the two autumn 02:00 hours apart")
# --------------------------------------------------------------------------- #
for zname, tz in ZONES:
with zone(tz):
s = SolarToday()
a = day_start_utc(AUTUMN) # 00:00 CEST = 22:00 UTC
t = a
while t <= a + timedelta(hours=4):
# 1 kW in the first 02:00 hour (00:00-01:00 UTC), 3 kW in the
# second (01:00-02:00 UTC), nothing else
w = 1000.0 if a + timedelta(hours=2) <= t < a + timedelta(
hours=3) else 3000.0 if a + timedelta(hours=3) <= t < a + \
timedelta(hours=4) else 0.0
limited(s.add, t, w)
t += timedelta(minutes=1)
now = a + timedelta(hours=4, minutes=30)
sl = [x for x in limited(dayplan._hour_slices, a, now)
if x[2] == 2]
got = [None if x is None else round(x, 2) for x in
(limited(s.kwh, x[0], x[1], now) for x in sl)]
print(f" {zname}: the two 02:00 hours {got}")
check(f"{zname}: two 02:00 slices, about 1 kWh and 3 kWh, each in "
"its own hour", len(sl) == 2 and got[0] is not None
and got[1] is not None and abs(got[0] - 1.0) < 0.06
and abs(got[1] - 3.0) < 0.06)
print()
if fails:
print(f"{len(fails)} FAILED")
sys.exit(1)
print("all passed")