Compare commits

..
4 Commits
Author SHA1 Message Date
glenn schrooyenandClaude 6d2b9b5335 UX-77: Home's roll shares the peak room by room left to full, as control does
checks / unit (pull_request) Successful in 26m34s
infra-pin / fresh (pull_request) Successful in 2s
The cheap-run roll (cheaproll._Charge) shared the peak room by rating, control
shares it by room left to full (packshare.weights). With unequal levels one
battery's drawn block ended too early (rig F-1). The roll now uses
packshare.weights over the simulated levels, each battery capped at its rating,
a full one dropping out. test_ux77 gains two unequal-level walks and a
by-rating mutant; its walks model control's split by room too.

Co-Authored-By: Claude <noreply@anthropic.com>
2026-11-06 09:11:00 +01:00
admin 47c89a13d0 Merge pull request 'UX-77: Home draws today's grid charge for the hours control will really charge' (#234) from UX-77 into release/1.0
infra-pin / fresh (push) Successful in 1s
2026-11-06 00:11:03 +01:00
glenn schrooyenandClaude 45535e19d7 UX-77: Home draws today's grid charge for the hours control will really charge
checks / unit (pull_request) Successful in 27m7s
infra-pin / fresh (pull_request) Successful in 1s
The cheap run is sized at the full rating, so under the peak limit a battery is
still short when it ends, and control plans the next run then. Home drew only
the run under way. `cheaproll` rolls the run on the way control does
(`modetrade.energy()` + `priceruns.plan()` at each run end, lifted from
`dayview._levels()`), reading the levels at the charge power the peak limit
leaves (peak aim less the house, else the rating). `homeview.mode_runs()` and
`price_windows()` use it for today: the grid block, the Today line and the
chart's cheap bars run to where control stops, each battery's block to its
own stop. Display only: no control change.

test_ux54: today's grid block of the night-before scenario now follows the roll.

