glenn schrooyenandClaude
50b567d6d1
Forecast.Solar's sensors stop updating in the evening, so at night every such site had no forecast: the night cheap run needed 1.5x the cycle cost and Leftover solar planned nothing. One missed poll also dropped it at once. solarfc.Hold keeps the last fresh reading of each sensor, for the day it is for. A missing or stale sensor is covered by it for up to FRESH_S (a short outage), and from the end of the bell (20:00) until local midnight + 3 h however old. A tomorrow figure is tomorrow's until midnight and then today's; a today figure is only today's. Stored in the controller's store (solar_fc_hold), so an evening restart keeps it. The forecast line says 'Solar forecast read 17:05, used for tonight's plan.' (EN, NL). test_tariff24_weather.py runs with a hold that never covers: it tests what the plan does when the forecast IS lost. Co-Authored-By: Claude <noreply@anthropic.com>
519 lines
24 KiB
Python
519 lines
24 KiB
Python
"""Runnable check for TARIFF-24 round 2 (AUDIT-02 F1). `python3 test_tariff24_weather.py`
|
|
|
|
With battery modes on, the price gap a mode needs was always widened x1.5
|
|
(`Uncertainty.NO_WEATHER`) even with a solar forecast read. Now the doubt is
|
|
added only when no usable forecast is read (`arbitrage.weather_uncertainty()`),
|
|
in the run plan (`_price_runs()`), the charge gate (`_mode_roles()`) and Home's
|
|
mirror (`homeview.trade_verdict()`). Modes off is untouched.
|
|
DECISIONS/TARIFF-24-modes-need-the-tariff-task.md.
|
|
|
|
1. The rule itself, and that modes off keeps the doubt.
|
|
2. Day/night 0.34/0.26 (gap 0.08), modes on, real Controller, night period:
|
|
with a forecast at cycle cost 0.07 the cheap run is planned and the Cheap
|
|
hours battery charges; without one, or with a stale one, the same cost needs
|
|
0.105 and the reason is `not_worth` / "Prices differ too little".
|
|
3. A dynamic day with a 0.12 gap at cost 0.10: on with a forecast, not worth
|
|
without.
|
|
4. The forecast coming (the gate follows in the same cycle) and going mid-run:
|
|
a run under way keeps its gap to its end, the next run needs the wider one;
|
|
a forecast unavailable for one or two reads mid-run does not stop or
|
|
restart the charge.
|
|
5. Home agrees with control in each case (Batteries Why, Home line, and
|
|
`trade_verdict()`, Home's mirror of the gate).
|
|
6. The hold's edges (TARIFF-24 review R4, R5): a forecast lost while only the
|
|
day's give run is under way holds to that run's end and no further, so the
|
|
night's cheap run needs the wider gap; and a give run that starts during a
|
|
hold runs to its end (`keep_started()`), with no grid charge after the hold.
|
|
|
|
Mutations (MUTATION.md, purged copies, graded over this file): the doubt always
|
|
added (the old code) fails parts 2 to 5; never added fails the without-forecast
|
|
rows; Home's mirror left on the old rule fails part 5; `_wx_track()` also
|
|
recording without a forecast (its `solar_fc is None` return dropped) fails part 6.
|
|
"""
|
|
|
|
import asyncio
|
|
import logging
|
|
import math
|
|
import os
|
|
import sys
|
|
import tempfile
|
|
import time
|
|
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 opt11p3_replay as rp # noqa: E402
|
|
from named_absence import first, nth, present # noqa: E402
|
|
import app.main as m # noqa: E402
|
|
from app import arbitrage, homeview, modetrade, solarfc # noqa: E402
|
|
from app.main import Controller # noqa: E402
|
|
from app.p1 import make_sample, split_signed # noqa: E402
|
|
from app.price import (PricePoint, PriceProduct, PriceRead, # noqa: E402
|
|
Uncertainty, brussels)
|
|
from app.static_tariff import TariffPlan # noqa: E402
|
|
from app.store import unit_key # noqa: E402
|
|
|
|
fails = []
|
|
|
|
|
|
class _NoHold(solarfc.Hold):
|
|
"""OPT-27: a hold that never stands in for a lost sensor."""
|
|
|
|
def stand_in(self, *a, **k):
|
|
return None
|
|
|
|
|
|
def check(name, cond):
|
|
print(f" {'ok ' if cond else 'FAIL'} {name}")
|
|
if not cond:
|
|
fails.append(name)
|
|
|
|
|
|
A, B = "inverter-a", "inverter-b"
|
|
PROFILE = ('{"periods": ['
|
|
'{"name": "night", "days": "all", "from": "22:00", "to": "07:00",'
|
|
' "eur_per_kwh": 0.26},'
|
|
'{"name": "day", "days": "all", "from": "07:00", "to": "22:00",'
|
|
' "eur_per_kwh": 0.34}]}')
|
|
L = timezone(timedelta(hours=1))
|
|
NIGHT = datetime(2026, 10, 30, 23, 0, tzinfo=L).astimezone(timezone.utc)
|
|
CHOICE = {"a": (True, "leftover_solar", False),
|
|
"b": (True, "cheap_hours", False)}
|
|
FC_TODAY, FC_TOMORROW = "sensor.fc_today", "sensor.fc_tomorrow"
|
|
TARIFF_REFRESH_S = m.TARIFF_REFRESH_S
|
|
|
|
|
|
def opts(cost, *, arb=False, sensors=True):
|
|
o = dict(rp.options(True, arb, 2))
|
|
o.update(cycle_cost_threshold_eur=cost, static_profile=PROFILE)
|
|
if sensors:
|
|
o.update(pv_forecast_entity=FC_TODAY,
|
|
pv_forecast_tomorrow_entity=FC_TOMORROW)
|
|
return o
|
|
|
|
|
|
def dynamic_points(start):
|
|
"""A dynamic day-ahead shape, 0.16 .. 0.28 (gap 0.12), 15-minute points."""
|
|
out, t = [], start.replace(minute=0, second=0, microsecond=0)
|
|
for _ in range(48 * 4):
|
|
h = brussels(t).hour + brussels(t).minute / 60
|
|
p = 0.22 + 0.06 * math.cos((h - 19) / 24 * 2 * math.pi)
|
|
out.append(PricePoint(start=brussels(t), eur_per_kwh=round(p, 4),
|
|
provisional=False))
|
|
t += timedelta(minutes=15)
|
|
return out
|
|
|
|
|
|
def sensor(kwh, at):
|
|
return {"state": f"{kwh}", "attributes": {"unit_of_measurement": "kWh"},
|
|
"last_updated": at.isoformat(), "last_reported": at.isoformat()}
|
|
|
|
|
|
def forecast_mode(mode, t, now):
|
|
"""What the two forecast sensors read: 'fresh', 'stale', 'gone', or a
|
|
function of the simulated seconds that returns one of those."""
|
|
if callable(mode):
|
|
mode = mode(t)
|
|
if mode == "gone":
|
|
return {"state": "unavailable", "attributes": {}}
|
|
at = now if mode == "fresh" else now - timedelta(hours=5)
|
|
return at
|
|
|
|
|
|
def take(c, now):
|
|
"""What Home and the Batteries screen say now, and Home's mirror of the
|
|
gate (it reads the trade switch, so ask it as a site that has it on next
|
|
to the modes)."""
|
|
keep = c.o
|
|
c.o = dict(keep, arbitrage_enabled=True)
|
|
try:
|
|
mirror, why = homeview.trade_verdict(c, c.fresh_tariff_plan(now), now)
|
|
finally:
|
|
c.o = keep
|
|
return dict(home=homeview.build_plan(c, now, False, True),
|
|
cards=homeview.battery_rows(c, False, True),
|
|
mirror=mirror, mirror_why=why)
|
|
|
|
|
|
def drive(o, *, fc="fresh", points=None, minutes=20, tick=5.0,
|
|
house_w=300.0, soc0=None, snap_at=(), t0=NIGHT):
|
|
"""-> (controller, rows). `points`: a feed-shaped plan injected every
|
|
cycle (no task); else the static profile via the real tariff task. `t0`:
|
|
the first cycle's instant."""
|
|
store = rp._Store()
|
|
for u, (use, mode, sell) in CHOICE.items():
|
|
for k, val in (("pack_use", use), ("pack_mode", mode),
|
|
("pack_sell", sell)):
|
|
store.data[unit_key(f"inverter-{u}", k)] = val
|
|
rows = []
|
|
snaps = {}
|
|
|
|
class H(rp.Hass):
|
|
async def state(self, e):
|
|
now = t0 + timedelta(seconds=self.t)
|
|
if fc is None:
|
|
return None
|
|
at = forecast_mode(fc, self.t, now)
|
|
if isinstance(at, dict):
|
|
return at
|
|
return sensor(0.0 if e == FC_TODAY else 12.0, at)
|
|
|
|
async def go():
|
|
hass = H(("a", "b"), lambda t: house_w, soc0 or {"a": 40.0, "b": 40.0})
|
|
tmp = tempfile.mkdtemp(prefix="t24w-")
|
|
c = Controller(dict(o), hass, store, rp._Pub(), stop=asyncio.Event(),
|
|
peak_db_path=os.path.join(tmp, "tariff.db"),
|
|
price_history_path=os.path.join(tmp, "ph.json"))
|
|
for i in c.inverters:
|
|
i.dev_min, i.dev_max = -5000.0, 5000.0
|
|
c.auto = True
|
|
# OPT-27: a sensor that is missing is covered for up to 3 h by its last
|
|
# fresh reading. This file is about what the plan does when the
|
|
# forecast IS lost, so it runs with a hold that never covers
|
|
# (test_opt27.py tests the cover).
|
|
c.fc_hold = _NoHold()
|
|
real_dt, real_hv = m.datetime, homeview.datetime
|
|
clock = [t0]
|
|
|
|
class _DT(real_dt):
|
|
@classmethod
|
|
def now(cls, tz=None):
|
|
return clock[0] if tz is None else clock[0].astimezone(tz)
|
|
|
|
m.datetime = homeview.datetime = _DT
|
|
pts = None if points is None else points
|
|
task = None
|
|
if pts is None:
|
|
task = asyncio.create_task(c.run_tariff_refresh())
|
|
try:
|
|
t = 0.0
|
|
while t < minutes * 60:
|
|
now = t0 + timedelta(seconds=t)
|
|
hass.t = t
|
|
clock[0] = now
|
|
if pts is not None:
|
|
c.tariff_plan = TariffPlan(
|
|
PriceRead.series(pts, PriceProduct.DAY_AHEAD,
|
|
"generic_dayahead", now), None,
|
|
"provider")
|
|
c.tariff_at = now
|
|
elif t == 0.0 or t % TARIFF_REFRESH_S == 0.0:
|
|
# The task's own 30 min wait is wall clock, so the
|
|
# simulated clock restarts it: one refresh per period.
|
|
if task is not None and t > 0.0:
|
|
task.cancel()
|
|
try:
|
|
await task
|
|
except asyncio.CancelledError:
|
|
pass
|
|
task = asyncio.create_task(c.run_tariff_refresh())
|
|
for _ in range(500):
|
|
if c.tariff_plan is not None and (
|
|
c.tariff_at is not None and c.tariff_at >= now):
|
|
break
|
|
await asyncio.sleep(0)
|
|
imp, exp = split_signed(hass.grid())
|
|
c.p1.submit(make_sample("ha_signed", imp, exp, phases=1,
|
|
ingest_ts=now,
|
|
ingest_mono=time.monotonic()))
|
|
await c.read_inputs(now_mono=t)
|
|
await c.cycle()
|
|
hass.advance(tick, 10.0)
|
|
rp_ = c.mode_runs
|
|
rows.append(dict(
|
|
t=t, now=now, fc_none=c.solar_fc is None,
|
|
runs_on=None if rp_ is None else rp_.on,
|
|
reason=None if rp_ is None else rp_.reason,
|
|
roles=dict(c.mode_tick.roles) if c.mode_tick else {},
|
|
free_to=c._wx_free_to, held=c.weather_held(now),
|
|
why=None if c.mode_tick is None else c.mode_tick.why,
|
|
charging=hass.plants["b"].p < -200.0,
|
|
soc_b=hass.plants["b"].soc,
|
|
cheap_end=None if rp_ is None or rp_.cheap is None
|
|
else rp_.cheap.end))
|
|
if t in snap_at:
|
|
snaps[t] = take(c, now)
|
|
t += tick
|
|
last = take(c, clock[0])
|
|
c.home_plan, c.cards = last["home"], last["cards"]
|
|
c.mirror, c.mirror_why = last["mirror"], last["mirror_why"]
|
|
c.snaps = snaps
|
|
finally:
|
|
if task is not None:
|
|
task.cancel()
|
|
try:
|
|
await task
|
|
except asyncio.CancelledError:
|
|
pass
|
|
m.datetime, homeview.datetime = real_dt, real_hv
|
|
return c
|
|
|
|
return asyncio.run(go()), rows
|
|
|
|
|
|
def why_of(c, name):
|
|
card = first((b for b in c.cards if b["name"] == name), f"card {name}")
|
|
return card["why_key"]
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
print("1. the rule")
|
|
NW = frozenset({Uncertainty.NO_WEATHER})
|
|
check("modes on with a forecast read: no doubt", arbitrage.weather_uncertainty(
|
|
True, object()) == frozenset())
|
|
check("modes on with no forecast (None): the doubt", arbitrage.weather_uncertainty(
|
|
True, None) == NW)
|
|
check("modes off, forecast or not: the doubt, as before",
|
|
arbitrage.weather_uncertainty(False, object()) == NW
|
|
and arbitrage.weather_uncertainty(False, None) == NW)
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
print("\n2. day/night 0.34 / 0.26 (gap 0.08), night period")
|
|
c1, r1 = drive(opts(0.07), fc="fresh")
|
|
l1 = nth(r1, -1, "last cycle")
|
|
print(f" forecast read, cost 0.07: mode_runs on={l1['runs_on']} "
|
|
f"reason={l1['reason']}; B charging {sum(r['charging'] for r in r1)} of "
|
|
f"{len(r1)}; Why A {why_of(c1, A)} B {why_of(c1, B)}")
|
|
check("with a forecast at cost 0.07 the cheap run is planned (needed 0.07 < "
|
|
"0.08) and the Cheap hours battery charges in the night period",
|
|
present(l1) and not l1["fc_none"] and l1["runs_on"] is True
|
|
and sum(r["charging"] for r in r1) > len(r1) // 2
|
|
and l1["roles"].get(B) == modetrade.CHARGE)
|
|
for label, mode in (("no forecast sensor reads", "gone"),
|
|
("a stale forecast", "stale")):
|
|
c, rr = drive(opts(0.07), fc=mode)
|
|
ll = nth(rr, -1, "last cycle")
|
|
print(f" {label}, cost 0.07: on={ll['runs_on']} reason={ll['reason']}; "
|
|
f"B charging {sum(r['charging'] for r in rr)}; Why A "
|
|
f"{why_of(c, A)} B {why_of(c, B)}")
|
|
check(f"{label}: the same cost needs 0.105 > 0.08, so not_worth and "
|
|
f"'Prices differ too little', no grid charge",
|
|
present(ll) and ll["fc_none"] and ll["runs_on"] is False
|
|
and ll["reason"] == "not_worth"
|
|
and why_of(c, A) == "no_spread" and why_of(c, B) == "no_spread"
|
|
and not any(r["charging"] for r in rr))
|
|
c0, r0 = drive(opts(0.07, sensors=False), fc=None)
|
|
check("no forecast sensor configured at all: not_worth too (as 0.19.0)",
|
|
present(nth(r0, -1)) and r0[-1]["runs_on"] is False)
|
|
c2, r2 = drive(opts(0.10), fc="fresh")
|
|
check("at the shipped cost 0.10 even a forecast does not make 0.08 enough "
|
|
"(needs 0.10): still not_worth, honestly",
|
|
present(nth(r2, -1)) and r2[-1]["runs_on"] is False
|
|
and r2[-1]["reason"] == "not_worth")
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
print("\n3. a dynamic day, gap 0.12, cost 0.10")
|
|
pts = dynamic_points(NIGHT - timedelta(hours=1))
|
|
cd, rd = drive(opts(0.10), fc="fresh", points=pts)
|
|
cn, rn = drive(opts(0.10), fc="gone", points=pts)
|
|
_ld, _ln = nth(rd, -1, "last cycle"), nth(rn, -1, "last cycle")
|
|
print(f" with forecast: on={_ld['runs_on']} reason={_ld['reason']}; "
|
|
f"without: on={_ln['runs_on']} reason={_ln['reason']}")
|
|
check("with a forecast the run plan is on (needs 0.10 < 0.12)",
|
|
present(nth(rd, -1)) and rd[-1]["runs_on"] is True
|
|
and why_of(cd, A) != "no_spread")
|
|
check("without one it is not_worth (needs 0.15 > 0.12) and says so",
|
|
present(nth(rn, -1)) and rn[-1]["runs_on"] is False
|
|
and rn[-1]["reason"] == "not_worth"
|
|
and why_of(cn, B) == "no_spread")
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
print("\n4. the forecast going and coming mid-run (a run under way keeps its gap)")
|
|
CUT = 600.0 # 10 min in
|
|
SOC = {"a": 40.0, "b": 60.0} # B's cheap run is then about 2 h
|
|
|
|
|
|
def going(t):
|
|
return "fresh" if t < CUT else "gone"
|
|
|
|
|
|
def coming(t):
|
|
return "gone" if t < CUT else "fresh"
|
|
|
|
|
|
def flap(n_reads):
|
|
"""Unavailable for `n_reads` forecast reads (60 s apart) from CUT."""
|
|
return lambda t: "gone" if CUT <= t < CUT + 60.0 * n_reads else "fresh"
|
|
|
|
|
|
MID, AFTER = 3000.0, 7800.0
|
|
cg, rg = drive(opts(0.07), fc=going, minutes=135, soc0=SOC,
|
|
snap_at=(MID, AFTER))
|
|
cc, rc = drive(opts(0.07), fc=coming, minutes=25)
|
|
_end = nth([r["cheap_end"] for r in rg if r["cheap_end"] is not None], 0,
|
|
"the cheap run's end")
|
|
_end_t = (_end - NIGHT).total_seconds() if present(_end) else 1e9
|
|
lost = [r for r in rg if r["fc_none"]]
|
|
print(f" going: cheap run ends at t={_end_t:.0f} s; forecast lost from "
|
|
f"t={lost[0]['t'] if lost else None}")
|
|
check("going: the run planned with the forecast ends before the snapshot "
|
|
"taken after it (the scenario is long enough)",
|
|
present(_end) and CUT + 300.0 < _end_t < AFTER - 300.0 and bool(lost)
|
|
and lost[0]["t"] - CUT <= 60.0)
|
|
check("going: the forecast read is lost mid-run and the run under way KEEPS "
|
|
"its gap: the plan stays on and B keeps charging, cycle after cycle, "
|
|
"until B is full or the run ends",
|
|
all(r["runs_on"] is True and (r["roles"].get(B) == modetrade.CHARGE
|
|
or r["soc_b"] >= 94.0)
|
|
for r in rg if r["fc_none"] and r["t"] < _end_t - 60.0)
|
|
and all(r["charging"] or r["soc_b"] >= 94.0 for r in rg
|
|
if 60.0 <= r["t"] < _end_t - 60.0)
|
|
and max(r["soc_b"] for r in rg if r["fc_none"]) >= 90.0)
|
|
check("going: from the run's end the wider gap is back: the next run is "
|
|
"not_worth (needs 0.105 at 0.07), no new run, and no grid charge",
|
|
all(r["runs_on"] is False and r["reason"] == "not_worth"
|
|
for r in rg if r["t"] >= _end_t + 30.0)
|
|
and not any(r["charging"] for r in rg if r["t"] >= _end_t + 60.0))
|
|
_sm, _sa = cg.snaps.get(MID), cg.snaps.get(AFTER)
|
|
check("going, Home agrees mid-run: its mirror still allows, the Batteries "
|
|
"screen says it charges from the grid, and Home does not say the gap "
|
|
"is too small",
|
|
_sm is not None and _sm["mirror"] is not None
|
|
and first((b for b in _sm["cards"] if b["name"] == B),
|
|
"a card")["why_key"] == "grid_charge"
|
|
and not any("price gap is too small" in ln
|
|
for ln in _sm["home"]["lines"]))
|
|
check("going, Home agrees after the run: its mirror refuses, the Batteries "
|
|
"screen says 'differ too little' and Home says the gap is too small",
|
|
_sa is not None and _sa["mirror"] is None
|
|
and first((b for b in _sa["cards"] if b["name"] == B),
|
|
"a card")["why_key"] == "no_spread"
|
|
and any("price gap is too small" in ln for ln in _sa["home"]["lines"]))
|
|
for n in (1, 2):
|
|
cf, rf = drive(opts(0.07), fc=flap(n), minutes=30, soc0=SOC,
|
|
snap_at=(CUT + 30.0,))
|
|
gone = [r for r in rf if r["fc_none"]]
|
|
print(f" flap, {n} read(s) lost: {len(gone)} cycles without a "
|
|
f"forecast; B charging in {sum(r['charging'] for r in rf)} of "
|
|
f"{len(rf)}")
|
|
check(f"flap: the forecast is unavailable for {n} read(s) mid-run: the "
|
|
f"plan stays on and B charges without a break (no stop, no "
|
|
f"restart)",
|
|
bool(gone) and all(r["runs_on"] is True for r in rf)
|
|
and all(r["charging"] and r["roles"].get(B) == modetrade.CHARGE
|
|
for r in rf if r["t"] >= 60.0))
|
|
sf = cf.snaps.get(CUT + 30.0)
|
|
check(f"flap, {n} read(s): Home agrees while the read is lost (mirror "
|
|
f"allows, Batteries 'charges from the grid')",
|
|
sf is not None and sf["mirror"] is not None
|
|
and first((b for b in sf["cards"] if b["name"] == B),
|
|
"a card")["why_key"] == "grid_charge")
|
|
|
|
|
|
def flips(rows):
|
|
"""Cycles where the forecast read changed, and whether the plan and the
|
|
charge gate changed in that very cycle."""
|
|
out = []
|
|
for i in range(1, len(rows)):
|
|
if rows[i]["fc_none"] != rows[i - 1]["fc_none"]:
|
|
out.append((i, rows[i]["t"], rows[i]["runs_on"],
|
|
rows[i]["charging"], rows[i - 1]["charging"]))
|
|
return out
|
|
|
|
|
|
fcm = flips(rc)
|
|
print(f" coming: forecast read arrives at cycle/t {fcm}")
|
|
check("coming (no run under way before): not_worth until the read is back, "
|
|
"then on in the same cycle, and the battery charges",
|
|
len(fcm) == 1 and fcm[0][1] - CUT <= 60.0 and fcm[0][2] is True
|
|
and all(r["runs_on"] is False for r in rc if r["fc_none"])
|
|
and all(r["runs_on"] is True for r in rc if not r["fc_none"])
|
|
and any(r["charging"] for r in rc if r["t"] >= fcm[0][1] + 40.0))
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
print("\n5. Home agrees with control")
|
|
CASES = (("forecast, cost 0.07", c1, r1, True),
|
|
("no forecast, cost 0.07", *drive(opts(0.07), fc="gone"), False),
|
|
("dynamic, forecast", cd, rd, True),
|
|
("dynamic, no forecast", cn, rn, False))
|
|
for label, cx, rx, on in CASES:
|
|
last = nth(rx, -1, "last cycle")
|
|
hp = cx.home_plan
|
|
mirror_ok = cx.mirror is not None
|
|
print(f" {label}: control on={last['runs_on']}; Home mirror "
|
|
f"{'allows' if mirror_ok else 'refuses'} ({cx.mirror_why}); "
|
|
f"lines {hp['lines'][:1]}; Why B {why_of(cx, B)}")
|
|
check(f"{label}: Home's mirror of the gate allows exactly when control "
|
|
f"plans", present(last) and mirror_ok is on
|
|
and last["runs_on"] is on)
|
|
check(f"{label}: the Batteries screen and the Home line say the gap is "
|
|
f"too small exactly when control does",
|
|
(why_of(cx, B) == "no_spread") is (not on)
|
|
and any("price gap is too small" in ln for ln in hp["lines"])
|
|
is (not on))
|
|
if label.startswith("dynamic"):
|
|
continue # its cheap run is tomorrow morning: not on today's chart
|
|
check(f"{label}: Home draws a grid charge block exactly when control "
|
|
f"charges", any(r["kind"] == "grid" for r in hp["runs"]) is on)
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
print("\n6. the hold's edges (review probes R4, R5)")
|
|
T_R4 = datetime(2026, 10, 30, 21, 40, tzinfo=L).astimezone(timezone.utc)
|
|
TEN = datetime(2026, 10, 30, 22, 0, tzinfo=L).astimezone(timezone.utc)
|
|
c4, r4 = drive(opts(0.07), fc=lambda t: "fresh" if t < 300.0 else "gone",
|
|
minutes=60, t0=T_R4)
|
|
_held4 = [r for r in r4 if r["held"]]
|
|
_l4 = nth(r4, -1, "last cycle")
|
|
print(f" R4: forecast lost 21:45; held {len(_held4)} cycles, to "
|
|
f"{_l4['free_to']}; after 22:00 on={ {r['runs_on'] for r in r4 if r['now'] >= TEN} } "
|
|
f"charging {sum(r['charging'] for r in r4)}")
|
|
check("R4, forecast lost before the next run: while only the day's give run "
|
|
"was under way the hold is that run's end, 22:00, and stays there",
|
|
present(_l4) and all(r["free_to"] == TEN for r in r4
|
|
if r["t"] >= 5.0)
|
|
and bool(_held4) and all(r["now"] < TEN for r in _held4))
|
|
check("R4: from 22:00 the night's cheap run needs the wider gap: "
|
|
"not_worth, and no grid charge at any time",
|
|
all(r["runs_on"] is False and r["reason"] == "not_worth"
|
|
for r in r4 if r["now"] >= TEN)
|
|
and not any(r["charging"] for r in r4))
|
|
|
|
|
|
def r5_points(t0):
|
|
"""04:00-07:00 at 0.20, 07:00-10:00 at 0.30, else 0.25 (local)."""
|
|
out, t = [], t0.replace(minute=0, second=0, microsecond=0) - timedelta(hours=1)
|
|
for _ in range(4 * 30):
|
|
h = brussels(t).hour
|
|
p = 0.20 if 4 <= h < 7 else 0.30 if 7 <= h < 10 else 0.25
|
|
out.append(PricePoint(start=brussels(t), eur_per_kwh=p,
|
|
provisional=False))
|
|
t += timedelta(minutes=15)
|
|
return out
|
|
|
|
|
|
T_R5 = datetime(2026, 10, 30, 5, 0, tzinfo=L).astimezone(timezone.utc)
|
|
GIVE_END = datetime(2026, 10, 30, 10, 0, tzinfo=L).astimezone(timezone.utc)
|
|
c5, r5 = drive(opts(0.07), fc=lambda t: "fresh" if t < 3600.0 else "gone",
|
|
points=r5_points(T_R5), minutes=320, t0=T_R5,
|
|
soc0={"a": 40.0, "b": 30.0})
|
|
_ft5 = nth([r["free_to"] for r in r5 if r["t"] == 3600.0], 0, "the hold")
|
|
_in = [r for r in r5 if present(_ft5) and _ft5 <= r["now"] < GIVE_END]
|
|
_out = [r for r in r5 if r["now"] >= GIVE_END + timedelta(seconds=30)]
|
|
print(f" R5: forecast lost 06:00, hold to {_ft5}; after it, until 10:00, "
|
|
f"on in {sum(r['runs_on'] is True for r in _in)} of {len(_in)}, B "
|
|
f"{sorted({str(r['roles'].get(B)) for r in _in})}; after 10:00 "
|
|
f"{ {r['reason'] for r in _out} }")
|
|
check("R5: the hold is the cheap run's end, before the give run's end",
|
|
present(_ft5) and T_R5 < _ft5 < GIVE_END)
|
|
check("R5, a run that has started runs to its end: the give run that began "
|
|
"during the hold stays on after it, B gives, and nothing charges from "
|
|
"the grid",
|
|
len(_in) > 100 and all(r["runs_on"] is True
|
|
and r["roles"].get(B) == modetrade.GIVE
|
|
and not r["charging"] for r in _in))
|
|
check("R5: when the give run ends the wider gap decides: not_worth, no grid "
|
|
"charge",
|
|
bool(_out) and all(r["runs_on"] is False and r["reason"] == "not_worth"
|
|
and not r["charging"] for r in _out))
|
|
|
|
print()
|
|
if fails:
|
|
print(f"{len(fails)} FAILED:")
|
|
for f in fails:
|
|
print(f" - {f}")
|
|
sys.exit(1)
|
|
print("all checks passed")
|