glenn schrooyenandClaude
eb9ccc2730
Review F1: a slot after B's run counts only if no slot from the expensive
hours on is cheaper (control plans B's charge left in the cheapest window
ahead), and x 0.95 of B's charge is stored. Three-price night: 1ac7fe9 left
the expensive hours 2.61 kWh short; now 0.
Review F2: a Solar first battery's drain until the expensive hours reads each
hour at no less than the night base (the profile is net of the sun). Read net,
a dark morning drained the battery the plan still counted; A's level fell,
the cheap run moved and was lost before it started (+0.40 EUR per 3 days).
The two real-price replays now cost what 0.20.0 does.
F3: the ux54 sweep's peak limit is 15 kW (the option's top). F4: the rest
comes from the grid (EN/NL). F5: CHANGELOG and DECISIONS wording.
Co-Authored-By: Claude <noreply@anthropic.com>
920 lines
45 KiB
Python
920 lines
45 KiB
Python
"""Runnable check for OPT-36. `python3 test_opt36.py`
|
|
|
|
Leftover solar counts what Cheap hours only batteries will give in the
|
|
expensive hours before it buys from the grid at night (hotfix 0.20.1).
|
|
|
|
Live 0.20.0, 2026-11-02 01:01: A (Leftover solar, 60 %) went to 95 %
|
|
(5.38 kWh) while B (Cheap hours only, 94 %) already covered the expensive
|
|
hours. DECISIONS/OPT-23-28 had left "netting what a Cheap hours battery will
|
|
hold" out; this adds it.
|
|
|
|
1. The live figures through the real Controller: A buys 0, nothing is
|
|
"uncovered", and Home says the same in EN and NL.
|
|
2. `dayplan.plan()` alone: B low with a short cheap run under the peak limit,
|
|
A buys only what B cannot reach; the numbers written out. Round 3: B's
|
|
charge counts the cheap slots left before the expensive hours, not only
|
|
the run (sized at the ratings), so the level does not fall in the night.
|
|
3. A Cheap hours battery that is off, refusing, in maintenance, not live or
|
|
without capacity counts 0; Solar first counts after the night's use
|
|
(owner, 2026-11-02); N=1 and Leftover solar only are unchanged.
|
|
4. The screens: the give line reads the level after the planned charge,
|
|
`plan_cover()` is the plan's own figure, the Why line says who covers.
|
|
5. The real Controller through a whole night (parallel processes, 5 s per
|
|
cycle): A's commands follow the new level, which holds all night while
|
|
the figure does; OPT-34 F-1 with B counted (a raised level is reached and
|
|
held); the OPT-34 latch still holds a started run (a lowered level stays
|
|
lowered). Round 4: a three-price night holds the shortfall at 07:00, and
|
|
two real-price replays with a Solar first battery cost no more than
|
|
0.20.0.
|
|
|
|
Mutations (MUTATION.md, purged copies): listed in the pull request.
|
|
"""
|
|
|
|
import asyncio
|
|
import json
|
|
import logging
|
|
import os
|
|
import sys
|
|
import tempfile
|
|
import time
|
|
from concurrent.futures import ProcessPoolExecutor
|
|
from datetime import datetime, timedelta, timezone
|
|
from multiprocessing import get_context
|
|
|
|
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)
|
|
|
|
from named_absence import nth, present # noqa: E402
|
|
import ux54_scenarios as S # noqa: E402
|
|
import opt11p3_replay as rp # noqa: E402
|
|
import app.main as m # noqa: E402
|
|
from app import dayplan, homeview, modetrade, packmode # noqa: E402
|
|
from app import priceruns, solarfc, uitext # noqa: E402
|
|
from app.main import Controller # noqa: E402
|
|
from app.p1 import make_sample, split_signed # noqa: E402
|
|
from app.peakguard import PeakDecision # noqa: E402
|
|
from app.static_tariff import static_plan # noqa: E402
|
|
from app.store import unit_key # noqa: E402
|
|
from app.tariff import BILLING_TZ # noqa: E402
|
|
|
|
fails = []
|
|
|
|
|
|
def check(name, cond):
|
|
print(f" {'ok ' if cond else 'FAIL'} {name}")
|
|
if not cond:
|
|
fails.append(name)
|
|
|
|
|
|
UTC = timezone.utc
|
|
EFF = dayplan.EFF
|
|
A, B = "inverter-a", "inverter-b"
|
|
FLOOR, CEIL = 20.0, 95.0
|
|
CAP_A, CAP_B = 15.36, 20.48
|
|
|
|
|
|
# ============================================================================ #
|
|
print("1. the live figures through the real Controller")
|
|
# ============================================================================ #
|
|
NOW = datetime(2026, 11, 2, 1, 1, 26, tzinfo=BILLING_TZ).astimezone(UTC)
|
|
|
|
|
|
def shape_of(scale=1.0):
|
|
"""The brief's owner example (0.6 kW at night, 1.2 kW 06-10, 0.9 kW by
|
|
day, 1.8 kW 17-22), scaled."""
|
|
return {h: scale * (600.0 if h < 6 or h >= 22 else 1200.0 if h < 10
|
|
else 900.0 if h < 17 else 1800.0) for h in range(24)}
|
|
|
|
|
|
def profile_of(shape, now=NOW):
|
|
d = now.astimezone(BILLING_TZ).date()
|
|
return dayplan.HouseProfile(
|
|
{(d - timedelta(days=k)).isoformat():
|
|
{str(h): w for h, w in shape.items()} for k in (1, 2, 3)})
|
|
|
|
|
|
def make(shape, soc=(60.0, 94.0), modes=("leftover_solar", "cheap_hours"),
|
|
today=10.2, tomorrow=10.0, now=NOW, peak_cap=None):
|
|
"""A Controller at `now`: day/night tariff, floor 20 %, ceiling 95 %,
|
|
three state-only sensors (10.2 kWh today), its mode roles run once."""
|
|
opts = dict(S.BASE)
|
|
opts.update(battery_modes_enabled=True,
|
|
pv_forecast_entities=[f"{S.FC1}_{i}" for i in range(3)],
|
|
pv_forecast_tomorrow_entities=[f"{S.FCT1}_{i}"
|
|
for i in range(3)],
|
|
static_profile=json.dumps(S.DN), arbitrage_enabled=False,
|
|
arbitrage_soc_floor=20, arbitrage_soc_ceiling=95,
|
|
cycle_cost_threshold_eur=0.04)
|
|
if peak_cap is not None:
|
|
opts.update(peak_cap_w=peak_cap)
|
|
plan = static_plan(opts, S.day0_of(now), S.day0_of(now) + timedelta(
|
|
hours=48), now=now)
|
|
c = Controller(opts, S.Hass(), S.Store(), S.Pub(), stop=S.asyncio.Event())
|
|
c.auto = True
|
|
for i, inv in enumerate(c.inverters):
|
|
inv.soc, inv.batt, inv.last_write = soc[i], 0.0, 0.0
|
|
inv.dev_min, inv.dev_max = -5000.0, 5000.0
|
|
c.store.data[unit_key(inv.name, packmode.MODE_KEY)] = modes[i]
|
|
c.grid, c.pv_w = 3000.0, 0.0
|
|
c.tariff_plan, c.tariff_at = plan, now
|
|
c.peak_guard = PeakDecision(engaged=False, floor_w=None, ceiling_w=7000.0,
|
|
ceiling_from="cap", projected_w=800.0)
|
|
c.arb_headroom_w = 3450.0 - 3000.0 + 2000.0
|
|
c.solar_fc = S.forecast(now, today, tomorrow, members=3)
|
|
c.solar_fc_why = None
|
|
c.house_profile = profile_of(shape, now)
|
|
c.tariff_at = now
|
|
c._mode_roles(now, None)
|
|
return c
|
|
|
|
|
|
def screen(c, lang="en", now=NOW):
|
|
st = homeview.build(c, level="ok", checks=[], now=now, lang=lang)
|
|
rows = {r["name"]: r for r in st["batteries"]}
|
|
return st, rows
|
|
|
|
|
|
LIVE = shape_of(0.973) # the scale whose plan matches live (11.3 kWh)
|
|
c = make(LIVE)
|
|
dp = c.day_plan
|
|
tg = c.plan_targets()
|
|
print(f" short {dp.short_kwh:.2f} kWh, Cheap hours give {dp.cheap_kwh:.2f}, "
|
|
f"A to {tg[A]} %, A buys {dp.need_kwh[A]:.2f} kWh, uncovered "
|
|
f"{dp.uncovered_kwh:.2f}")
|
|
check("live: A (Leftover solar 60 %) buys nothing while B (Cheap hours 94 %) "
|
|
"covers the expensive hours (was: to 95 %, 5.38 kWh)",
|
|
dp.need_kwh[A] == 0.0 and tg[A] <= 60.0
|
|
and homeview.plan_need(c, A) == 0.0)
|
|
check("live: nothing is left uncovered (was 0.4 kWh), and B's give is "
|
|
"counted up to the shortfall and no more",
|
|
dp.uncovered_kwh == 0.0 and abs(dp.cheap_kwh - dp.short_kwh) < 1e-6)
|
|
check("live: control gives A no grid charge at all (the run charges B only)",
|
|
[p.name for p in modetrade.chargers(
|
|
c._mode_packs(None), CEIL, targets=c.plan_targets())] == [B])
|
|
for lang in ("en", "nl"):
|
|
st, rows = screen(c, lang)
|
|
mine = [r for r in st["plan"]["runs"]
|
|
if r["battery"] == 0 and r["kind"] != "house"]
|
|
theirs = [r for r in st["plan"]["runs"]
|
|
if r["battery"] == 1 and r["kind"] != "house"]
|
|
line = st["solar"]["line"]
|
|
check(f"{lang}: A has no grid block, only its give block "
|
|
f"({[r['kind'] for r in mine]})",
|
|
[r["kind"] for r in mine] == ["give"])
|
|
check(f"{lang}: A's give block is what A holds (60 % -> 6.1 kWh)",
|
|
abs(nth(mine, 0)["kwh"] - (60.0 - FLOOR) / 100.0 * CAP_A) < 0.01)
|
|
check(f"{lang}: B's give block reads B after its charge (to 95 %: "
|
|
"15.4 kWh, not 15.2 from 94 %)",
|
|
abs(nth(theirs, -1)["kwh"] - (CEIL - FLOOR) / 100.0 * CAP_B) < 0.01)
|
|
check(f"{lang}: the solar line has no 'the rest' sentence",
|
|
"0.4" not in line and "0,4" not in line
|
|
and "rest, " not in line and "rest van" not in line
|
|
and "rest," not in line)
|
|
check(f"{lang}: no line anywhere says Leftover solar charges from the "
|
|
"grid to 95",
|
|
all("95" not in r["why"] for r in rows.values())
|
|
and all("95 %" not in t["what"] for r in rows.values()
|
|
for t in r["today"] if "grid" in t["kind"]
|
|
and r is rows[A]))
|
|
check("live: while B charges, A's Why says another battery charges (it "
|
|
"rests)", screen(c, "en")[1][A]["why_key"] == "rests_charge")
|
|
|
|
# B at 100 % of its run done: the cheap run is over, B only gives. A's Why
|
|
# says who covers the hours, not the sun.
|
|
c2 = make(LIVE, soc=(60.0, 95.0))
|
|
c2._mode_roles(NOW, None)
|
|
role_a = c2.mode_tick.roles[A]
|
|
key_a = homeview.day_key(c2, A)
|
|
print(f" B full: A's role {role_a}, day key {key_a}")
|
|
check("B already at the ceiling: A has no charge, and the day key says the "
|
|
"Cheap hours battery covers, not the sun",
|
|
key_a == "others_cover" and c2.day_plan.need_kwh[A] == 0.0
|
|
and c2.day_plan.saved_kwh[A] > 5.0)
|
|
for lang in ("en", "nl"):
|
|
with homeview.in_lang(lang):
|
|
txt = homeview.why_words("others_cover")
|
|
sentences = [x for x in txt.split(".") if x.strip()]
|
|
check(f"{lang}: the 'others_cover' line is at most 2 sentences of at "
|
|
f"most 12 words and has no raw placeholder: {txt!r}",
|
|
len(sentences) <= 2 and all(len(x.split()) <= 12
|
|
for x in sentences) and "{" not in txt)
|
|
check("EN and NL say that other batteries cover, in plain words",
|
|
"Other batteries" in uitext.words("en")["home_words"]["why"][
|
|
"others_cover"]
|
|
and "Andere batterijen" in uitext.words("nl")["home_words"]["why"][
|
|
"others_cover"])
|
|
|
|
# Owner, 2026-11-02: a Solar first battery counts too, at its level when the
|
|
# expensive hours start: level now less the house's night use to 07:00
|
|
# (01:01 to 07:00: 0.58 kW to 06:00, then 1.17 kW), shared among Solar first
|
|
# batteries, x 0.95. A Leftover solar at its floor buys only what is left.
|
|
c3 = make(LIVE, soc=(20.0, 94.0), modes=("leftover_solar", "solar_first"))
|
|
# Round 4: each hour at no less than the night base (the profile is net of
|
|
# the sun the house used).
|
|
_hw, base_n = c3.house_profile.estimator()
|
|
use_n = sum(max(LIVE[h_], base_n) / 1000.0 * (e - s_).total_seconds() / 3600.0
|
|
for s_, e, h_ in dayplan._hour_slices(
|
|
NOW, datetime(2026, 11, 2, 7, 0, tzinfo=BILLING_TZ)))
|
|
have_b = (94.0 - FLOOR) / 100.0 * CAP_B
|
|
give_sf = (have_b - use_n / EFF) * EFF
|
|
d3 = c3.day_plan
|
|
want_hold = (d3.short_kwh - give_sf) / EFF
|
|
print(f" B Solar first 94 %: night use {use_n:.2f}, B gives {give_sf:.2f} "
|
|
f"of {d3.short_kwh:.2f}; A (20 %) buys {d3.need_kwh[A]:.3f}")
|
|
check("B on Solar first at 94 % counts: its level at 07:00 is 94 % less the "
|
|
"night's use, (level - floor) x cap x 0.95",
|
|
abs(d3.solar_first_kwh - give_sf) < 0.01 and d3.cheap_kwh == 0.0)
|
|
check("...and A (at its floor) buys only the shortfall after that "
|
|
"(was 5.38 kWh and to 95 %)",
|
|
abs(d3.need_kwh[A] - want_hold) < 0.02 and d3.need_kwh[A] < 1.5
|
|
and d3.uncovered_kwh == 0.0)
|
|
c3b = make(LIVE, soc=(60.0, 94.0), modes=("leftover_solar", "solar_first"))
|
|
check("...and A at 60 % with B Solar first 94 % buys nothing",
|
|
c3b.day_plan.need_kwh[A] == 0.0 and c3b.day_plan.uncovered_kwh == 0.0)
|
|
|
|
# ============================================================================ #
|
|
print("2. dayplan.plan() alone: B low, a short cheap run under the peak limit")
|
|
# ============================================================================ #
|
|
|
|
|
|
def loc(day, hour, minute=0):
|
|
return datetime(2026, 10, day, hour, minute,
|
|
tzinfo=BILLING_TZ).astimezone(UTC)
|
|
|
|
|
|
SHAPE = shape_of(1.0)
|
|
T = loc(30, 4, 0) # 04:00, three cheap hours left
|
|
RUNS = priceruns.RunPlan(
|
|
priceruns.Run(loc(29, 22), loc(30, 7), 0.2333),
|
|
(priceruns.Run(loc(30, 7), loc(30, 22), 0.2801),
|
|
priceruns.Run(loc(31, 7), loc(31, 22), 0.2801)),
|
|
None, 0.2801, None, True)
|
|
|
|
|
|
def pk(name, mode, soc, cap, rating=4600.0, **kw):
|
|
return modetrade.Pack(name, True, mode, False, rating, soc, cap, **kw)
|
|
|
|
|
|
def fc_of(kwh, now=T):
|
|
d0, _d1 = solarfc._bounds(now)
|
|
return solarfc.Forecast(solarfc.Day(d0, kwh),
|
|
solarfc.Day(d0 + timedelta(days=1), kwh), now)
|
|
|
|
|
|
def mk_plan(packs, *, aim=3450.0, runs=RUNS, now=T, kwh=6.0, shape=SHAPE):
|
|
return dayplan.plan(fc_of(kwh, now), profile_of(shape, now).estimator(),
|
|
runs, now, packs, trust_pct=70, floor_pct=FLOOR,
|
|
ceiling_pct=CEIL, aim_w=aim)
|
|
|
|
|
|
def expect_short(kwh, shape=SHAPE, trust=0.7):
|
|
"""The shortfall written out: per hour 07..21 max(0, house - 0.7 x sun),
|
|
an hour with sun read at no less than the night base."""
|
|
base = sorted(shape[h] for h in (0, 1, 2, 3, 4, 5, 22, 23))
|
|
base = (base[3] + base[4]) / 2.0
|
|
tot = 0.0
|
|
for h in range(7, 22):
|
|
pv = kwh * (solarfc._bell(h + 1.0) - solarfc._bell(float(h)))
|
|
use = (max(shape[h], base) if pv > 0 else shape[h]) / 1000.0
|
|
tot += max(0.0, use - trust * pv)
|
|
return tot
|
|
|
|
|
|
def expect_b_level(soc, cap, gain_kwh):
|
|
"""`gain_kwh` drawn from the grid; x 0.95 of it is stored (round 4)."""
|
|
return max(soc, min(CEIL, soc + gain_kwh * EFF / cap * 100.0))
|
|
|
|
|
|
SH = expect_short(6.0)
|
|
# 04:00-07:00 at the night 0.6 kW (04, 05) and 1.2 kW (06): the peak aim 3450
|
|
# leaves 2850, 2850 and 2250 W of the battery's 4600 W.
|
|
GAIN = 2.85 + 2.85 + 2.25
|
|
LB = expect_b_level(30.0, CAP_B, GAIN)
|
|
GIVE_B = (LB - FLOOR) / 100.0 * CAP_B * EFF
|
|
A_LVL = FLOOR + (SH - GIVE_B) / EFF / CAP_A * 100.0
|
|
p_low = mk_plan([pk(A, packmode.LEFTOVER_SOLAR, FLOOR, CAP_A),
|
|
pk(B, packmode.CHEAP_HOURS, 30.0, CAP_B)])
|
|
print(f" short {SH:.2f}; B 30 % + {GAIN:.2f} kWh -> {LB:.1f} % gives "
|
|
f"{GIVE_B:.2f}; A to {A_LVL:.1f} % (plan: {p_low.targets[A]} %, B "
|
|
f"gives {p_low.cheap_kwh})")
|
|
check("the shortfall is the written-out sum", abs(p_low.short_kwh - SH) < 0.005)
|
|
check("B's give is (its level at the start of the dear hours - floor) x "
|
|
"capacity x 0.95, the level being what 3 h under the peak aim reach",
|
|
abs(p_low.cheap_kwh - GIVE_B) < 0.005 and LB < CEIL - 20.0)
|
|
check("A buys only what B cannot reach: its level is floor + (shortfall - "
|
|
"B's give) / 0.95",
|
|
abs(p_low.targets[A] - A_LVL) < 0.05
|
|
and abs(p_low.need_kwh[A] - (A_LVL - FLOOR) / 100.0 * CAP_A) < 0.01
|
|
and p_low.uncovered_kwh == 0.0)
|
|
alone = mk_plan([pk(A, packmode.LEFTOVER_SOLAR, FLOOR, CAP_A)])
|
|
check("...which is less than A would buy with no Cheap hours battery",
|
|
alone.need_kwh[A] > p_low.need_kwh[A] + 4.0 and alone.cheap_kwh == 0.0)
|
|
check("the grid kWh A saves is reported (`saved_kwh`)",
|
|
abs(p_low.saved_kwh[A] - (alone.need_kwh[A] - p_low.need_kwh[A])) < 0.01)
|
|
|
|
# No peak limit given: the batteries' ratings alone (4.6 kW x 3 h reaches 95).
|
|
p_free = mk_plan([pk(A, packmode.LEFTOVER_SOLAR, FLOOR, CAP_A),
|
|
pk(B, packmode.CHEAP_HOURS, 30.0, CAP_B)], aim=None)
|
|
GIVE_FREE = (expect_b_level(30.0, CAP_B, 4.6 * 3.0) - FLOOR) / 100.0 \
|
|
* CAP_B * EFF
|
|
check("with no peak aim the run is held by the rating only: 4.6 kW x 3 h "
|
|
"x 0.95 stored takes B to 94 %; A buys only what is left",
|
|
abs(p_free.cheap_kwh - GIVE_FREE) < 0.005
|
|
and abs(p_free.need_kwh[A] - (SH - GIVE_FREE) / EFF) < 0.01
|
|
and p_free.uncovered_kwh == 0.0)
|
|
# A house at the aim leaves no room: B cannot charge, counts what it holds.
|
|
p_full = mk_plan([pk(A, packmode.LEFTOVER_SOLAR, FLOOR, CAP_A),
|
|
pk(B, packmode.CHEAP_HOURS, 30.0, CAP_B)], aim=500.0)
|
|
check("a house at the peak aim leaves no room: B counts its level now "
|
|
"((30 - 20) % x 20.48 x 0.95)",
|
|
abs(p_full.cheap_kwh - 10.0 / 100.0 * CAP_B * EFF) < 0.005)
|
|
# The run is not worth charging in: B stays where it is.
|
|
NOCH = priceruns.RunPlan(RUNS.cheap, RUNS.give, None, RUNS.level_eur_kwh,
|
|
None, False)
|
|
p_noch = mk_plan([pk(A, packmode.LEFTOVER_SOLAR, FLOOR, CAP_A),
|
|
pk(B, packmode.CHEAP_HOURS, 30.0, CAP_B)], runs=NOCH,
|
|
aim=None)
|
|
check("a cheap run not worth charging in: B counts its level now",
|
|
abs(p_noch.cheap_kwh - 10.0 / 100.0 * CAP_B * EFF) < 0.005)
|
|
# A cheap run that is already over (now past its end): the level now.
|
|
p_over = mk_plan([pk(A, packmode.LEFTOVER_SOLAR, FLOOR, CAP_A),
|
|
pk(B, packmode.CHEAP_HOURS, 80.0, CAP_B)],
|
|
now=loc(30, 7, 0) - timedelta(seconds=1), aim=None)
|
|
check("the last second of the run: B counts its level (80 %), no more",
|
|
abs(p_over.cheap_kwh - (80.0 - FLOOR) / 100.0 * CAP_B * EFF) < 0.01)
|
|
# Two Cheap hours batteries share the peak room by rating.
|
|
p_two = mk_plan([pk(A, packmode.LEFTOVER_SOLAR, FLOOR, CAP_A),
|
|
pk(B, packmode.CHEAP_HOURS, 30.0, CAP_B, rating=3000.0),
|
|
pk("inverter-c", packmode.CHEAP_HOURS, 30.0, CAP_B,
|
|
rating=1500.0)])
|
|
g_b = GAIN * 3000.0 / 4500.0
|
|
g_c = GAIN * 1500.0 / 4500.0
|
|
give_two = sum((expect_b_level(30.0, CAP_B, g) - FLOOR) / 100.0 * CAP_B * EFF
|
|
for g in (g_b, g_c))
|
|
# 2.85 kW is under the 4.5 kW of the two together: shared 2:1. Hours 04-05
|
|
# take 2.85 of 4.5, hour 06 takes 2.25 of 4.5.
|
|
check("two Cheap hours batteries share the peak room by rating",
|
|
abs(p_two.cheap_kwh - min(give_two, p_two.short_kwh)) < 0.01)
|
|
|
|
# Round 3: the run is sized at the ratings, not the peak limit, so a run too
|
|
# short to fill B is planned again later in the night. B's charge counts the
|
|
# firm price slots after the run, up to the expensive hours, that a run would
|
|
# still be worth charging in: price + the spread control asks (`spread`, 0.04
|
|
# here) at most the give run's dearest slot (0.2801). Without them A's level
|
|
# fell through the night after A had bought.
|
|
RUN1 = priceruns.RunPlan(
|
|
priceruns.Run(loc(30, 4), loc(30, 5), 0.2333), RUNS.give, None,
|
|
RUNS.level_eur_kwh, None, True)
|
|
|
|
|
|
def quarters(t0, t1, eur):
|
|
out, t = [], t0
|
|
while t < t1:
|
|
out.append((t, t + timedelta(minutes=15), eur))
|
|
t += timedelta(minutes=15)
|
|
return out
|
|
|
|
|
|
NIGHT_Q = quarters(loc(30, 4), loc(30, 7), 0.2333)
|
|
DAY_Q = quarters(loc(30, 7), loc(30, 22), 0.2801)
|
|
LATE_Q = quarters(loc(30, 22), loc(31, 7), 0.2333)
|
|
PK2 = [pk(A, packmode.LEFTOVER_SOLAR, FLOOR, CAP_A),
|
|
pk(B, packmode.CHEAP_HOURS, 30.0, CAP_B)]
|
|
|
|
|
|
def mk_slots(slots, runs=RUN1, spread=0.04):
|
|
return dayplan.plan(fc_of(6.0, T), profile_of(SHAPE, T).estimator(),
|
|
runs, T, PK2, trust_pct=70, floor_pct=FLOOR,
|
|
ceiling_pct=CEIL, aim_w=3450.0, slots=slots,
|
|
spread=spread)
|
|
|
|
|
|
p_run1 = mk_slots(None)
|
|
p_q = mk_slots(NIGHT_Q + DAY_Q + LATE_Q)
|
|
GIVE_1H = (expect_b_level(30.0, CAP_B, 2.85) - FLOOR) / 100.0 * CAP_B * EFF
|
|
print(f" a 1 h run (04:00-05:00): B gives {p_run1.cheap_kwh:.3f} without "
|
|
f"the slots, {p_q.cheap_kwh:.3f} with the night's slots to 07:00")
|
|
check("a 1 h run with no slots given: B counts what 1 h reaches (as before)",
|
|
abs(p_run1.cheap_kwh - GIVE_1H) < 0.005)
|
|
check("...and with the slots but no spread given, the same",
|
|
abs(mk_slots(NIGHT_Q + DAY_Q, spread=None).cheap_kwh - GIVE_1H)
|
|
< 0.005)
|
|
check("the same run with the night's quarters (0.2333) to 07:00: B "
|
|
"counts what 04:00-07:00 reach, as the 3 h run does (and A buys the "
|
|
"same); the day and the next night are not counted",
|
|
abs(p_q.cheap_kwh - GIVE_B) < 0.005
|
|
and abs(p_q.targets[A] - p_low.targets[A]) < 0.01)
|
|
p_dear = mk_slots(quarters(loc(30, 5), loc(30, 7), 0.2500) + DAY_Q)
|
|
check("slots after the run not worth charging in (0.25 + 0.04 over the "
|
|
"dearest 0.2801) are not counted: B counts the 1 h run only",
|
|
abs(p_dear.cheap_kwh - GIVE_1H) < 0.005)
|
|
p_part = mk_slots(quarters(loc(30, 5), loc(30, 6), 0.2400)
|
|
+ quarters(loc(30, 6), loc(30, 7), 0.2500) + DAY_Q)
|
|
GIVE_2H = (expect_b_level(30.0, CAP_B, 2.85 + 2.85) - FLOOR) / 100.0 \
|
|
* CAP_B * EFF
|
|
check("...and only those are: 05:00-06:00 at 0.24 (dearer than the run, "
|
|
"still worth it) adds one hour (04:00-06:00 reach)",
|
|
abs(p_part.cheap_kwh - GIVE_2H) < 0.005)
|
|
|
|
# ============================================================================ #
|
|
print("3. who counts, and who does not")
|
|
# ============================================================================ #
|
|
base_pk = [pk(A, packmode.LEFTOVER_SOLAR, FLOOR, CAP_A)]
|
|
want = mk_plan(base_pk)
|
|
for label, b in (
|
|
("off", modetrade.Pack(B, False, packmode.CHEAP_HOURS, False, 4600.0,
|
|
94.0, CAP_B)),
|
|
("refusing to give", pk(B, packmode.CHEAP_HOURS, 94.0, CAP_B,
|
|
refused=1)),
|
|
("refusing to charge", pk(B, packmode.CHEAP_HOURS, 94.0, CAP_B,
|
|
refused=-1)),
|
|
("in maintenance", pk(B, packmode.CHEAP_HOURS, 94.0, CAP_B,
|
|
idle=False)),
|
|
("not live", pk(B, packmode.CHEAP_HOURS, 94.0, CAP_B, live=False)),
|
|
("without capacity", modetrade.Pack(B, True, packmode.CHEAP_HOURS,
|
|
False, 4600.0, 94.0, None)),
|
|
("without a level", modetrade.Pack(B, True, packmode.CHEAP_HOURS,
|
|
False, 4600.0, None, CAP_B)),
|
|
("Solar first refusing to give",
|
|
pk(B, packmode.SOLAR_FIRST, 94.0, CAP_B, refused=1)),
|
|
("Solar first in maintenance",
|
|
pk(B, packmode.SOLAR_FIRST, 94.0, CAP_B, idle=False)),
|
|
("Solar first with the night draining it to the floor",
|
|
pk(B, packmode.SOLAR_FIRST, 24.0, CAP_B))):
|
|
pl = mk_plan(base_pk + [b])
|
|
check(f"B {label}: counts 0, A's plan is as with no B",
|
|
pl.cheap_kwh == 0.0 and pl.targets == want.targets
|
|
and pl.need_kwh == want.need_kwh
|
|
and pl.uncovered_kwh == want.uncovered_kwh
|
|
and pl.cover_kwh == want.cover_kwh)
|
|
on = mk_plan(base_pk + [pk(B, packmode.CHEAP_HOURS, 94.0, CAP_B)])
|
|
check("(control: the same B, on and idle, counts)",
|
|
on.cheap_kwh > 5.0 and on.need_kwh[A] < want.need_kwh[A] - 1.0)
|
|
# Solar first at 94 % at 04:00: three night hours (0.6, 0.6, 1.2 kW) are used
|
|
# before 07:00, shared with nobody else; the rest counts.
|
|
p_sf = mk_plan(base_pk + [pk(B, packmode.SOLAR_FIRST, 94.0, CAP_B)])
|
|
use_t = 0.6 + 0.6 + 1.2
|
|
give_t = ((94.0 - FLOOR) / 100.0 * CAP_B - use_t / EFF) * EFF
|
|
check("Solar first at 94 % at 04:00: counts its level at 07:00 "
|
|
f"({give_t:.2f} kWh), A buys only the rest",
|
|
abs(p_sf.solar_first_kwh - give_t) < 0.005 and p_sf.cheap_kwh == 0.0
|
|
and p_sf.need_kwh[A] < want.need_kwh[A] - 4.0)
|
|
# Two Solar first batteries share the night's use by the energy they hold.
|
|
p_sf2 = mk_plan(base_pk + [pk(B, packmode.SOLAR_FIRST, 60.0, CAP_B),
|
|
pk("inverter-c", packmode.SOLAR_FIRST, 40.0,
|
|
CAP_B)])
|
|
tot_h = (60.0 - FLOOR) / 100.0 * CAP_B + (40.0 - FLOOR) / 100.0 * CAP_B
|
|
check("two Solar first batteries: the night's use is shared by the energy "
|
|
"held; the total is the same sum",
|
|
abs(p_sf2.solar_first_kwh - min(
|
|
p_sf2.short_kwh, (tot_h - use_t / EFF) * EFF)) < 0.005)
|
|
# Both kinds: Cheap hours and Solar first add up (capped at the shortfall).
|
|
p_both = mk_plan(base_pk + [pk(B, packmode.SOLAR_FIRST, 60.0, CAP_B),
|
|
pk("inverter-c", packmode.CHEAP_HOURS, 30.0,
|
|
CAP_B)])
|
|
check("Cheap hours and Solar first batteries both count",
|
|
p_both.cheap_kwh > 5.0 and p_both.solar_first_kwh > 3.0
|
|
and abs(p_both.cheap_kwh + p_both.solar_first_kwh
|
|
- min(p_both.short_kwh, p_both.cheap_kwh
|
|
+ p_both.solar_first_kwh)) < 1e-6)
|
|
# A Cheap hours battery with Leftover solar A: when A is not Leftover solar
|
|
# (N=1 on Cheap hours alone) the plan has no Leftover solar to size.
|
|
solo = mk_plan([pk(B, packmode.CHEAP_HOURS, 94.0, CAP_B)])
|
|
check("Cheap hours only sites: no Leftover solar battery, nothing is sized",
|
|
solo.targets == {} and solo.need_kwh == {} and solo.cheap_kwh == 0.0
|
|
and solo.uncovered_kwh == 0.0)
|
|
# Everything the plan says about a Leftover solar only site is the same as
|
|
# before OPT-36: the new fields are at rest.
|
|
two_ls = mk_plan([pk(A, packmode.LEFTOVER_SOLAR, FLOOR, CAP_A),
|
|
pk(B, packmode.LEFTOVER_SOLAR, 94.0, CAP_B)])
|
|
check("Leftover solar only (two batteries): cheap_kwh 0, nothing saved",
|
|
two_ls.cheap_kwh == 0.0
|
|
and all(v == 0.0 for v in two_ls.saved_kwh.values())
|
|
and abs(sum(two_ls.cover_kwh.values()) - two_ls.short_kwh) < 0.01)
|
|
# More than the Cheap hours battery holds: A buys the rest, "the rest" is said
|
|
# only for what nothing holds.
|
|
big = shape_of(2.0)
|
|
p_big = mk_plan([pk(A, packmode.LEFTOVER_SOLAR, FLOOR, CAP_A),
|
|
pk(B, packmode.CHEAP_HOURS, 94.0, CAP_B)], shape=big,
|
|
aim=None)
|
|
want_big = expect_short(6.0, big)
|
|
cheap_big = (CEIL - FLOOR) / 100.0 * CAP_B * EFF
|
|
check("a shortfall B cannot cover: A holds the rest, to the ceiling; what "
|
|
"neither holds is 'uncovered'",
|
|
abs(p_big.cheap_kwh - cheap_big) < 0.005
|
|
and p_big.targets[A] == CEIL
|
|
and abs(p_big.uncovered_kwh - max(
|
|
0.0, want_big - cheap_big - (CEIL - FLOOR) / 100.0 * CAP_A * EFF))
|
|
< 0.01)
|
|
check("the plan is pure: the same call twice gives the same plan",
|
|
mk_plan(base_pk + [pk(B, packmode.CHEAP_HOURS, 30.0, CAP_B)])
|
|
== mk_plan(base_pk + [pk(B, packmode.CHEAP_HOURS, 30.0, CAP_B)]))
|
|
|
|
# ============================================================================ #
|
|
print("4. the screens agree with control")
|
|
# ============================================================================ #
|
|
# A tops up (B low, the house large): the give block reads A after its charge.
|
|
cs = make(shape_of(1.5), soc=(20.0, 30.0), now=NOW)
|
|
dps = cs.day_plan
|
|
tgs = cs.plan_targets()
|
|
print(f" A 20 %, B 30 %: B gives {dps.cheap_kwh:.2f}, A to {tgs[A]} %, "
|
|
f"buys {dps.need_kwh[A]:.2f}, uncovered {dps.uncovered_kwh:.2f}")
|
|
check("(A does buy here: B cannot cover the larger house)",
|
|
dps.need_kwh[A] > 1.0 and dps.cheap_kwh > 5.0)
|
|
for lang in ("en", "nl"):
|
|
st, rows = screen(cs, lang)
|
|
runs_a = [r for r in st["plan"]["runs"] if r["battery"] == 0]
|
|
grid = next((r for r in runs_a if r["kind"] == "grid"), None)
|
|
give = next((r for r in runs_a if r["kind"] == "give"), None)
|
|
cov = homeview.plan_cover(cs, A)
|
|
with homeview.in_lang(lang):
|
|
pct, kwh = homeview._pct(tgs[A]), homeview._kwh(dps.cover_kwh[A])
|
|
check(f"{lang}: A tops up: the grid block names the plan's level and "
|
|
"the plan's own cover figure",
|
|
grid is not None and pct in grid["what"] and kwh in grid["why"]
|
|
and cov == (tgs[A], dps.cover_kwh[A]))
|
|
check(f"{lang}: the give block is A at the level it charges to "
|
|
f"({(tgs[A] - FLOOR) / 100.0 * CAP_A:.2f} kWh), not at 20 % (0)",
|
|
give is not None and abs(
|
|
give["kwh"] - (tgs[A] - FLOOR) / 100.0 * CAP_A) < 0.02)
|
|
# The Batteries line for a battery that holds (`tops_up`) is made from the
|
|
# same figures: the plan's level and its own cover figure.
|
|
check("(A's day key is tops_up)", homeview.day_key(cs, A) == "tops_up")
|
|
for lang in ("en", "nl"):
|
|
with homeview.in_lang(lang):
|
|
words = homeview.plan_words(cs, A, "tops_up")
|
|
txt = homeview.why_words("tops_up", **words)
|
|
pct, kwh = homeview._pct(tgs[A]), homeview._kwh(dps.cover_kwh[A])
|
|
check(f"{lang}: A's tops_up line names the plan's level ({pct}) and "
|
|
f"cover figure ({kwh}): {txt!r}", pct in txt and kwh in txt)
|
|
# The controller hands the plan its peak aim (peak_cap_w - peak_margin_w): a
|
|
# lower peak limit leaves B less room in its run, so it covers less.
|
|
lo_cap = make(shape_of(1.5), soc=(20.0, 30.0), peak_cap=2000)
|
|
hi_cap = make(shape_of(1.5), soc=(20.0, 30.0), peak_cap=20000)
|
|
print(f" peak limit 2000 W: B gives {lo_cap.day_plan.cheap_kwh:.2f} kWh; "
|
|
f"20000 W: {hi_cap.day_plan.cheap_kwh:.2f} kWh")
|
|
check("the Controller passes its peak aim: a 2 kW limit leaves B less room "
|
|
"than a 20 kW one, and A buys the difference",
|
|
lo_cap.day_plan.cheap_kwh < hi_cap.day_plan.cheap_kwh - 2.0
|
|
and lo_cap.day_plan.need_kwh[A] > hi_cap.day_plan.need_kwh[A] + 1.0)
|
|
# A hysteresis-held level: the cover figure is still the plan's own.
|
|
real = cs.plan_targets
|
|
cs.plan_targets = lambda: {A: tgs[A] + 0.7}
|
|
cv = homeview.plan_cover(cs, A)
|
|
check("plan_cover() is the plan's cover figure whatever level control holds "
|
|
"(hysteresis), the level being the held one",
|
|
cv == (tgs[A] + 0.7, dps.cover_kwh[A]))
|
|
cs.plan_targets = real
|
|
# A give run before the cheap run (10:00: tomorrow night's charge is after
|
|
# today's give): the give block reads the level now.
|
|
c10 = make(shape_of(2.0), soc=(40.0, 25.0), modes=("leftover_solar", "solar_first"),
|
|
now=datetime(2026, 11, 2, 10, 0, tzinfo=BILLING_TZ).astimezone(UTC))
|
|
st10 = homeview.build(c10, level="ok", checks=[],
|
|
now=datetime(2026, 11, 2, 10, 0,
|
|
tzinfo=BILLING_TZ).astimezone(UTC),
|
|
lang="en")
|
|
g10 = [r for r in st10["plan"]["runs"]
|
|
if r["battery"] == 0 and r["kind"] == "give"]
|
|
check("a give run that comes BEFORE the next cheap run is read at the level "
|
|
"now (40 % -> 3.1 kWh), not at tonight's charge",
|
|
g10 and abs(g10[0]["kwh"] - (40.0 - FLOOR) / 100.0 * CAP_A) < 0.02
|
|
and any(r["kind"] == "grid" and r["battery"] == 0
|
|
for r in st10["plan"]["runs"]))
|
|
|
|
# ============================================================================ #
|
|
print("5. the real Controller through a whole night")
|
|
# ============================================================================ #
|
|
L = timezone(timedelta(hours=1)) # Brussels in winter (CET)
|
|
|
|
|
|
def at(day, hh, mm=0):
|
|
return datetime(2026, 11, day, hh, mm, tzinfo=L).astimezone(UTC)
|
|
|
|
|
|
def st_of(kwh, when, *, unit="kWh"):
|
|
return {"state": f"{kwh}", "attributes": {"unit_of_measurement": unit},
|
|
"last_updated": when.isoformat(),
|
|
"last_reported": when.isoformat()}
|
|
|
|
|
|
GONE = {"state": "unavailable", "attributes": {}}
|
|
T_ENT, M_ENT = "sensor.fc_today", "sensor.fc_tomorrow"
|
|
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}]}')
|
|
CHOICE = {"a": (True, "leftover_solar", False),
|
|
"b": (True, "cheap_hours", False)}
|
|
EVENING = [at(3, 16, 0) + timedelta(minutes=30 * i) for i in range(12)] \
|
|
+ [at(3, 21, 55)]
|
|
|
|
|
|
def night_opts(cap, profile=PROFILE):
|
|
o = dict(rp.options(True, False, 2))
|
|
o.update(cycle_cost_threshold_eur=0.07, static_profile=profile,
|
|
pv_forecast_entity=T_ENT, pv_forecast_tomorrow_entity=M_ENT,
|
|
peak_cap_w=cap)
|
|
return o
|
|
|
|
|
|
def run_night(spec):
|
|
"""One whole night in this process -> a plain dict. `spec`: start (UTC),
|
|
hours, soc, house_w, cap, fresh (today, tomorrow) before `change`,
|
|
change (UTC) and the figures after it, silent_from (UTC): the sensors
|
|
show their last reading from then on."""
|
|
start = spec["start"]
|
|
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
|
|
clock = [EVENING[0]]
|
|
figs = spec.get("fresh", (1.0, 5.0))
|
|
change = spec.get("change")
|
|
after = spec.get("after", figs)
|
|
silent = spec.get("silent_from")
|
|
|
|
def sens(now, ent):
|
|
f = after if change is not None and now >= change else figs
|
|
when = now if silent is None else min(now, silent)
|
|
return st_of(f[0] if ent == T_ENT else f[1], when)
|
|
|
|
class H(rp.Hass):
|
|
async def state(self, e):
|
|
return sens(clock[0], e)
|
|
|
|
hass = H(("a", "b"), lambda t: 300.0, spec["soc"])
|
|
out = dict(marks={}, soc_at={}, p_a=[])
|
|
tmp = tempfile.mkdtemp(prefix="o36-")
|
|
c = Controller(dict(night_opts(spec.get("cap", 2500),
|
|
spec.get("profile", PROFILE))), 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
|
|
hw = spec.get("house_w", 700.0)
|
|
c.house_profile.load({d: {str(hh): hw for hh in range(24)}
|
|
for d in ("2026-10-31", "2026-11-01", "2026-11-02")})
|
|
|
|
async def go():
|
|
real_dt, real_hv = m.datetime, homeview.datetime
|
|
|
|
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
|
|
try:
|
|
for t_pre in EVENING:
|
|
clock[0] = t_pre
|
|
await c._read_forecast()
|
|
marks = sorted(spec.get("marks", ()))
|
|
first, t = True, 0.0
|
|
hz = spec["hours"] * 3600.0
|
|
while t < hz:
|
|
now = start + timedelta(seconds=t)
|
|
clock[0] = now
|
|
if first or int(t) % 600 == 0:
|
|
hor = float(m.price_option(c.o, "price_horizon_h"))
|
|
c.tariff_plan = await m.resolve_tariff(
|
|
c.o, now, now + timedelta(hours=hor), now=now,
|
|
history=c.price_history, watch=c.price_watch)
|
|
c.tariff_at = now
|
|
first = False
|
|
hass.t = t
|
|
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(5.0, 10.0)
|
|
if int(t) % 300 == 0:
|
|
tgt = c.plan_targets()
|
|
out["soc_at"][now.astimezone(L).strftime("%H:%M")] = (
|
|
hass.plants["a"].soc, hass.plants["b"].soc,
|
|
hass.plants["a"].p, c.solar_fc is None,
|
|
None if tgt is None else tgt.get(A))
|
|
if int(t) % 60 == 0:
|
|
out["p_a"].append(hass.plants["a"].p)
|
|
while marks and now >= marks[0]:
|
|
marks.pop(0)
|
|
cards = homeview.battery_rows(c, False, True)
|
|
blk = homeview.solar_block(c, now)
|
|
a_card = next(b for b in cards if b["name"] == A)
|
|
dpl = c.day_plan
|
|
out["marks"][now.astimezone(L).strftime("%H:%M")] = dict(
|
|
line=blk["line"], why_a=a_card["why_key"],
|
|
fc_none=c.solar_fc is None,
|
|
target_a=(c.plan_targets() or {}).get(A),
|
|
need_a=None if dpl is None else dpl.need_kwh.get(A),
|
|
cheap=None if dpl is None else dpl.cheap_kwh,
|
|
short=None if dpl is None else dpl.short_kwh,
|
|
soc_a=hass.plants["a"].soc,
|
|
p_a=hass.plants["a"].p)
|
|
t += 5.0
|
|
out["soc"] = (hass.plants["a"].soc, hass.plants["b"].soc)
|
|
out["latch"] = getattr(c, "_charge_latch", None) is not None
|
|
finally:
|
|
m.datetime, homeview.datetime = real_dt, real_hv
|
|
return out
|
|
|
|
return asyncio.run(go())
|
|
|
|
|
|
# Round 4 (review F1): three night prices, the cheapest in the middle.
|
|
PROFILE3 = ('{"periods": ['
|
|
'{"name": "eve", "days": "all", "from": "22:00", "to": "01:00",'
|
|
' "eur_per_kwh": 0.15},'
|
|
'{"name": "low", "days": "all", "from": "01:00", "to": "03:00",'
|
|
' "eur_per_kwh": 0.10},'
|
|
'{"name": "dawn", "days": "all", "from": "03:00", "to": "07:00",'
|
|
' "eur_per_kwh": 0.15},'
|
|
'{"name": "day", "days": "all", "from": "07:00", "to": "22:00",'
|
|
' "eur_per_kwh": 0.34}]}')
|
|
|
|
|
|
def run_replay(variant, fc):
|
|
"""One 3-day opt11p4 replay (real 2026 prices, 2026-09-02) -> net EUR."""
|
|
import contextlib
|
|
import io
|
|
import opt11p4_replay as p4
|
|
rp.VARIANTS["own_sf"] = (True, False, {
|
|
"a": (True, "leftover_solar", False),
|
|
"b": (True, "solar_first", False)})
|
|
argv = sys.argv
|
|
sys.argv = ["p4", "--variant", variant, "--fc", fc, "--prices",
|
|
"real2026", "--shape", "periods", "--first", "2", "--days",
|
|
"3"]
|
|
buf = io.StringIO()
|
|
try:
|
|
with contextlib.redirect_stdout(buf):
|
|
p4.main()
|
|
finally:
|
|
sys.argv = argv
|
|
return json.loads(buf.getvalue().strip().splitlines()[-1])
|
|
|
|
|
|
NIGHT = at(3, 22, 0)
|
|
HOURLY = [NIGHT + timedelta(minutes=30 * i) for i in range(18)]
|
|
SCEN = {
|
|
# The live night: A 60 % Leftover solar, B 94 % Cheap hours, a fresh
|
|
# figure all night. A must not charge from the grid.
|
|
"live": dict(start=NIGHT, hours=9, soc={"a": 60.0, "b": 94.0},
|
|
house_w=700.0, marks=[at(4, 1, 0), at(4, 5, 0)]),
|
|
# A mid case: both low (25 %), the house 0.8 kW, a 2.5 kW peak limit
|
|
# leaves B 1.65 kW in a 3 h run (the run is sized to the batteries'
|
|
# ratings, not the limit), so B counts for part and A buys the rest.
|
|
"mid": dict(start=NIGHT, hours=9, soc={"a": 25.0, "b": 25.0},
|
|
house_w=800.0, marks=HOURLY),
|
|
# Round 3: the same with one figure for the 4th all night (today 5.0
|
|
# after midnight = tomorrow 5.0 before): A's level must not move at all.
|
|
# On round 2's code it went 55 -> 69 -> 41 -> 37 % and A bought to 56 %.
|
|
"mid_flat": dict(start=NIGHT, hours=9, soc={"a": 25.0, "b": 25.0},
|
|
house_w=800.0, fresh=(5.0, 5.0), marks=HOURLY),
|
|
# Round 3, OPT-34 F-1 with B on Cheap hours and counted: a darker fresh
|
|
# figure at 02:00, then silent (no forecast from 05:00). The level rises,
|
|
# A charges to it, and the latch holds it after 05:00.
|
|
# Round 4 (review F1): the three-price night. B's charge after the run is
|
|
# planned in the next night's 0.10 hours, not in tonight's 0.15 ones, so
|
|
# those may not count. On 1ac7fe9 the plan counted B to 95 %, B ended at
|
|
# 67.5 % and the expensive hours were 2.61 kWh short; 0.20.0: none.
|
|
"three": dict(start=NIGHT, hours=9, soc={"a": 30.0, "b": 20.0},
|
|
house_w=700.0, cap=10000, fresh=(1.0, 1.0),
|
|
profile=PROFILE3, marks=HOURLY),
|
|
"f1_dark_b": dict(start=at(4, 0, 30), hours=6.5, cap=3500,
|
|
soc={"a": 50.0, "b": 50.0}, house_w=900.0,
|
|
fresh=(5.0, 5.0), change=at(4, 2, 0),
|
|
after=(1.0, 1.0), silent_from=at(4, 2, 0),
|
|
marks=[at(4, 1, 55)] + [at(4, 2, 15) + timedelta(
|
|
minutes=15 * i) for i in range(19)]),
|
|
# OPT-34 with the new level: a fresh, brighter figure at 03:10 lowers A's
|
|
# level below its SoC (A stops), then the sensors stay silent: the
|
|
# forecast ends and the latch holds the LOWERED level to 07:00.
|
|
"lowered": dict(start=at(4, 2, 0), hours=5.0, soc={"a": 25.0, "b": 25.0},
|
|
house_w=1200.0, fresh=(1.0, 5.0), change=at(4, 3, 10),
|
|
after=(30.0, 30.0), silent_from=at(4, 3, 10),
|
|
marks=[at(4, 3, 5), at(4, 3, 20), at(4, 6, 30)]),
|
|
}
|
|
t0 = time.time()
|
|
with ProcessPoolExecutor(max_workers=min(4, os.cpu_count() or 2),
|
|
mp_context=get_context("fork")) as pool:
|
|
futs = {k: pool.submit(run_night, v) for k, v in SCEN.items()}
|
|
# Round 4 (review F2): the real-price replays with a Solar first battery.
|
|
reps = {k: pool.submit(run_replay, *v) for k, v in (
|
|
("ownsf_hourly_real", ("own_sf", "perfect")),
|
|
("p4_ls_dark_real", ("p4_ls", "zero")))}
|
|
R = {k: f.result() for k, f in futs.items()}
|
|
REP = {k: f.result() for k, f in reps.items()}
|
|
print(f" {len(R)} nights in {time.time() - t0:.0f} s")
|
|
|
|
lv = R["live"]
|
|
print(f" live: A {lv['soc'][0]:.2f} % B {lv['soc'][1]:.2f} % "
|
|
f"marks {lv['marks']}")
|
|
check("live night: A takes no grid charge: it never charges (its power "
|
|
"stays above -200 W) and ends where it started",
|
|
min(lv["p_a"]) > -200.0 and abs(lv["soc"][0] - 60.0) < 1.5)
|
|
check("live night: B fills to the ceiling as before",
|
|
lv["soc"][1] > 94.0)
|
|
check("live night: the plan gives A a level at or under its SoC and the "
|
|
"screen says nothing about 'the rest'",
|
|
all(mk["need_a"] == 0.0 and mk["cheap"] > 0.0
|
|
and "rest" not in mk["line"] for mk in lv["marks"].values()))
|
|
|
|
md = R["mid"]
|
|
m1, m2 = md["marks"]["22:30"], md["marks"]["05:00"]
|
|
print(f" mid: A level {m1['target_a']} % at 22:30, {m2['target_a']} % at "
|
|
f"05:00; A ends {md['soc'][0]:.2f} %, B {md['soc'][1]:.2f} %; B gives "
|
|
f"{m1['cheap']}")
|
|
check("mid night: the plan counts B's give and A's level is a partial one "
|
|
"(not the ceiling, as 0.20.0 gave: 95 %)",
|
|
m1["target_a"] is not None and 25.0 < m1["target_a"] < CEIL - 15.0
|
|
and m1["cheap"] > 3.0)
|
|
check("mid night: B's counted give is the same at every mark (B reaches the "
|
|
"ceiling in the night left), and A's level moves only at midnight, "
|
|
"when the 4th's figure changes from tomorrow's 5.0 to today's 1.0",
|
|
len({mk["cheap"] for mk in md["marks"].values()}) == 1
|
|
and len({mk["target_a"] for h, mk in md["marks"].items()
|
|
if h >= "12"}) == 1
|
|
and len({mk["target_a"] for h, mk in md["marks"].items()
|
|
if h < "12"}) == 1)
|
|
check("mid night: A's commands follow the levels: it charges to its level "
|
|
"and stops there, well under the ceiling (0.20.0: 95 %)",
|
|
md["soc"][0] < 85.0 and abs(md["soc"][0] - m2["target_a"]) < 0.5)
|
|
check("mid night: B fills as before (to the ceiling)", md["soc"][1] > 94.0)
|
|
mf = R["mid_flat"]
|
|
lv_f = [mk["target_a"] for mk in mf["marks"].values()]
|
|
lv0 = nth(lv_f, 0)
|
|
print(f" mid_flat: A level(s) {sorted(set(map(str, lv_f)))}; A ends "
|
|
f"{mf['soc'][0]:.2f} %")
|
|
check("one figure all night: A's level never moves (round 2: 55 -> 69 -> "
|
|
"41 -> 37 %), A charges to it and stops (round 2: 56 %)",
|
|
present(lv0) and lv0 is not None and len(set(lv_f)) == 1
|
|
and abs(mf["soc"][0] - lv0) < 0.5)
|
|
fd = R["f1_dark_b"]
|
|
dk = fd["marks"]
|
|
held = [dk[h]["target_a"] for h in dk if "05:15" <= h < "07:00"]
|
|
print(f" f1_dark_b: A level 01:55 {dk['01:55']['target_a']} -> 02:15 "
|
|
f"{dk['02:15']['target_a']}, held {sorted(set(held))}; A "
|
|
f"{fd['soc_at']['06:55'][0]:.1f} % at 06:55, B {fd['soc'][1]:.1f} %")
|
|
check("F-1 with B counted: a darker fresh figure at 02:00 raises A's level, "
|
|
"the level does not move after that (B's give is the same), it is held "
|
|
"with no forecast from 05:15, and A charges to it",
|
|
dk["01:55"]["target_a"] is not None
|
|
and dk["02:15"]["target_a"] > dk["01:55"]["target_a"] + 3.0
|
|
and dk["01:55"]["cheap"] > 3.0
|
|
and all(dk[h]["target_a"] == dk["02:15"]["target_a"] for h in dk
|
|
if h >= "02:15")
|
|
and all(dk[h]["fc_none"] for h in dk if "05:15" <= h < "07:00")
|
|
and abs(fd["soc_at"]["06:55"][0] - dk["02:15"]["target_a"]) < 1.5
|
|
and fd["soc"][1] > 94.0)
|
|
|
|
th = R["three"]
|
|
sh3 = [mk["short"] for mk in th["marks"].values() if mk["short"] is not None]
|
|
held3 = sum(max(0.0, x - FLOOR) for x in th["soc"]) / 100.0 \
|
|
* rp.CAP_KWH * EFF
|
|
sh3_0 = nth(sh3, 0)
|
|
print(f" three prices: A {th['soc'][0]:.2f} %, B {th['soc'][1]:.2f} %; "
|
|
f"held at 07:00 {held3:.2f} kWh for a shortfall of {sh3_0} kWh "
|
|
"(1ac7fe9: 7.20 for 9.81, 0.20.0: 13.89)")
|
|
check("three night prices: B's charge is not counted in the 0.15 hours "
|
|
"after its run (the next night's 0.10 hours are cheaper), so A and B "
|
|
"together hold the expensive hours' shortfall at 07:00 (1ac7fe9: "
|
|
"2.61 kWh short, bought at 0.34)",
|
|
present(sh3_0) and held3 >= sh3_0 - 0.05)
|
|
for k, base_eur in (("ownsf_hourly_real", 4.00), ("p4_ls_dark_real", 3.84)):
|
|
r_ = REP[k]
|
|
print(f" replay {k}: {r_['net_eur']} EUR (0.20.0: {base_eur}), end "
|
|
f"{r_['end_soc']}")
|
|
check(f"real 2026 prices, 3 days, a Solar first battery counted ({k}): "
|
|
f"it costs no more than 0.20.0 ({base_eur} EUR; 1ac7fe9: "
|
|
f"{base_eur + 0.40:.2f})", r_["net_eur"] <= base_eur + 0.005)
|
|
|
|
lw = R["lowered"]
|
|
l0, l1, l2 = (lw["marks"][k] for k in ("03:05", "03:20", "06:30"))
|
|
print(f" lowered: A level {l0['target_a']} -> {l1['target_a']} -> "
|
|
f"{l2['target_a']}; SoC {l1['soc_a']:.1f} -> {lw['soc'][0]:.1f} %")
|
|
check("a fresh, brighter figure lowers A's level below its SoC during the "
|
|
"charge, and A stops",
|
|
l0["target_a"] is not None and l1["target_a"] is not None
|
|
and l1["target_a"] < l0["target_a"] - 5.0
|
|
and lw["soc_at"]["03:30"][0] < lw["soc_at"]["03:20"][0] + 0.5)
|
|
check("the sensors go quiet and the forecast ends (06:30): the latch holds "
|
|
"the LOWERED level, A does not charge back up to the old one",
|
|
l2["fc_none"] and l2["target_a"] is not None
|
|
and l2["target_a"] < l0["target_a"] - 5.0 and l2["p_a"] > -200.0
|
|
and lw["soc_at"]["06:55"][0] < lw["soc_at"]["03:30"][0] + 1.0)
|
|
|
|
# ============================================================================ #
|
|
print("\n" + ("all checks passed" if not fails
|
|
else f"FAILED: {len(fails)}"))
|
|
for f in fails:
|
|
print(" -", f)
|
|
sys.exit(1 if fails else 0)
|