Co-Authored-By: Claude <noreply@anthropic.com>
2026-11-05 21:48:56 +01:00
admin af2d0d2f61 Merge pull request 'Sync 0.20.1 (master) into release/1.0' (#233) from sync-0.20.1 into release/1.0
infra-pin / fresh (push) Successful in 2s
2026-11-05 21:15:14 +01:00
7 changed files with 624 additions and 39 deletions
+1
View File
@@ -2,6 +2,7 @@
### Not in a release yet
- Home now draws today's charge from the grid for every hour the add-on will really charge, not only the run under way ([[UX-77]]).
- When Home Assistant passes readings on late, control no longer swings on them: it waits until the battery shows its last power request, and the peak limit no longer asks for full power on them. The page says so, and if the readings stay too late every inverter goes to 0 W ([[SAFETY-128]]).
## 0.20.1
@@ -97,9 +97,14 @@ is `SolarToday.add()` after the solar read, in a `try`, in memory.
- Prices are planned about a day ahead (`priceruns.HORIZON`), so a block of
tomorrow later than 24 h from now, or one that needs prices not yet
published, appears as the day comes near. The tab says so.
- Today's blocks are Home's, and are not extended or shortened: control may
plan more once they end, and Home draws it then. The curve stops charging
at its level (`charge_to()`) and giving at the floor.
- Today's blocks are Home's. UX-77: Home draws today's grid block to where
control will stop charging, as the tab does for tomorrow's: the cheap run
rolls on at its end while a battery is short (`cheaproll.roll_end()`, the
same loop as `dayview._levels()`), the level read at the charge power the
peak limit leaves (peak aim less the house, else the rating) and shared by
room left to full as control shares it (`packshare.weights()`, re-read as
the levels move), and each battery's block ends at its own stop. The curve stops charging at its level
(`charge_to()`) and giving at the floor.
- Today, a give run in the plan on a battery that is empty now has no block on
Home (no energy to give). The tab equals Home, and says so: "The add-on plans
to give power. No block shows while a battery has no energy to give." The
+201
View File
@@ -0,0 +1,201 @@
"""UX-77: where the cheap run rolls on. Display only; control is unchanged.
A cheap run is sized for the batteries charging together, at their full
rating, and starts at the first of the cheapest hours. The peak limit leaves
less power than that, so a battery is still short when the run ends, and
control plans the next run then (`Controller._price_runs()`). On a flat night
that goes on until the batteries are at their level or the cheap hours end.
`roll_end()` asks what control asks at each run end, of the same functions
(`modetrade.energy()`, `priceruns.plan()`), until nothing is short or the next
run does not follow on. `for_today()` reads the batteries at the level they
will have by then: the charge power is the peak limit's room (peak aim minus
the house, as `dayplan` counts it), shared as control shares it: by room
left to full (`packshare.weights()`), read again as the levels move, so a
battery that fills drops out and its share goes to the others. Where that
room is not known (no house profile or no aim), the rating is used.
DECISIONS/UX-54-forecast-tab.md has the record.
"""
import dataclasses
import math
from dataclasses import dataclass
from datetime import datetime, timedelta, timezone
from . import arbitrage, dayplan, modetrade, packshare, priceruns
from .dayplan import EFF
from .price import required_spread, uncertainty_of
SLOT_S = 900 # a price slot
STEP = timedelta(minutes=5) # how finely the level is followed
LIMIT = 24 # rolls at most (a day of one-hour runs)
def roll_end(run, points, packs_at, *, required, floor_pct, ceiling_pct,
targets=None, limit=LIMIT):
"""Where the cheap run `run` ends once control has rolled it on.
`packs_at(end)` gives the `modetrade.Pack`s as they stand when a run
ends at `end`. At each end the charge still short is asked of
`modetrade.energy()` and the next run of `priceruns.plan()`, as control
asks them. It stops when nothing is short or the next run does not follow
on, and is `run.end` when control would not roll it."""
end_t = run.end
for _ in range(limit):
e = modetrade.energy(packs_at(end_t), floor_pct, ceiling_pct,
targets=targets)
if e[0] <= 1e-6:
break
new = priceruns.plan(points, end_t, required_eur_kwh=required,
charge_kwh=e[0], charge_kw=e[1], sell_kwh=0.0,
sell_kw=0.0)
nxt = new.cheap
if nxt is None or not new.charge_useful() or nxt.start > end_t \
or nxt.end <= end_t:
break
end_t = nxt.end
return end_t
@dataclass(frozen=True)
class Roll:
"""`end`: where control stops charging, `stops`: name -> when that battery
reaches its level (rounded up to a price slot), None while it still
charges at `end`. A battery not in `stops` does not charge in the run."""
end: datetime
stops: dict
def end_for(self, name, base_end):
"""The end of `name`'s block: its stop, no later than `end`."""
if name not in self.stops:
return base_end
stop = self.stops[name]
return self.end if stop is None else min(self.end, stop)
def last(self, base_end):
"""The last end of any battery's block (the chart's cheap bar)."""
return max([base_end] + [self.end_for(n, base_end)
for n in self.stops])
def _ceil_slot(t):
return datetime.fromtimestamp(
math.ceil(t.timestamp() / SLOT_S) * SLOT_S, tz=t.tzinfo
or timezone.utc)
class _Charge:
"""The batteries' levels followed from `t0` while the chargers charge at
the room the peak limit leaves: `min(sum of ratings, aim - house)`, shared
by room left to full as control does (`_shares()`), each stopping at `modetrade.charge_to()`."""
def __init__(self, packs, t0, ceiling_pct, targets, aim_w, house):
self.packs = {p.name: p for p in packs}
self.t = t0
self.ceil = ceiling_pct
self.targets = targets
self.aim = aim_w
self.house = house
self.stops = {}
def at(self, end):
"""The packs when a run ends at `end`."""
self._to(end)
return list(self.packs.values())
def _room_w(self, hour, total):
if self.aim is None or self.house is None:
return total
house_w, base_w = self.house
return min(total, max(0.0, float(self.aim)
- max(house_w(hour), base_w, 0.0)))
def _to(self, end):
while self.t < end:
nxt = min(end, self.t + STEP)
for s, e, hour in dayplan._hour_slices(self.t, nxt):
self._charge(s, e, hour)
self.t = nxt
def _shares(self, ch, w):
"""Watts per charger: `w` shared by `packshare.weights()` over all
packs (room left to full, as control passes it), no charger above its
rating, the surplus going to the others. By rating where the room left
is not known (no capacity)."""
wt = packshare.weights(list(self.packs.values()), -1.0)
if wt is None:
wt = {p.name: abs(p.rating_w) for p in ch}
wt = {p.name: wt.get(p.name, 0.0) for p in ch}
cap = {p.name: abs(p.rating_w) for p in ch}
out, open_, left = {}, set(wt), float(w)
while open_:
tot = sum(wt[n] for n in open_)
if tot <= 0.0:
for n in open_:
out[n] = 0.0
break
over = [n for n in open_ if left * wt[n] / tot > cap[n] + 1e-9]
if not over:
for n in open_:
out[n] = left * wt[n] / tot
break
for n in over:
out[n] = cap[n]
left -= cap[n]
open_.discard(n)
return out
def _charge(self, s, e, hour):
ch = modetrade.chargers(list(self.packs.values()), self.ceil,
targets=self.targets)
total = sum(abs(p.rating_w) for p in ch)
if not ch or total <= 0.0:
return
w = self._room_w(hour, total)
hrs = (e - s).total_seconds() / 3600.0
share = self._shares(ch, w)
for p in ch:
self.stops.setdefault(p.name, None)
if not p.capacity_kwh:
continue
to = modetrade.charge_to(p, self.ceil, self.targets)
kwh = share[p.name] * hrs / 1000.0 * EFF
room = (to - p.soc) / 100.0 * float(p.capacity_kwh)
if kwh <= 0.0:
continue
if kwh >= room - 1e-9:
self.stops[p.name] = _ceil_slot(
s + (e - s) * (room / kwh))
kwh = room
self.packs[p.name] = dataclasses.replace(
p, soc=p.soc + kwh / float(p.capacity_kwh) * 100.0)
def for_today(c, rp, now):
"""The `Roll` of the cheap run `rp.cheap` now under way or to come today,
or None when control does not roll it on (no run left, charging not
useful, the trade gate shut, nothing short at its end). Display only:
nothing here is fed back to control."""
if rp is None or not rp.on or rp.cheap is None or rp.cheap.end <= now \
or not rp.charge_useful() \
or getattr(c, "mode_gate", None) is not None:
return None
plan = c.fresh_tariff_plan(now)
if plan is None:
return None
floor, ceil = arbitrage.soc_bounds(c.o)
hp = getattr(c, "house_profile", None)
sim = _Charge(c._mode_packs(), max(now, rp.cheap.start), ceil,
c.plan_targets(), c.peak_aim_w(),
None if hp is None else hp.estimator())
required = required_spread(
arbitrage.threshold_eur_kwh(c.o),
uncertainty_of(plan.read.product)
| arbitrage.weather_uncertainty(True, c.solar_fc,
bool(c.weather_held(now))))
end = roll_end(rp.cheap, plan.read.points, sim.at, required=required,
floor_pct=floor, ceiling_pct=ceil,
targets=c.plan_targets())
if end <= rp.cheap.end:
return None
return Roll(end, dict(sim.stops))
+13 -19
View File
@@ -19,7 +19,8 @@ empty chart. DECISIONS/UX-54-forecast-tab.md has the record.
import dataclasses
from datetime import datetime, timedelta
from . import arbitrage, dayplan, homeview as hv, modetrade, priceruns
from . import arbitrage, cheaproll, dayplan, homeview as hv, modetrade
from . import priceruns
from . import solarfc
from .daycurve import Batt, curve
from .packmode import SOLAR_FIRST
@@ -210,25 +211,18 @@ def _levels(c, est, fc, trust, now, end, runs, floor, ceil, notes_b, shadow,
arbitrage.threshold_eur_kwh(c.o),
uncertainty_of(plan.read.product)
| arbitrage.weather_uncertainty(True, c.solar_fc))
end_t = cheap.end
for _ in range(24):
ext = dataclasses.replace(rp, cheap=priceruns.Run(
def packs_at(end_t):
"""The packs at the curve's level when a run ends at `end_t`."""
at = dataclasses.replace(rp, cheap=priceruns.Run(
cheap.start, end_t, cheap.mean_eur_kwh))
rt, lv = settle(ext, lv0)
at_end = [dataclasses.replace(packs[n], soc=v)
for n, v in ((n, _level_at(lv.get(n, ()), end_t))
for n in lv) if v is not None]
e = modetrade.energy(at_end, floor, ceil, targets=targets)
if e[0] <= 1e-6:
break
new = priceruns.plan(
plan.read.points, end_t, required_eur_kwh=required,
charge_kwh=e[0], charge_kw=e[1], sell_kwh=0.0, sell_kw=0.0)
nxt = new.cheap
if nxt is None or not new.charge_useful() or nxt.start > end_t \
or nxt.end <= end_t:
break
end_t = nxt.end
_rt, lv = settle(at, lv0)
return [dataclasses.replace(packs[n], soc=v)
for n, v in ((n, _level_at(lv.get(n, ()), end_t))
for n in lv) if v is not None]
end_t = cheaproll.roll_end(
cheap, plan.read.points, packs_at, required=required,
floor_pct=floor, ceiling_pct=ceil, targets=targets)
ext = dataclasses.replace(rp, cheap=priceruns.Run(
cheap.start, end_t, cheap.mean_eur_kwh))
rt, lv = settle(ext, lv0)
+19 -6
View File
@@ -19,7 +19,7 @@ section at the bottom and is held to STYLE.md by test_messages.py.
import contextlib
from datetime import datetime, timedelta, timezone
from . import arbitrage, messages, peakguard, uitext
from . import arbitrage, cheaproll, messages, peakguard, priceruns, uitext
from .dayplan import EFF
from .maintenance import CHARGE as M_CHARGE, DRAIN as M_DRAIN, HOLD as M_HOLD
from .maintenance import IDLE as M_IDLE, next_slot_start
@@ -218,14 +218,20 @@ def price_windows(c, plan, verdict, now=None):
"""`[(kind, start, end)]`: where the add-on itself buys ("cheap") and
gives ("dear"). With battery modes on that is the kept `priceruns.RunPlan`
(`c.mode_runs`): the cheap run when charging in it pays and the trade gate
allows it (UX-53), and every give run. With them off it is the arbitrage
verdict's legs the add-on will take, the ones `trade_runs()` draws
(`trade_legs()`, given `now`). Nothing is classified here."""
allows it (UX-53), and every give run. UX-77: the cheap run is drawn to
where control stops charging as it rolls on (`cheaproll`). With them off
it is the arbitrage verdict's legs the add-on will take, the ones
`trade_runs()` draws (`trade_legs()`, given `now`). Nothing is classified
here."""
if getattr(c, "modes_on", False):
rp = getattr(c, "mode_runs", None)
if rp is None or not rp.on or not _price_mode_in_use(c):
return []
out = [("cheap", rp.cheap.start, rp.cheap.end)] \
end = rp.cheap.end
roll = None if now is None else cheaproll.for_today(c, rp, now)
if roll is not None:
end = roll.last(end) # UX-77: to where control stops
out = [("cheap", rp.cheap.start, end)] \
if rp.charge_useful() and getattr(c, "mode_gate", None) is None \
else []
return out + [("dear", g.start, g.end) for g in rp.give]
@@ -422,6 +428,7 @@ def mode_runs(c, day0, day1, now, shadow, ahead=False, rp=None):
`ahead=True`) a charge from the floor and a give from the ceiling. The
same filter, the same `MIN_KWH`. `rp`: a run plan to read instead of
`c.mode_runs`. Home never passes either."""
today = rp is None and not ahead
rp = rp if rp is not None else getattr(c, "mode_runs", None)
if getattr(c, "mode_tick", None) is None:
return [], None # no cycle yet: nothing is known
@@ -431,6 +438,9 @@ def mode_runs(c, day0, day1, now, shadow, ahead=False, rp=None):
return [], "not_worth"
floor, ceiling = arbitrage.soc_bounds(c.o)
gate = getattr(c, "mode_gate", None)
# UX-77: today's grid block runs to where control stops charging (the run
# rolls on when it ends), each battery's to its own stop.
roll = cheaproll.for_today(c, rp, now) if today else None
out = []
for i, inv in enumerate(c.inverters):
pm = c.pack_mode(inv)
@@ -465,7 +475,10 @@ def mode_runs(c, day0, day1, now, shadow, ahead=False, rp=None):
if gate is None and rp.charge_useful() and (
need > 0.0 if need is not None
else pm.mode in GRID_CHARGE_MODES):
blocks.append(("grid", rp.cheap))
cheap = rp.cheap if roll is None else priceruns.Run(
rp.cheap.start, roll.end_for(inv.name, rp.cheap.end),
rp.cheap.mean_eur_kwh)
blocks.append(("grid", cheap))
sell = rp.sell if pm.sell and pm.can_sell else None
for g in rp.give:
if sell is not None and g.start <= sell.start < g.end:
+24 -11
View File
@@ -39,7 +39,8 @@ sys.path.insert(0, os.path.join(HERE, "tools"))
logging.disable(logging.WARNING)
import ux54_scenarios as S # noqa: E402
from app import arbitrage, dayplan, dayview, homeview, modetrade # noqa: E402
from app import arbitrage, cheaproll, dayplan, dayview, homeview # noqa: E402
from app import modetrade # noqa: E402
from app import packmode, packshare, peakguard, solarfc, uitext # noqa: E402
from app.main import pack_energy_of # noqa: E402
from app.price import Uncertainty # noqa: E402
@@ -232,11 +233,19 @@ END_S = S.brussels(END).strftime("%H:%M")
check(f"the cheap run is 01:00 to {END_S} (02:30 for this house)",
c_n.mode_runs.cheap.start == at(1, 0, 0) and END == at(2, 30, 0))
nb = tab(c_n, at(0, 18))["days"][0]["batteries"]
check("in the night before the cheap run both charge from the grid in it "
f"(01:00 to {END_S}), then give at 17:00",
all([r["kind"] for r in b["blocks"]] == ["grid", "give"] for b in nb)
and all(b["blocks"][0]["from"] == "01:00"
and b["blocks"][0]["to"] == END_S for b in nb))
# UX-77: today's block runs to where control stops charging: the 1.5 h run is
# sized at the full rating, the peak limit leaves 3.1 kW, so control plans the
# next run when it ends (`cheaproll`), and each battery's block ends at its stop.
RL = cheaproll.for_today(c_n, c_n.mode_runs, at(0, 18))
ENDS = [S.brussels(RL.end_for(inv.name, END)).strftime("%H:%M")
for inv in c_n.inverters]
check("in the night before the cheap run both charge from the grid from "
f"01:00 to where control stops ({ENDS}, the run rolls on after "
f"{END_S}), then give at 17:00",
RL is not None and RL.end > END and ENDS[0] < ENDS[1]
and all([r["kind"] for r in b["blocks"]] == ["grid", "give"] for b in nb)
and all(b["blocks"][0]["from"] == "01:00" for b in nb)
and [b["blocks"][0]["to"] for b in nb] == ENDS)
# UX-53: no grid block while the gate refuses.
for why in ("peak", "not_worth"):
c_n.mode_gate = why
@@ -255,12 +264,16 @@ def at_h(b, h):
LVL_A = c_n.plan_targets()[c_n.inverters[0].name]
END_HS = [(S.brussels(RL.end_for(inv.name, END)).replace(tzinfo=None)
- S.brussels(DAY0).replace(tzinfo=None)).total_seconds() / 3600.0
for inv in c_n.inverters]
check(f"the curve charges in a grid block: A (planned level {LVL_A:.0f} %) "
"and B (Cheap hours, to the end of its run) are well above where they "
"were at 01:00, and stay put after the run",
all(at_h(b, END_H) - at_h(b, 1.0) > 8.0
and abs(at_h(b, END_H + 0.5) - at_h(b, END_H)) < 1e-6
for b in nb) and nb[0]["levels"][0][1] == 30.0)
"and B (Cheap hours, to the end of its block) are well above where "
"they were at 01:00, and stay put after the block",
all(at_h(b, eh) - at_h(b, 1.0) > 8.0
and all(abs(v - at_h(b, eh)) < 1e-6 for x, v in b["levels"]
if eh + 1e-6 < x < eh + 3.0)
for b, eh in zip(nb, END_HS)) and nb[0]["levels"][0][1] == 30.0)
# Modes off: the verdict's legs.
c_f = S.make("off", at(10, 7))
fb = tab(c_f, at(10, 7))
+358
View File
@@ -0,0 +1,358 @@
"""UX-77: Home draws today's grid charge for the hours control will charge.
The real `Controller._mode_roles()` and `homeview.build()`, stepped each
minute over a flat night (the walk of the investigation): at five times, every
minute control charges battery X in that local day lies inside X's drawn grid
block, the block does not run past the hour control stops charging X by more
than one slot, and the chart's cheap bars cover every hour with a charge.
The peak-bound night is the one that needs the peak limit's room (a mutant that
rolls at the full rating fails it, part 3); dynamic prices draw what they drew
before (part 4). Mutations (MUTATION.md, purged copies): listed in the pull
request. Run: python3 test_ux77.py
"""
import json
import logging
import os
import sys
from datetime import datetime, timedelta, timezone
HERE = os.path.dirname(os.path.abspath(__file__))
sys.path.insert(0, HERE)
sys.path.insert(0, os.path.join(HERE, "tools"))
logging.disable(logging.WARNING)
import ux54_scenarios as S # noqa: E402
from app import dayplan, homeview, modetrade, packmode, packshare # noqa: E402
from app.main import Controller # 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
try: # not on the base, where it is red
from app import cheaproll
except ImportError:
cheaproll = None
UTC = timezone.utc
EFF = 0.95
AIM = 3450.0
SLOT = timedelta(hours=1) # the price slot of the day/night tariff
QUARTER = timedelta(minutes=15) # the run's own slot
SAMPLES = ("23:30", "00:30", "01:30", "02:30", "04:30")
OLD = ("leftover_solar", "cheap_hours") # the owner's old shape
NOW = ("leftover_solar", "leftover_solar") # the current one
PASS = FAIL = 0
def check(name, ok, detail=""):
global PASS, FAIL
if ok:
PASS += 1
else:
FAIL += 1
print(f"FAIL {name} {detail}")
def no_ties(h):
"""Two dips, no two slots at the same price: nothing rolls on."""
return (0.05 if 1.0 <= h < 2.0 else 0.08 if 5.0 <= h < 6.0
else 0.30) + 0.0001 * h
def shape(night=None):
d = {h: 0.973 * (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)}
if night is not None:
for h in list(range(0, 7)) + [22, 23]:
d[h] = night
return d
def profile(sh, now):
d = now.astimezone(BILLING_TZ).date()
return dayplan.HouseProfile({(d - timedelta(days=k)).isoformat():
{str(h): w for h, w in sh.items()} for k in (1, 2, 3)})
def loc(t):
return t.astimezone(BILLING_TZ)
def hm(s, day):
h, m = map(int, s.split(":"))
return day.replace(hour=h, minute=m)
def split_by_room(w, ch, rating, wt, invs):
"""Control's split of `w` over the chargers `ch`: by `packshare.weights()`,
no battery over its rating, the surplus to the others."""
p, open_, left = [0.0, 0.0], set(ch), w
while open_:
tot = sum(wt[invs[i].name] for i in open_)
over = [i for i in open_
if left * wt[invs[i].name] / tot > rating[i] + 1e-9]
if not over:
for i in open_:
p[i] = left * wt[invs[i].name] / tot
break
for i in over:
p[i] = rating[i]
left -= rating[i]
open_.discard(i)
return p
def walk(modes, soc0, house_w, night=None, dynamic=False, samples=SAMPLES,
start_h=22, end_h=7, plain=None, by_room=True, fc=(10.2, 10.0)):
"""Walk 22:00-07:15. Returns (charging, homes, commands):
charging[minute] = set of batteries control charges (power > 0),
homes[sample] = Home's `/status` plan at that minute, commands = the
(minute, unit, role, size) control gave."""
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)
c = Controller(opts, S.Hass(), S.Store(), S.Pub(), stop=S.asyncio.Event())
c.auto = True
soc = list(soc0)
caps = [inv.capacity_kwh for inv in c.inverters]
rating = [inv.max_w for inv in c.inverters]
for i, inv in enumerate(c.inverters):
inv.batt, inv.last_write = 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.pv_w = 0.0
c.peak_guard = PeakDecision(engaged=False, floor_w=None, ceiling_w=7000.0,
ceiling_from="cap", projected_w=800.0)
start = datetime(2026, 11, 1, start_h, 0, tzinfo=BILLING_TZ).astimezone(UTC)
end = datetime(2026, 11, 2, end_h, 15, tzinfo=BILLING_TZ).astimezone(UTC)
c.house_profile = profile(shape(night), start)
now, plan_at = start, None
charging, homes, cmds = {}, {}, []
while now < end:
lt = loc(now)
today, tomorrow = fc if lt.hour < 12 else fc[::-1]
c.solar_fc = S.forecast(now, today, tomorrow, members=3)
c.solar_fc_why = None
if plan_at is None or (now - plan_at).total_seconds() >= 1800:
c.tariff_plan = (S.quarters(now, dynamic) if dynamic else
static_plan(opts, now, now + timedelta(hours=48),
now=now))
plan_at = now
c.tariff_at = plan_at
for i, inv in enumerate(c.inverters):
inv.soc = round(soc[i], 1)
room = AIM - house_w
c.arb_headroom_w = room
c.grid = house_w
tick = c._mode_roles(now, None)
size = modetrade.size(tick, (-sum(rating), sum(rating)), room) \
if tick.action else 0.0
ch = [i for i, inv in enumerate(c.inverters)
if tick.roles.get(inv.name) == modetrade.CHARGE]
p = [0.0, 0.0]
if size < 0 and ch:
if by_room: # as control: room left to full, rating caps
wt = packshare.weights(c.pack_energy, -1.0)
p = split_by_room(-size, ch, rating, wt, c.inverters)
else:
tot = sum(rating[i] for i in ch)
for i in ch:
p[i] = -size * rating[i] / tot
for i in range(2):
soc[i] = min(100.0, soc[i] + p[i] / 1000.0 / 60.0 * EFF
/ caps[i] * 100.0)
charging[lt.replace(second=0, microsecond=0)] = {
i for i in range(2) if p[i] > 0.0}
cmds.append((lt.strftime("%H:%M"), tick.action,
tuple(sorted(tick.roles.items())), round(size)))
if lt.strftime("%H:%M") in samples:
homes[lt.replace(second=0, microsecond=0)] = homeview.build(
c, level="ok", checks=[], now=now, lang="en")
if plain is not None: # the same, with no roll
real = cheaproll.for_today
cheaproll.for_today = lambda *_a: None
try:
plain[lt.replace(second=0, microsecond=0)] = \
homeview.build(c, level="ok", checks=[], now=now,
lang="en")
finally:
cheaproll.for_today = real
now += timedelta(minutes=1)
return charging, homes, cmds
def blocks_of(st, i):
"""Battery i's drawn grid blocks as (start, end) local datetimes."""
return [(loc(datetime.fromisoformat(r["start"])),
loc(datetime.fromisoformat(r["end"])))
for r in st["plan"]["runs"]
if r["kind"] == "grid" and r["battery"] == i]
def today_of(st, i):
return [(t["from"], t["to"]) for t in st["batteries"][i]["today"]]
def violations(charging, homes, slot=SLOT):
"""What is wrong with Home's drawing at each sample."""
bad = []
for t, st in sorted(homes.items()):
day_end = (t.replace(hour=0, minute=0) + timedelta(days=1))
mins = {m: who for m, who in charging.items() if t <= m < day_end}
for i in range(2):
mine = sorted(m for m, who in mins.items() if i in who)
bl = blocks_of(st, i)
for m in mine:
if not any(a <= m < z for a, z in bl):
bad.append(f"{t:%H:%M} battery {i}: charges at "
f"{m:%H:%M}, not in a block {bl}")
break
if mine:
stop = mine[-1] + timedelta(minutes=1)
if slot == QUARTER: # the last charging quarter's end
stop += timedelta(minutes=-stop.minute % 15)
for a, z in bl:
if z > stop + slot:
bad.append(f"{t:%H:%M} battery {i}: block to "
f"{z:%H:%M}, stops {stop:%H:%M}")
# the Today line says the same as the block
tl = today_of(st, i)
for m in mine:
if not any(f <= m.strftime("%H:%M") < (z if z != "24:00"
else "99") for f, z in tl):
bad.append(f"{t:%H:%M} battery {i}: Today line {tl} "
f"has no {m:%H:%M}")
break
cheap = {b["label"][:2] for b in st["plan"]["prices"]
if b["kind"] == "cheap"}
for m in mins:
if mins[m] and m.strftime("%H") not in cheap:
bad.append(f"{t:%H:%M}: charge at {m:%H:%M}, hour not cheap")
break
return bad
# --------------------------------------------------------------------------- #
# Run
# --------------------------------------------------------------------------- #
def show(tag, ch, homes, want_bad, slot=SLOT):
bad = violations(ch, homes, slot)
if want_bad:
check(f"{tag}: red", bool(bad), "no violation found")
else:
check(f"{tag}: every charging minute is drawn, no block runs on",
not bad, "\n " + "\n ".join(bad))
return bad
def charged(ch, i, after):
return [m for m, who in ch.items() if i in who and m >= after]
def main():
global cheaproll
if cheaproll is None:
print("no app/cheaproll.py: the base; the walk is red there")
cheaproll = type("Off", (), {"for_today": staticmethod(lambda *a: None)})
print("1. the walk: no-bind (0.6 kW house) and peak-bound (2.85 kW house)")
CASES = (("old shape, no bind", OLD, (20.0, 20.0), 584.0, None),
("current shape, no bind", NOW, (20.0, 20.0), 584.0, None),
("old shape, peak-bound", OLD, (40.0, 90.0), 2850.0, 2850.0),
("current shape, peak-bound", NOW, (40.0, 90.0), 2850.0, 2850.0))
# the rig's checks 2 and 3 (UX-77 round 2): room left and rating disagree
UNEQUAL = (("unequal levels, A+B Leftover solar, no bind", NOW,
(30.0, 20.0), 584.0, None, (5.0, 5.0)),
("unequal levels, A Leftover solar + B Cheap hours", OLD,
(40.0, 88.0), 2850.0, 2850.0, (10.2, 10.0)))
walked = {}
for tag, modes, soc, hw, night in CASES:
ch, homes, _cmds = walk(modes, soc, hw, night=night)
walked[tag] = (ch, homes)
show(tag, ch, homes, False)
# the check has something to check: charging goes on past 00:30
day = datetime(2026, 11, 2, 0, 30, tzinfo=BILLING_TZ)
check(f"{tag}: control charges after 00:30", bool(
charged(ch, 0, day) or charged(ch, 1, day)))
print("2. a battery with nothing to charge gets no block")
ch, homes, _c = walk(OLD, (40.0, 94.0), 2850.0, night=2850.0)
st = homes[datetime(2026, 11, 2, 2, 30, tzinfo=BILLING_TZ)]
check("B is at its level at 02:30: no block", blocks_of(st, 1) == [],
str(blocks_of(st, 1)))
check("A still has one at 02:30", bool(blocks_of(st, 0)))
print("3. a roll at the full rating fails the peak-bound night")
if hasattr(cheaproll, "_Charge"):
real = cheaproll._Charge._room_w
cheaproll._Charge._room_w = lambda self, hour, total: total
try:
for tag, modes, soc, hw, night in CASES[2:]:
ch, homes, _c = walk(modes, soc, hw, night=night)
show(f"mutant, {tag}", ch, homes, True)
finally:
cheaproll._Charge._room_w = real
print("4. dynamic prices: Home is what it was")
for tag, modes, soc, hw, night in (CASES[2], CASES[3], CASES[1]):
plain = {}
ch, homes, _c = walk(modes, soc, hw, night=night, dynamic=no_ties,
plain=plain)
same = all(json.dumps(homes[t], sort_keys=True, default=str)
== json.dumps(plain[t], sort_keys=True, default=str)
for t in homes)
check(f"dynamic, {tag}: same as with no roll", same)
check(f"dynamic, {tag}: and there are grid blocks to compare",
any(r["kind"] == "grid" for st in homes.values()
for r in st["plan"]["runs"]))
print("5. unequal levels: the roll shares the room as control does")
for tag, modes, soc, hw, night, fc in UNEQUAL:
ch, homes, _c = walk(modes, soc, hw, night=night, fc=fc)
show(tag, ch, homes, False, QUARTER)
day = datetime(2026, 11, 2, 0, 30, tzinfo=BILLING_TZ)
check(f"{tag}: control charges after 00:30", bool(
charged(ch, 0, day) or charged(ch, 1, day)))
if hasattr(cheaproll, "_Charge") \
and hasattr(cheaproll._Charge, "_shares"):
real = cheaproll._Charge._shares
cheaproll._Charge._shares = lambda self, ch, w: {
p.name: w * abs(p.rating_w) / sum(abs(q.rating_w) for q in ch)
for p in ch}
try:
for tag, modes, soc, hw, night, fc in UNEQUAL:
ch, homes, _c = walk(modes, soc, hw, night=night, fc=fc)
show(f"mutant by rating, {tag}", ch, homes, True, QUARTER)
finally:
cheaproll._Charge._shares = real
print("6. the pieces")
if hasattr(cheaproll, "Roll"):
t = datetime(2026, 11, 2, 1, 46, tzinfo=UTC)
check("a stop is rounded up to the slot", cheaproll._ceil_slot(t)
== datetime(2026, 11, 2, 2, 0, tzinfo=UTC))
t = datetime(2026, 11, 2, 1, 45, tzinfo=UTC)
check("a stop on a slot stays", cheaproll._ceil_slot(t) == t)
e0 = datetime(2026, 11, 2, 0, 0, tzinfo=UTC)
e1 = datetime(2026, 11, 2, 6, 0, tzinfo=UTC)
r = cheaproll.Roll(e1, {"a": None, "b": datetime(2026, 11, 2, 2, 0,
tzinfo=UTC)})
check("a battery still charging ends at the roll", r.end_for("a", e0) == e1)
check("a battery at its level ends at its stop",
r.end_for("b", e0) == datetime(2026, 11, 2, 2, 0, tzinfo=UTC))
check("a battery that does not charge keeps the run's end",
r.end_for("c", e0) == e0)
check("the chart's bar runs to the last block", r.last(e0) == e1)
print(f"\n{PASS} passed, {FAIL} failed")
return FAIL
if __name__ == "__main__":
sys.exit(1 if main() else 0)