507 lines
23 KiB
Python
507 lines
23 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 # 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 = []
|
|
|
|
|
|
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
|
|
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")
|