OPT-36: Leftover solar counts what Cheap hours and Solar first batteries will give before buying
checks / unit (pull_request) Failing after 59m31s
infra-pin / fresh (pull_request) Successful in 2s

Live 0.20.0: A (Leftover solar, 60 %) charged to 95 % while B (Cheap hours,
94 %) already covered the expensive hours. dayplan.plan() now takes off the
shortfall what Cheap hours batteries will hold when the hours start (the
ceiling or what their cheap run reaches under the peak aim) and what Solar
first batteries will still hold after the night's use. Home reads a give
block after the planned charge, plan_cover() is the plan's own figure, and
the Why line says other batteries cover.

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
glenn schrooyen
2026-11-02 02:03:04 +01:00
co-authored by Claude
parent 01e88a24f2
commit 05746c0564
11 changed files with 867 additions and 23 deletions
+4
View File
@@ -2,6 +2,10 @@
## 0.20.0
### Not in a release yet
- Leftover solar no longer charges from the grid for hours that Cheap hours only or Solar first batteries will cover. It counts what they will hold when the expensive hours start (a Solar first battery after the night's use), and buys only the rest. The Batteries screen says which battery covers, and the "gives up to" figure is read after the planned charge.
**A minor bump: a Forecast tab, a night grid charge sized to what the sun will not give, a kept evening forecast, and no option changes.**
No option was added, removed or renamed, and no default changed (86 options). It is a minor bump because Leftover solar now buys different amounts from the grid and the night charge keeps its level when the forecast goes quiet; every saved setting still starts.
**Existing installs need no option change:** options that start 0.19.2 also
@@ -7,5 +7,6 @@
- **Shortfall.** `sum_h max(0, house_h - trust x sun_h)` over every expensive hour, `/ EFF`, from the floor. It replaces the use over the run with no sun counted, which on a day and night tariff bought for the whole day (G1, `reports/UX-69-design-report.md`). The sun before the run is unchanged. A run with its dear hours after sunset gives the old number.
- **House use.** An hour with sun is read at no less than the night base, as the sun sum reads it (the profile is net of the sun). An hour without is read as measured. An hour with no forecast counts no sun.
- **Several batteries.** One total, shared by capacity with the ceiling passed on (`dayplan._spread()`). With the same floor and ceiling for all, they reach the ceiling together, so a shortfall that one battery cannot hold fills both to the same level, not one first. More than all hold: all go to the ceiling, `Plan.uncovered_kwh` is the rest, carried by another battery (Cheap hours, Solar first) or the grid, as the screen says.
- **Cheap hours** still fills up and is not counted in the share; **Solar first** still never charges from the grid. The mode help says so.
- **Not done:** the run length (OPT-25); netting what a Cheap hours or Solar first battery will hold against the shortfall.
- **Cheap hours** still fills up; **Solar first** still never charges from the grid. The mode help says so.
- **OPT-36 (0.20.1):** what Cheap hours batteries will give in the expensive hours is taken off the shortfall before Leftover solar shares it (`dayplan._cheap_give_kwh()`: its level when they start, the ceiling or what its cheap run reaches under the peak aim, minus the floor, x capacity x EFF). Owner, 2026-11-02: Solar first batteries count too (`_solar_first_give_kwh()`): their level now less the house's night use until the hours start, shared by the energy held, then the same sum; a battery the night drains to the floor counts 0. Live, 2026-11-02: A at 60 % went to 95 % while B at 94 % covered the hours. Test: `test_opt36.py`.
- **Not done:** the run length (OPT-25).
+1 -1
View File
@@ -549,7 +549,7 @@ With battery modes on, these choices replace the *Charging and selling on price*
- The hourly shape is kept only if every sensor has one. Otherwise the total is spread over the day as for Forecast.Solar.
- Before each cheap run the add-on asks: will the sun fill the batteries before the expensive hours? It counts on `solar_trust_pct` of the forecast (70 %), less what the house uses. Solar first batteries get that sun first.
- *Leftover solar* then charges from the grid only the shortfall: for each expensive hour, what the house uses less the sun counted for that hour (never below 0), added up over all the expensive hours, early morning and evening. With a day and night tariff that is 07:00 to 22:00. Not up to the top level.
- All *Leftover solar* batteries hold that shortfall together, shared by their size. A battery that reaches the top level passes the rest to the others. If all of them together cannot hold it, they all fill to the top level. In the expensive hours the rest comes from another battery (*Cheap hours only* or *Solar first*) or from the grid. The Batteries screen says so. The sun before the expensive hours still fills first, and is taken off what is bought.
- All *Leftover solar* batteries hold that shortfall together, shared by their size. A battery that reaches the top level passes the rest to the others. If all of them together cannot hold it, they all fill to the top level. *Cheap hours only* and *Solar first* batteries give first: what they will hold when the expensive hours start (a *Cheap hours only* battery full, or as far as its cheap run reaches under the peak limit; a *Solar first* battery after the night's use) is taken off the shortfall. In the expensive hours the rest comes from the grid. The Batteries screen says so. The sun before the expensive hours still fills first, and is taken off what is bought.
- A forecast that gives only one figure for the day (Forecast.Solar) is spread over the hours from now, so the hours add up to that figure.
- *Cheap hours only* ignores the forecast and fills to the top level.
- *Solar first* never charges from the grid.
+92 -8
View File
@@ -22,7 +22,10 @@ OPT-28 for the second):
all of them cannot hold (`Plan.uncovered_kwh`) is not bought: another
battery (Cheap hours, Solar first) or the grid carries it.
* Cheap hours only ignores the plan and fills to the ceiling, as in P3 and
the design (section 1). Skipping the sun's share cost money in the replay
the design (section 1). OPT-36: what it, and a Solar first battery after
the night's use, will give in the expensive hours is taken off `short`
before the Leftover solar batteries share it (`_cheap_give_kwh()`,
`_solar_first_give_kwh()`). Skipping the sun's share cost money in the replay
(PR #156 review: 29.72 against 29.41 EUR on real 2026 prices).
* A Leftover solar battery charges from the grid up to the level it needs
for its part of `short`, less its share of the sun before the run:
@@ -45,7 +48,7 @@ import statistics
from dataclasses import dataclass, field
from datetime import datetime, timedelta, timezone
from .packmode import LEFTOVER_SOLAR, SOLAR_FIRST
from .packmode import CHEAP_HOURS, LEFTOVER_SOLAR, SOLAR_FIRST
from .tariff import BILLING_TZ
EFF = 0.95 # charge (and discharge) efficiency
@@ -170,6 +173,14 @@ class Plan:
short_kwh: float = 0.0
cover_kwh: dict = field(default_factory=dict)
uncovered_kwh: float = 0.0
# OPT-36: what Cheap hours batteries will give in those hours (kWh on the
# house's side), and, per Leftover solar battery, the grid kWh it buys
# less because of it.
cheap_kwh: float = 0.0
saved_kwh: dict = field(default_factory=dict)
# ...and the same for Solar first batteries (owner, 2026-11-02): their
# level when the expensive hours start, after the night's use.
solar_first_kwh: float = 0.0
def _hour_slices(t0, t1):
@@ -219,8 +230,66 @@ def _room(p, to_pct):
p.capacity_kwh)
def _cheap_give_kwh(runs, now, house, packs, *, floor_pct, ceiling_pct,
aim_w):
"""OPT-36: what the Cheap hours batteries in `packs` give in the
expensive hours, in kWh on the house's side: (the level at the start of
those hours - the floor) x capacity x `EFF`.
That level is the one its own cheap run reaches: the ceiling, or less when
the run cannot get there. From now to the run's end (or the whole run
ahead of it) the batteries together charge at their ratings, held to what
the peak aim leaves above the house's use that hour (`aim_w`, None: no
limit), shared by rating. No cheap run worth charging in: the level now.
A battery that refuses a direction counts 0."""
ps = [p for p in packs if p.mode == CHEAP_HOURS and p.refused == 0]
if not ps:
return 0.0
house_w, base_w = house
gain = {p.name: 0.0 for p in ps}
rate = {p.name: abs(float(p.rating_w)) for p in ps}
total = sum(rate.values())
if runs.charge_useful() and total > 0.0:
for s, e, h in _hour_slices(max(now, runs.cheap.start),
runs.cheap.end):
w = total
if aim_w is not None:
w = min(w, max(0.0, float(aim_w)
- max(house_w(h), base_w, 0.0)))
hrs = (e - s).total_seconds() / 3600.0
for p in ps:
gain[p.name] += w * rate[p.name] / total * hrs / 1000.0
out = 0.0
for p in ps:
cap = float(p.capacity_kwh)
lvl = max(float(p.soc), min(float(ceiling_pct),
float(p.soc) + gain[p.name] / cap * 100.0))
out += max(0.0, lvl - float(floor_pct)) / 100.0 * cap * EFF
return out
def _solar_first_give_kwh(house, now, start, packs, *, floor_pct):
"""OPT-36 (owner, 2026-11-02): what the Solar first batteries in `packs`
will still give when the expensive hours start at `start`, in kWh on the
house's side. They give at any hour, so the house's use from now to
`start` (the plan's own profile) is taken off first, shared by the energy
each holds above the floor, and what is left x `EFF` counts. A battery
that runs down to the floor before then counts 0; one that refuses a
direction counts 0."""
ps = [p for p in packs if p.mode == SOLAR_FIRST and p.refused == 0]
if not ps:
return 0.0
house_w, _base = house
use = sum(max(0.0, house_w(h)) * (e - s).total_seconds() / 3.6e6
for s, e, h in _hour_slices(now, start))
have = {p.name: max(0.0, float(p.soc) - float(floor_pct)) / 100.0
* float(p.capacity_kwh) for p in ps}
took, _ = _spread(use / EFF, have, have)
return sum(have[n] - took[n] for n in have) * EFF
def plan(forecast, house, runs, now, packs, *, trust_pct, floor_pct,
ceiling_pct):
ceiling_pct, aim_w=None):
"""The day plan, or None when there is none to make: no forecast, no
house profile yet, no price plan, or no expensive run after the cheap
run. `house` is `HouseProfile.estimator()`'s answer. `packs` are
@@ -273,11 +342,22 @@ def plan(forecast, house, runs, now, packs, *, trust_pct, floor_pct,
- trust * pv)
caps = {p.name: float(p.capacity_kwh) for p in price}
span = max(0.0, float(ceiling_pct) - float(floor_pct))
held, left = _spread(short / EFF,
{n: span / 100.0 * c for n, c in caps.items()},
caps)
# OPT-36: the Cheap hours and Solar first batteries give first;
# Leftover solar holds the rest. A Leftover solar battery's own stored energy counts through its
# level: it buys from where it is to the level it needs.
cheap = _cheap_give_kwh(runs, now, house, usable, floor_pct=floor_pct,
ceiling_pct=ceiling_pct, aim_w=aim_w) \
if price else 0.0
cheap = min(cheap, short)
sfirst = _solar_first_give_kwh(house, now, until.start, usable,
floor_pct=floor_pct) if price else 0.0
sfirst = min(sfirst, short - cheap)
others = cheap + sfirst
room_of = {n: span / 100.0 * c for n, c in caps.items()}
held, left = _spread((short - others) / EFF, room_of, caps)
held0 = _spread(short / EFF, room_of, caps)[0] if others > 0.0 else held
targets, need, share, cover = {}, {}, {}, {}
targets, need, share, cover, saved = {}, {}, {}, {}, {}
for p in price:
sh = excess * rooms[p.name] / total_room if total_room > 0 else 0.0
sh_pct = sh / caps[p.name] * 100.0
@@ -285,6 +365,9 @@ def plan(forecast, house, runs, now, packs, *, trust_pct, floor_pct,
level = max(0.0, top - sh_pct)
targets[p.name] = round(level, 2)
need[p.name] = round(_room(p, level), 3)
top0 = float(floor_pct) + held0[p.name] / caps[p.name] * 100.0
saved[p.name] = round(
_room(p, max(0.0, top0 - sh_pct)) - need[p.name], 3)
share[p.name] = round(sh, 3)
cover[p.name] = round(held[p.name] * EFF, 3)
if sun < sf_room:
@@ -295,7 +378,8 @@ def plan(forecast, house, runs, now, packs, *, trust_pct, floor_pct,
verdict = SUNNY
return Plan(verdict, round(sun, 3), round(fc_sum, 3), until.start,
targets, need, share, round(short, 3), cover,
round(left * EFF if price else 0.0, 3))
round(left * EFF if price else 0.0, 3), round(cheap, 3),
saved, round(sfirst, 3))
def hold(old: dict | None, new: dict, capacity: dict) -> dict:
+24 -8
View File
@@ -444,9 +444,16 @@ def mode_runs(c, day0, day1, now, shadow, ahead=False, rp=None):
return got if got is not None else (
floor if kind == "grid" else ceiling)
grids = [] # (end, kWh) of the grid blocks drawn
def room(kind, hrs, start):
kwh = inv.max_w / 1000.0 * hrs
lvl = start_pct(kind, start) if ahead else soc
if kind != "grid" and not ahead and cap and lvl is not None:
# OPT-36: a give block reads the level after the grid block
# before it, not the level now.
lvl = min(float(ceiling), lvl + sum(
k for e, k in grids if e <= start) / cap * 100.0)
if cap and lvl is not None:
kwh = min(kwh, max(0.0, (ceiling - lvl) if kind == "grid"
else (lvl - floor)) / 100.0 * cap)
@@ -474,6 +481,8 @@ def mode_runs(c, day0, day1, now, shadow, ahead=False, rp=None):
kwh = min(kwh, need)
if kwh < MIN_KWH:
continue
if kind == "grid":
grids.append((run.end, kwh))
price = run.mean_eur_kwh
if kind == "grid":
cov = plan_cover(c, inv.name)
@@ -522,17 +531,16 @@ def plan_cover(c, name):
grid: the level it charges to, and the expensive-hour use of the house
that charge covers and the sun will not. Read from the level control
holds (`plan_targets()`), so Home says what control does. None when the
plan buys nothing for it."""
plan buys nothing for it. OPT-36: the kWh is the plan's own cover figure
(`Plan.cover_kwh`), not one rebuilt from the level."""
dp = getattr(c, "day_plan", None)
targets = c.plan_targets() if hasattr(c, "plan_targets") else None
inv = next((i for i in c.inverters if i.name == name), None)
if dp is None or not targets or name not in targets or inv is None \
or not inv.capacity_kwh or targets[name] <= 0.0:
or not inv.capacity_kwh or targets[name] <= 0.0 \
or name not in dp.cover_kwh:
return None
floor, _ceil = arbitrage.soc_bounds(c.o)
cap = float(inv.capacity_kwh)
top = targets[name] + dp.share_kwh.get(name, 0.0) / cap * 100.0
return targets[name], max(0.0, top - floor) / 100.0 * cap * EFF
return targets[name], dp.cover_kwh[name]
def _latched_below(c) -> bool:
@@ -931,7 +939,8 @@ def battery_rows(c, shadow, running, held=False):
def day_key(c, name):
"""P4: the day plan's word for one Leftover solar or Cheap hours
battery: `sun_fills`, `tops_up`, `no_forecast`, or None."""
battery: `sun_fills`, `others_cover`, `tops_up`, `no_forecast`, or
None."""
st = plan_state(c)
if st == "no_forecast":
mode = c.pack_mode(next(i for i in c.inverters if i.name == name))
@@ -941,7 +950,14 @@ def day_key(c, name):
need = plan_need(c, name)
if need is None:
return None
return "tops_up" if need >= MIN_KWH else "sun_fills"
if need >= MIN_KWH:
return "tops_up"
# OPT-36: no grid charge because other batteries (Cheap hours only,
# Solar first) cover the hours, not the sun.
dp = getattr(c, "day_plan", None)
if dp is not None and dp.saved_kwh.get(name, 0.0) >= MIN_KWH:
return "others_cover"
return "sun_fills"
def battery_why(*, modes_on, mode, off, reading, phase, shadow, running,
+1 -1
View File
@@ -2375,7 +2375,7 @@ class Controller:
else self.house_profile.estimator(),
priceruns.keep_started(self.mode_runs, runs, now), now, packs,
trust_pct=self.solar_trust_pct, floor_pct=floor_pct,
ceiling_pct=ceiling_pct)
ceiling_pct=ceiling_pct, aim_w=self.peak_aim_w())
self.day_plan = dp
if dp is None:
self._plan_targets = None
+1
View File
@@ -357,6 +357,7 @@
"no_prices": "No prices, so it follows the house.",
"no_spread": "Prices differ too little, so it follows the house.",
"sun_fills": "Enough sun is forecast, so it does not charge from the grid.",
"others_cover": "Other batteries cover the expensive hours. It does not charge from the grid.",
"tops_up": "It charges from the grid to {pct}. That covers {kwh} of expensive hours the sun will not.",
"no_forecast": "No solar forecast, so it charges from solar only."
},
+1
View File
@@ -297,6 +297,7 @@
"no_prices": "Geen prijzen, dus ze volgt het huis.",
"no_spread": "De prijzen verschillen te weinig, dus ze volgt het huis.",
"sun_fills": "Er is genoeg zon voorspeld, dus ze laadt niet van het net.",
"others_cover": "Andere batterijen dekken de dure uren. Ze laadt niet van het net.",
"tops_up": "Ze laadt van het net tot {pct}. Dat dekt {kwh} dat de zon in dure uren niet levert.",
"no_forecast": "Geen zonneprognose, dus ze laadt alleen met zonne-energie."
},
@@ -323,6 +323,9 @@ _cases = [
day="sun_fills"), "sun_fills"),
(dict(mode=packmode.LEFTOVER_SOLAR, role="hold", day="tops_up"),
"tops_up"),
# OPT-36: a other batteries cover the hours, not the sun.
(dict(mode=packmode.LEFTOVER_SOLAR, role="hold", day="others_cover"),
"others_cover"),
(dict(mode=packmode.LEFTOVER_SOLAR, role="hold", day="no_forecast"),
"no_forecast"),
(dict(mode=packmode.LEFTOVER_SOLAR, role="charge", day="tops_up"),
+14 -3
View File
@@ -254,9 +254,16 @@ check("B on Solar first never gets a level (never charges from the grid)",
more = owner_plan(6.0, [pk(A, packmode.LEFTOVER_SOLAR),
pk(B, packmode.CHEAP_HOURS, cap=CAP_B,
rating=4600.0)])
check("B on Cheap hours fills up and is no part of A's share: A's level is "
"the same, B is not in the plan",
more.targets == res[6.0].targets and B not in more.targets
# OPT-36 (on purpose, was: "B on Cheap hours ... A's level is the same"): a
# Cheap hours battery still fills up and is still not in the plan, but what
# its own cheap run will give in the expensive hours is taken off the
# shortfall before Leftover solar shares it. Here B (10 %, 4.6 kW, the run
# 22:00-07:00) reaches the ceiling and covers all of the 6 kWh case.
check("B on Cheap hours fills up and gives first (OPT-36): A buys nothing, "
"B is not in the plan",
more.need_kwh[A] == 0.0 and more.targets[A] == FLOOR
and more.cheap_kwh > more.short_kwh - 1e-6
and more.uncovered_kwh == 0.0 and B not in more.targets
and modetrade.charge_to(pk(B, packmode.CHEAP_HOURS, cap=CAP_B), CEIL,
more.targets) == CEIL)
# The profile is net of the sun the house used, so an hour with sun is read
@@ -439,6 +446,10 @@ print("5. the screens say what control does (EN and NL)")
now = DAY0 + timedelta(minutes=18, seconds=5)
c = S.make("dynamic", now, soc=(20.0, 20.0))
c.solar_fc = S.forecast(now, 3.0, 3.0)
# OPT-36: B (Cheap hours) now gives first, so A buys only what B's cheap run
# cannot reach. A 2.5 kW house leaves B 0.95 kW of peak room in the 3.5 h run:
# B gets part of the way, A buys the rest.
c.house_profile = profile_of({h: 2500.0 for h in range(24)})
c._runs_key = None
c._mode_roles(now, None)
tg = c.plan_targets()
+723
View File
@@ -0,0 +1,723 @@
"""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.
3. A Cheap hours battery that is off, refusing, in maintenance, not live or
without capacity counts 0; Solar first is not counted; 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, and the OPT-34 latch still
holds a started run (a lowered level stays lowered).
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)
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(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(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.
use_n = sum(LIVE[h_] / 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
c3 = make(LIVE, soc=(20.0, 94.0), modes=("leftover_solar", "solar_first"))
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):
return max(soc, min(CEIL, soc + gain_kwh / 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)
check("with no peak aim the run is held by the rating only: B reaches the "
"ceiling and gives all it can; A buys only the 0.16 kWh left",
abs(p_free.cheap_kwh - (CEIL - FLOOR) / 100.0 * CAP_B * EFF) < 0.005
and 0.0 < p_free.need_kwh[A] < 0.3 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)
# ============================================================================ #
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 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(1.0), soc=(60.0, 60.0),
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 (60 % -> 6.1 kWh), not at a later charge",
g10 and abs(g10[0]["kwh"] - (60.0 - FLOOR) / 100.0 * CAP_A) < 0.02)
# ============================================================================ #
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):
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))), 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,
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())
NIGHT = at(3, 22, 0)
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=[at(3, 22, 30), at(4, 5, 0)]),
# 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()}
R = {k: f.result() for k, f in futs.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: A's commands follow the levels: it stops charging at "
"its level and ends well under the ceiling (0.20.0: 95 %), and not "
"under the level the plan ends the night with",
md["soc"][0] < 85.0 and md["soc"][0] >= m2["target_a"] - 1.0)
check("mid night: B fills as before (to the ceiling)", md["soc"][1] > 94.0)
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)