Merge pull request 'OPT-11 P3: price runs per battery' (#153) from opt11-p3-price-runs into release/1.0
infra-pin / fresh (push) Successful in 2s

OPT-11 P3: price runs per battery (owner option A: selling off, Leftover solar solar-only until P4) (#153)
This commit was merged in pull request #153.
This commit is contained in:
2026-09-29 11:37:46 +02:00
20 changed files with 2350 additions and 127 deletions
@@ -29,4 +29,49 @@ P1 is its section 6 P1, and the pull request lists what was measured.
page's English names, pinned by a test, and are not renamed with the page:
automations send them.
- In P1 only Off closes a direction; Leftover solar differs by charging last.
Price runs, selling (`pack_sell`, reserved) and the forecast are P3/P4.
- P3 (price runs, `priceruns.py`, `modetrade.py`): each battery's range is
narrowed by its role this tick (hold, give, charge, rest, sell); main.py
files the intersection with the switch, so it still holds 0 W. With modes on
the site trade is filed by the modes, sized from the loop bound (so no
battery a mode or a refusal holds is asked for watts, P1's unmet share) and
the guard's headroom; `arbitrage.decide()` runs only with modes off, which
stays byte-identical to 7a00f12 (`test_modes_p3.py` part 1).
- SAFETY-46 item 2 is decided as the owner answered (2026-09-25): a trade is
a run of consecutive firm slots. It applies with battery modes on; the
modes-off trade keeps its one-period shape, because that path must not move.
- Choices that were measured, not assumed (tools/opt11p3_replay.py):
- a sale needs the guard to have measured the quarter, not to be clear: a
5 kW discharge alone reads "engaged" on export and flipped the sale every
5 s;
- a grid charge makes the others rest, so it starts only at 500 W and, once
the guard or the headroom stops it, stays off for that 15-minute period
(without: 13 flips in 4 minutes under a 3.5 kW house);
- stored energy is valued at the last cheap run for 12 h, and a charge
needs its own run to clear the spread (without: a night charge was spent
on flat morning prices, then charging went on at them);
- a run edge reseeds the loop at the house's level (a 3 kW import spike
removed; about 0.1 x rating still overshoots as export).
- Replay (tools/opt11p3_replay.py, 16 recorded days, EPEX hourly prices a
year earlier, modelled batteries), net EUR, lower is better: P2 19.33, P3
Solar first + Leftover solar 19.57, with a sale 19.60, the old one-period
trade 18.14. Review of #153 (same model, with a per-action split):
- the cheapest hours were at midday on 15 of 16 days, so Leftover solar (acting
as Cheap hours until P4) bought 12.2 kWh there that the sun would have
given: 7 kWh more exported, 12 kWh less solar stored. Holding for the
expensive hours saved 4 kWh of import, not enough. With no grid charge on
Leftover solar: 19.44;
- the old trade charged one hour, so the afternoon sun still fit;
- a sale earned the wholesale price (2.27 EUR for 11.4 kWh) and the house
then bought the same energy back (2.29 EUR): a wash before wear. It pays
only when the sale hour's wholesale price beats the later import price,
levies included (about 0.13 EUR/kWh above that hour's wholesale here);
- valuing the charge left at the end, wear and real Flemish levies make
selling worse and do not change the order.
- Owner decision 2026-09-29 (option A): selling is off until it can be judged
on the price the grid pays (`packmode.SELLING_ENABLED`, code kept), and
Leftover solar takes solar only until P4's forecast
(`packmode.GRID_CHARGE_MODES`); Cheap hours only still charges from the grid. Replayed
(16 days, as above): Solar first + Leftover solar 19.44 EUR net against
P2's 19.33; Solar first + Cheap hours only 19.57.
- Not done in P3: the forecast (P4), and a headroom too short for the run
moving the rest to the next cheap periods.
+23 -5
View File
@@ -520,9 +520,26 @@ below.
Off by default. Turn it on, restart the add-on, and the Batteries screen (and Settings, Simple) gives each battery two controls. They apply at once, with no restart.
- **On or off.** Off asks the battery for 0 W and starts no monthly cycle on it. A cycle that is running stops, and the add-on log says so. The peak limit may still use it, but only for what the other batteries cannot cover.
- **Mode.** *Solar first* charges from spare solar before the other batteries. *Leftover solar* charges from the solar they cannot take. Both cover the house at any hour. Charging from the grid in cheap hours, and selling, come in a later update.
- **Mode.**
- *Solar first* charges from spare solar before the other batteries and covers the house at any hour. It never charges from the grid and never sells.
- *Leftover solar* charges from the solar the others cannot take, never from the grid (until the solar forecast arrives). It saves its power for the expensive hours and covers the house then.
- *Cheap hours only* charges from the grid in the cheapest hours, saves its power for the expensive hours and covers the house then.
- **Selling** is off until it can be judged on the price the grid pays (owner decision 2026-09-29, option A).
The peak limit, the safety stops and the monthly cycle always come before a battery's mode.
With battery modes on, these choices replace the *Charging and selling on price* switch (`arbitrage_enabled`); that switch is only read with modes off.
**Cheap and expensive hours.** They come from the published day-ahead prices, as runs of consecutive 15-minute periods, not a single period.
- The cheap run is the cheapest stretch long enough to fill the batteries that may charge, at their power.
- An expensive hour costs at least *Battery cycle cost* (`cycle_cost_threshold_eur`, widened as for trading) more than the cheap run. The batteries hold their power until then.
- The sale uses the most expensive stretch inside those hours, and never a period below 0.
- A carried (not yet published) price is never used. A run that has started runs to its end.
- With no prices, or too small a gap, every battery follows the house.
**What always comes first.**
- The peak limit, the safety stops and the monthly cycle come before a battery's mode.
- A grid charge is sized so the 15-minute period stays under the peak limit less its margin (`peak_margin_w`). It starts only at 500 W or more. Once the peak limit stops it, it stays off for the rest of that 15-minute period.
- While one battery charges from the grid, the others rest, and the house uses the grid.
- A battery that refuses to charge or give power takes no part in a run in that direction.
**In Home Assistant.** With MQTT set up, each battery also gets a *Mode* list and a *Use* switch, for example `select.goodwe_inverter_a_mode` and `switch.goodwe_inverter_a_use`. They do the same as the Batteries screen. Turn battery modes off and restart, and they are removed.
@@ -530,6 +547,8 @@ The peak limit, the safety stops and the monthly cycle always come before a batt
*For installers:* a mode is a limit at priority 35, kept apart from the hard limits. It gives way only as far as a limit above it (the peak guard) needs. A mode range always holds 0 W, so a mode can forbid a direction and never force one. With one inverter the mode also bounds the loop. The record is `DECISIONS/OPT-11-battery-modes.md`.
*For installers:* when a charge or sale ends, the loop restarts at the house's own level rather than at the trade's watts. On the replay model that removes a ~3 kW import spike after a 2.5 kW charge and halves the settling time; about 0.1 × the battery's power still overshoots as export. The *Mode* list gains *Cheap hours only* in Home Assistant, at the end.
### Reading the grid-following loop: `Loop integrator`, `Loop integrator lead`, `Loop reason`
The control law is a bounded integrator: an accumulator that is **not** the
@@ -3263,9 +3282,8 @@ The **Batteries** tab has a card per battery: its level, and three lines.
**Now** says what it does, **Today** lists the blocks planned for it (price
trades, the monthly cycle, the peak limit), and **Why** says the reason. With
battery modes on, each card also has the On/Off switch and the mode list; both
apply at once. Charging in cheap hours and selling are greyed out until a later
update, and Today reads "Not planned yet" when nothing is planned for the
battery.
apply at once. Today reads "Not planned yet" until the add-on has made the
day's price runs.
An Off battery that the peak limit still needs says so, with the power it
gives: "Off, but gives 0.5 kW for the peak limit."
+142 -17
View File
@@ -10,8 +10,9 @@ note, section 3.3.
The plan shows only what the add-on does today: trading on price (the
arbitrage verdict, the same call the control cycle makes), the monthly battery
cycle, the peak limit and "follows the house". OPT-11 P1 adds each battery's
switch and mode to its row; the mode plan arrives with P3/P4, so nothing here
invents one. Every sentence is in the words
switch and mode to its row; P3 adds each battery's price runs and its role this
cycle (`Controller.mode_runs`, `mode_tick`), read as the control cycle left
them, so nothing here invents a plan. Every sentence is in the words
section at the bottom and is held to STYLE.md by test_messages.py.
"""
@@ -20,7 +21,7 @@ from datetime import datetime, timedelta, timezone
from . import arbitrage, messages, peakguard, uitext
from .maintenance import CHARGE as M_CHARGE, DRAIN as M_DRAIN, HOLD as M_HOLD
from .maintenance import IDLE as M_IDLE
from .packmode import LEFTOVER_SOLAR, SOLAR_FIRST
from .packmode import CHEAP_HOURS, GRID_CHARGE_MODES, LEFTOVER_SOLAR, SOLAR_FIRST
from .price import Absence, Uncertainty, brussels
from .static_tariff import static_plan
from .tariff import BILLING_TZ
@@ -312,6 +313,75 @@ def trade_runs(c, plan, verdict, day0, day1, now, shadow):
return runs, later, None
PRICE_MODES = (LEFTOVER_SOLAR, CHEAP_HOURS)
def mode_runs(c, day0, day1, now, shadow):
"""OPT-11 P3: each battery's price runs today, as the control cycle keeps
them (`c.mode_runs`): the cheap run on the batteries that may charge
from the grid, the give runs on Leftover solar and Cheap hours, and the
sell run on those with selling on. Amounts are upper bounds ("up to"),
from the battery's own room, rating and the run's length. Returns
`(runs, no_trade_key)`."""
rp = getattr(c, "mode_runs", None)
if getattr(c, "mode_tick", None) is None:
return [], None # no cycle yet: nothing is known
if rp is None:
return [], "no_prices"
if not rp.on:
return [], "not_worth"
floor, ceiling = arbitrage.soc_bounds(c.o)
out = []
for i, inv in enumerate(c.inverters):
pm = c.pack_mode(inv)
if not pm.use or pm.mode not in PRICE_MODES:
continue
soc, cap = inv.soc, inv.capacity_kwh
def room(kind, hrs):
kwh = inv.max_w / 1000.0 * hrs
if cap and soc is not None:
kwh = min(kwh, max(0.0, (ceiling - soc) if kind == "grid"
else (soc - floor)) / 100.0 * cap)
return kwh
blocks = []
if rp.charge_useful() and pm.mode in GRID_CHARGE_MODES:
blocks.append(("grid", rp.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:
blocks.append(("sell", sell))
else:
blocks.append(("give", g))
for kind, run in blocks:
if run.end <= now or not (run.start < day1 and run.end > day0):
continue
hrs = (run.end - run.start).total_seconds() / 3600.0
kwh = room(kind, hrs)
if kwh < MIN_KWH:
continue
price = run.mean_eur_kwh
if kind == "grid":
what, why, amount = (run_grid_what(kwh, shadow),
run_cheap_why(price),
run_grid_amount(kwh * price))
elif kind == "sell":
# No money line: the price is what the house pays, and a sale
# earns the lower injection price (OPT-11 P3 review).
what, why, amount = (run_sell_what(kwh, shadow),
run_dear_why(price), None)
else:
what, why, amount = (run_give_what(kwh, shadow),
run_dear_why(price),
run_give_amount(kwh * price))
out.append(_run(i, "give" if kind == "sell" else kind,
max(run.start, day0), min(run.end, day1), day0,
day1, what, why, amount, kwh=kwh,
eur=None if kind == "sell" else kwh * price,
price=price))
return out, None
def maintenance_runs(c, day0, day1, now, shadow):
"""The monthly battery cycle: under way now, or starting later today.
The cycle starts in the hour `maintenance_start_hour` names, read as the
@@ -434,10 +504,17 @@ def build_plan(c, now, shadow, running, held=False, settled=True):
runs, lines = [], []
# The control cycle trades on the fresh plan only, so this does too.
verdict, trade_why = trade_verdict(c, fresh)
modes_on = bool(getattr(c, "modes_on", False))
if running:
off = {i for i, inv in enumerate(c.inverters) if _off(c, inv)}
runs += house_runs(len(c.inverters), day0, day1, now, shadow, off)
if verdict is not None:
if modes_on:
# OPT-11 P3: with battery modes on, each battery's price runs.
mine, no_trade = mode_runs(c, day0, day1, now, shadow)
runs += mine
if no_trade:
lines.append(line_no_trade(no_trade))
elif verdict is not None:
trade, later, no_trade = trade_runs(c, fresh, verdict, day0, day1,
now, shadow)
runs += trade
@@ -544,6 +621,11 @@ def battery_rows(c, shadow, running, held=False):
pms = {inv.name: c.pack_mode(inv) for inv in c.inverters} \
if modes_on else {}
peak = peak_engaged(c)
# OPT-11 P3: each battery's role this cycle and the price runs, as the
# control cycle left them.
tick = getattr(c, "mode_tick", None) if modes_on else None
rp = getattr(c, "mode_runs", None)
now = datetime.now(timezone.utc)
rows = []
for i, inv in enumerate(c.inverters):
m = c.maints[i] if i < len(c.maints) else None
@@ -558,13 +640,22 @@ def battery_rows(c, shadow, running, held=False):
# carries its share (SAFETY-73)?
other = next((o.name for o in c.inverters if o is not inv
and refusing(c, o) > 0 and not _off(c, o)), None)
role = None if tick is None else tick.roles.get(inv.name)
# Only once a cycle has run with modes on (`tick`): before that
# nothing is known, so nothing is claimed about the prices.
plan = (None if tick is None or pm.mode not in PRICE_MODES
else "no_prices" if rp is None
else "no_spread" if not rp.on else None)
nxt = rp.next_give(now) if rp is not None and rp.on else None
at = "" if nxt is None else (
f"{hhmm(nxt.start)} - {hhmm(nxt.end)}")
why = battery_why(
modes_on=modes_on, mode=None if pm is None else pm.mode, off=off,
reading=inv.soc is not None and inv.batt is not None,
phase=phase, shadow=shadow, running=running, held=held,
refused=refusing(c, inv), peak=peak,
giving=flow(inv.batt) == "giving", other_refuses=other,
sf_ahead=sf_ahead)
sf_ahead=sf_ahead, role=role, plan=plan, has_at=bool(at))
rows.append({
"name": inv.name,
"level_pct": None if inv.soc is None else float(inv.soc),
@@ -582,7 +673,7 @@ def battery_rows(c, shadow, running, held=False):
# UX-13 P3: why it does what "now" says, from what the add-on
# knows this cycle. Price and forecast reasons are OPT-11 P3/P4.
"why_key": why,
"why": why_words(why, name=other),
"why": why_words(why, name=other, at=at),
})
if modes_on:
# OPT-11 P1: the battery's switch and mode, for the Batteries
@@ -590,13 +681,18 @@ def battery_rows(c, shadow, running, held=False):
# without them.
rows[-1].update({"use": pm.use, "mode_key": pm.key,
"mode_choice": pm.mode,
"mode": mode_word(pm.key)})
"mode": mode_word(pm.key),
# P3: selling, for Leftover solar and Cheap
# hours only (Solar first never sells).
"sell": pm.sell, "can_sell": pm.can_sell,
"role": role})
return rows
def battery_why(*, modes_on, mode, off, reading, phase, shadow, running,
held, refused, peak, sf_ahead, giving=False,
other_refuses=None) -> str:
other_refuses=None, role=None, plan=None,
has_at=False) -> str:
"""The key of one battery's "why" line (`home_words.why`), most urgent
first. Pure, so every branch is tested without a controller."""
if not reading:
@@ -620,18 +716,31 @@ def battery_why(*, modes_on, mode, off, reading, phase, shadow, running,
return "other_refuses"
if not modes_on:
return "modes_off"
# OPT-11 P3: the battery's role in the price runs this cycle.
if role in ROLE_WHY:
if role == "hold" and not has_at:
return "saves_later"
return ROLE_WHY[role]
if plan in ("no_prices", "no_spread"):
return plan
if giving:
return "covers_house" # both modes cover the house until P3
if mode == LEFTOVER_SOLAR:
return "covers_house"
if mode in PRICE_MODES:
return "leftover" if sf_ahead else "leftover_alone"
return "solar_first"
# OPT-11 P3: a role (modetrade) -> its Why key.
ROLE_WHY = {"charge": "grid_charge", "rest": "rests_charge",
"sell": "selling", "give": "expensive", "hold": "saves"}
def attach_today(rows, runs, *, modes_on, running=True, held=False,
shadow=False) -> None:
"""UX-13 P3: each battery's runs today from the plan (the same blocks as
Home's chart), or one line saying there are none. With battery modes on
and no run, the mode's own plan is OPT-11 P3/P4, so it says so."""
Home's chart), or one line saying there are none. With battery modes on,
a Leftover solar or Cheap hours battery with no price run left today
says so; the others follow the house."""
for i, row in enumerate(rows):
mine = [r for r in runs if r["battery"] == i and r["kind"] != "house"]
row["today"] = [{"from": r["from"], "to": r["to"], "kind": r["kind"],
@@ -644,8 +753,11 @@ def attach_today(rows, runs, *, modes_on, running=True, held=False,
row["today_line"] = today_words("stopped")
elif row.get("use") is False:
row["today_line"] = today_words("off")
elif modes_on:
row["today_line"] = today_words("not_planned")
elif modes_on and row.get("mode_choice") in PRICE_MODES:
# OPT-11 P3: a price mode with no run left today, or no plan
# made yet (no cycle has run with modes on).
row["today_line"] = today_words("no_runs" if row.get("role")
else "not_planned")
else:
row["today_line"] = today_words("house_would" if shadow
else "house")
@@ -954,10 +1066,11 @@ def battery_now(soc, batt_w, *, phase, shadow, running, held=False,
return _t("now.resting")
def why_words(key, name=None) -> str:
def why_words(key, name=None, at="") -> str:
"""One battery's "why" line (UX-13 P3), from `battery_why()`'s key.
`name`: the battery that refuses, for `other_refuses`."""
return _t("why." + key, name=name or "")
`name`: the battery that refuses, for `other_refuses`; `at`: the
give run's time, for `saves`."""
return _t("why." + key, name=name or "", at=at)
def today_words(key) -> str:
@@ -989,6 +1102,18 @@ def run_grid_amount(eur) -> str:
return _t("run.grid_amount", eur=_eur(eur))
def run_sell_what(kwh, shadow) -> str:
return _w(shadow, "run.sell_what", kwh=_kwh(kwh))
def run_cheap_why(price) -> str:
return _t("run.cheap_why", eur=_eur(price))
def run_dear_why(price) -> str:
return _t("run.dear_why", eur=_eur(price))
def run_give_what(kwh, shadow) -> str:
return _w(shadow, "run.give_what", kwh=_kwh(kwh))
+294 -12
View File
@@ -37,11 +37,13 @@ import aiohttp
from . import arbitrage
from .arbiter import (
P_LOOP, P_MAINT_SHAPE, P_MAINTENANCE, P_MODE, P_PEAK, P_SAFETY,
Claim, Resolution, actuator_for, pre_resolve, resolve,
ACTUATOR_SITE, P_LOOP, P_MAINT_SHAPE, P_MAINTENANCE, P_MODE, P_PEAK,
P_SAFETY, SET, Claim, Resolution, actuator_for, pre_resolve, resolve,
)
from . import allocator
from . import modetrade
from . import packmode
from . import priceruns
from . import control as _ctl
from .control import Tuning, compute, maintenance_charge_floor, peak_at_risk
from .delivery import (
@@ -80,7 +82,9 @@ from . import packshare
from . import pv as pv_read
from . import shadow
from .p1 import P1Ingest, build_source, is_enabled
from .price import Uncertainty, brussels, price_option, provider_configured
from .price import Absence, Uncertainty, brussels, price_option, provider_configured
from .price import required_spread as price_required_spread
from .price import uncertainty_of as price_uncertainty_of
from .price_history import PriceHistory
from .static_tariff import resolve_tariff, static_status
from .tariff import PeakStore, PeakTracker
@@ -226,6 +230,24 @@ def load_options() -> dict:
# OPT-11 P3: how long the last cheap run stays the price stored energy is
# valued at (`priceruns.plan(ref_eur_kwh=)`): the rest of that day.
REF_CHEAP_KEEP = timedelta(hours=12)
def why_raised(resolution, claims) -> bool:
"""F2 (OPT-11 P1 test): did the limit that bound the site target push it
ABOVE what the winner asked (the peak floor asking for more power)? Then
the Why line must not say "held back". False when nothing bound it."""
if not resolution.bound_by or resolution.contradiction:
return False
asked = next((c.value for c in sorted(
(c for c in claims if c.kind == SET
and c.actuator == ACTUATOR_SITE and c.source == resolution.winner),
key=lambda c: c.priority, reverse=True)), None)
return asked is not None and resolution.target_w > float(asked)
def pack_energy_of(inv):
"""OPT-09. ONE roster entry as the view `packshare` consumes.
@@ -909,6 +931,17 @@ class Controller:
self.modes_on = packmode.enabled(opts)
# The units whose mode fault is logged already (edge-logged).
self._mode_faults = set()
# OPT-11 P3: the price runs (`priceruns.RunPlan`, or None), the key
# they were built for, this tick's roles (`modetrade.Tick`), the
# batteries that left a run (full, or at the floor) keyed by the run,
# and whether last tick filed a mode trade (a run edge reseeds).
self.mode_runs = None
self._runs_key = None
self.mode_tick = None
self._run_done = {}
self._mode_trade_was = False
self._charge_blocked_q = None
self._ref_cheap = None
# OPT-11 P2: (topic, reason) of MQTT commands refused and logged once.
self._mqtt_refused = set()
self._mqtt_refused_more = 0
@@ -1982,7 +2015,11 @@ class Controller:
# OPT-11: the solar order. While the site charges, Leftover solar
# batteries weigh 0, so Solar first batteries fill first. A comfort
# choice like the rest of this method: permission is the mode claim.
return packmode.charge_weights(base, self.pack_modes(), target_w)
w = packmode.charge_weights(base, self.pack_modes(), target_w)
# P3: in a price run, the batteries whose run it is carry it first.
tick = self.mode_tick
return w if tick is None else modetrade.weights(tick.roles, w,
target_w)
def pack_mode(self, inv) -> "packmode.PackMode":
"""OPT-11. This battery's on/off switch and mode, read from the store."""
@@ -2011,6 +2048,214 @@ class Controller:
for u in self.units}
return allocator.loop_bound(bounds, live=self.live_units())
# --- OPT-11 P3: price runs, per battery -------------------------------
def _mode_packs(self, maint_results=None) -> list:
"""Each battery as `modetrade.Pack`, in roster order."""
live = set(self.live_units())
out = []
for i, inv in enumerate(self.inverters):
pm = self.pack_mode(inv)
idle = (maint_results is None
or maint_results[i].phase == IDLE)
refused = (inv.refused if len(self.inverters) > 1
else self.batt_refused) or 0
out.append(modetrade.Pack(
inv.name, pm.use, pm.mode, pm.sell and pm.can_sell,
float(inv.max_w), inv.soc, inv.capacity_kwh,
live=inv.name in live, idle=idle, refused=int(refused)))
return out
def _price_runs(self, now, packs, floor_pct, ceiling_pct):
"""The kept run plan, rebuilt at most once per price slot (15 min) or
when the prices or a battery's choice change; a run under way is
kept to its end (`priceruns.keep_started()`). None with no fresh
firm prices: then no battery holds or trades (design section 5)."""
plan = self.fresh_tariff_plan(now)
if plan is None or not plan.has_prices:
self.mode_runs, self._runs_key = None, None
return None
slot = int(now.timestamp() // 900)
key = (id(plan), slot, tuple((p.name, p.use, p.mode, p.sell)
for p in packs))
if key != self._runs_key:
done_c = self._run_done.get("charge", (None, set()))[1]
done_s = self._run_done.get("sell", (None, set()))[1]
e = modetrade.energy(packs, floor_pct, ceiling_pct, done_c,
done_s)
required = price_required_spread(
arbitrage.threshold_eur_kwh(self.o),
price_uncertainty_of(plan.read.product)
| {Uncertainty.NO_WEATHER})
ref = self._ref_cheap
if ref is not None and now - ref.end > REF_CHEAP_KEEP:
ref = self._ref_cheap = None
new = priceruns.plan(
plan.read.points, now, required_eur_kwh=required,
charge_kwh=e[0], charge_kw=e[1], sell_kwh=e[2],
sell_kw=e[3],
ref_eur_kwh=None if ref is None else ref.mean_eur_kwh)
self.mode_runs = priceruns.keep_started(self.mode_runs, new, now)
self._runs_key = key
runs = self.mode_runs
# The cheapest cheap run under way lately is the price the stored
# energy was bought at; it stays the reference for REF_CHEAP_KEEP
# after it ends, unless a cheaper one comes.
ref = self._ref_cheap
if runs is not None and runs.cheap is not None \
and runs.cheap.holds(now) and (
ref is None or now - ref.end > REF_CHEAP_KEEP
or runs.cheap.mean_eur_kwh <= ref.mean_eur_kwh):
self._ref_cheap = runs.cheap
return runs
def _mode_roles(self, now, maint_results):
"""OPT-11 P3: each battery's role and range this tick. Grid charging
and selling need a fresh plan that `TariffPlan.arbitrage()` allows
with the MEASURED peak guard (None refuses: nothing measures the
quarter), exactly the gate `arbitrage.decide()` has."""
floor_pct, ceiling_pct = arbitrage.soc_bounds(self.o)
packs = self._mode_packs(maint_results)
runs = self._price_runs(now, packs, floor_pct, ceiling_pct)
guard = arbitrage.guard_state(self.peak_guard)
plan = self.fresh_tariff_plan(now)
charge_ok = sell_ok = False
trade_why = "not_covered"
def allowed(active):
return plan.arbitrage(
peak_guard_active=active,
min_spread_eur_kwh=arbitrage.threshold_eur_kwh(self.o),
uncertainty=frozenset({Uncertainty.NO_WEATHER})).match(
allowed=lambda _v: True, refused=lambda _v: False)
if guard is not None and plan is not None and runs is not None:
# A charge asks the guard (section 9.1 interlock); a sell only
# needs it to have measured the quarter, since discharge is never
# capped by it (a big discharge alone engages it on export).
sell_ok = allowed(False)
charge_ok = sell_ok and allowed(guard)
trade_why = (None if charge_ok else "peak" if sell_ok
else "not_worth")
price_now = None
if plan is not None:
try:
pt = plan.read.match(series=lambda r: r.at(now),
no_data=lambda _r: None)
price_now = (None if pt is None or pt.provisional
else float(pt.eur_per_kwh))
except Absence:
price_now = None
house = (None if self.grid is None or self.site_batt_w is None
else float(self.grid) + float(self.site_batt_w))
done_c = self._run_mark("charge", runs and runs.cheap, packs,
lambda p: p.soc is not None
and p.soc >= ceiling_pct)
done_s = self._run_mark("sell", runs and runs.sell, packs,
lambda p: p.soc is not None
and p.soc <= floor_pct)
quarter = int(now.timestamp() // 900)
was = self.mode_tick is not None and \
self.mode_tick.action == modetrade.CHARGE
self.mode_tick = modetrade.decide(
runs, now, packs, charge_ok=charge_ok, sell_ok=sell_ok,
trade_why=trade_why, price_now=price_now,
headroom_w=self.arb_headroom_w, floor_pct=floor_pct,
ceiling_pct=ceiling_pct, house_w=house, charging_now=was,
blocked=self._charge_blocked_q == quarter,
done_charge=done_c, done_sell=done_s)
# A charge the guard or the headroom stopped stays off this period.
if was and self.mode_tick.action != modetrade.CHARGE and \
self.mode_tick.why in ("peak", "headroom"):
self._charge_blocked_q = quarter
return self.mode_tick
def _run_mark(self, kind, run, packs, out) -> set:
"""The batteries that left run `run` (full, or at the floor); they
stay out until it ends, so a level edge does not re-arm them."""
key = None if not run else (run.start, run.end)
was_key, done = self._run_done.get(kind, (None, set()))
if key != was_key:
done = set()
if key is not None:
done |= {p.name for p in packs if out(p)}
self._run_done[kind] = (key, done)
return done
def _mode_trade(self, claims, now) -> bool:
"""OPT-11 P3: file the site trade the roles ask for, sized inside the
modes and the hard limits (the loop bound) and, for a charge, the
billed quarter's headroom (SAFETY-45, aim cap - margin, SAFETY-79).
So no battery that its mode or a refusal holds is asked for watts it
may not give (P1's unmet share). Returns True when a trade is filed.
"""
tick = self.mode_tick
if tick is None:
return False
size = modetrade.size(tick, self.loop_bound(claims)
if tick.action else None, self.arb_headroom_w)
runs = self.mode_runs
if tick.action and (size == 0.0 or runs is None):
# Nothing the batteries may place (a refusal, a hard limit): no
# trade, and the roles say so rather than "charging".
back = {modetrade.CHARGE: modetrade.HOLD,
modetrade.REST: modetrade.HOUSE,
modetrade.SELL: modetrade.GIVE}
tick = self.mode_tick = modetrade.Tick(
tick.ranges, {n: back.get(r, r) for n, r in tick.roles.items()},
None, 0.0, "size")
if size == 0.0 or runs is None:
reason = ("battery modes: no price run now" if tick.why is None
else f"battery modes: no trade ({tick.why})")
self.arb = arbitrage.ArbitrageDecision(
action=None, power_w=None, reason=reason,
spread_eur_kwh=None, required_eur_kwh=None,
price_now_eur_kwh=None)
else:
charge = tick.action == modetrade.CHARGE
run = runs.cheap if charge else runs.sell
who = ", ".join(n for n, r in sorted(tick.roles.items())
if r == tick.action)
act = arbitrage.CHARGE if charge else arbitrage.DISCHARGE
self.arb = arbitrage.ArbitrageDecision(
action=act, power_w=size,
reason=(f"{act}: battery modes, "
f"{'cheap' if charge else 'sell'} run "
f"{run.start.isoformat()} to {run.end.isoformat()} "
f"at {run.mean_eur_kwh:.4f} EUR/kWh; "
f"{abs(size):.0f} W on {who}"),
spread_eur_kwh=None, required_eur_kwh=runs.level_eur_kwh,
price_now_eur_kwh=None)
claims.append(arbitrage.claim(self.arb))
if self.arb.action != self._arb_action:
self._arb_action = self.arb.action
self.log_event(self.arb.explain(), hint=messages.hint(
"trade_charge" if self.arb.action == arbitrage.CHARGE else
"trade_give" if self.arb.action == arbitrage.DISCHARGE
else "trade_off"))
return self.arb.action is not None
def _reseed_run_edge(self, claims) -> None:
"""OPT-11 P3: a mode trade just ended, so the loop takes over from a
command no loop chose. Start it at the house's own point E, the
command that brings the grid to its target now (`site battery + grid
- target`, inside the loop bound), instead of at the trade's watts
(the trade began with the same unslewed step). This tick's error still
measures the trade's command (it is exactly E - battery), so the
integrator is seeded that much short and lands on E after its step.
Measured (tools/opt11p3_replay.py, lagged battery, 2.5 kW charge
ending under an 800 W house): no reseed -258 W export then a 2976 W
import, settled in 55 s; this -257 W / +615 W, settled in 20 s.
"""
batt = self.site_batt_w
if batt is None or self.grid is None:
return
want = float(batt) + float(self.grid) - self.tuning.target_grid_w
bound = self.loop_bound(claims)
if bound is not None:
want = max(bound[0], min(bound[1], want))
want = float(round(max(-self.site_max_w, min(self.site_max_w, want))))
self.target = want
self.i_w = want - self.tuning.gain * (want - float(batt))
def _mode_fault_log(self, resolution) -> None:
"""OPT-11: say once, per battery, that its mode was dropped as a fault.
@@ -2032,7 +2277,8 @@ class Controller:
self._mode_faults = now
def _pack_action(self, what, unit, value, now) -> dict:
"""OPT-11: switch one battery on or off, or set its mode.
"""OPT-11: switch one battery on or off, set its mode, or (P3) turn
its selling on or off.
Store state, not an option, so it applies on the next cycle with no
restart. Written to the action log. Switching a battery off also stops
@@ -2056,6 +2302,15 @@ class Controller:
if not value:
self._maint_stop_off(self.maints[self.inverters.index(inv)],
now)
elif what == "pack_sell":
if not packmode.SELLING_ENABLED:
return {"error": messages.pack_refused("sell_off")}
if not isinstance(value, bool):
return {"error": messages.pack_refused("value")}
inv.store_set(packmode.SELL_KEY, value)
self.log_event(f"{inv.name} selling switched "
f"{'on' if value else 'off'} by operator",
hint=messages.hint("pack_sell", unit=inv.name))
else:
if value not in packmode.MODES:
return {"error": messages.pack_refused("value")}
@@ -2064,7 +2319,10 @@ class Controller:
f"by operator",
hint=messages.hint("pack_mode", unit=inv.name))
pm = self.pack_mode(inv)
return {"unit": inv.name, "use": pm.use, "mode": pm.mode}
out = {"unit": inv.name, "use": pm.use, "mode": pm.mode}
if what == "pack_sell":
out["sell"] = pm.sell
return out
def mode_values(self) -> dict:
"""OPT-11 P2: each battery's mode and switch, for their HA state
@@ -3579,15 +3837,20 @@ class Controller:
# decision presented as current.
self.arb, self.arb_at = None, now
self.arb_headroom_w = None
if arbitrage.enabled(self.o):
floor_pct, ceiling_pct = arbitrage.soc_bounds(self.o)
charge_w, discharge_w = self.trade_power_w(floor_pct, ceiling_pct)
# OPT-11 P3: with battery modes on, each battery's mode decides who
# trades (`_mode_roles()`, `_mode_trade()`, below, which read the
# same headroom), and this site-wide trade is not run.
site_trade = arbitrage.enabled(self.o) and not self.modes_on
if site_trade or self.modes_on:
# SAFETY-47. HOISTED OUT OF THE CALL, NOT COPIED. One evaluation of
# the shipped expression feeds both the sizing below and
# `arbitrage.sized_from_meter()` in `_shadow_settle()`, so the
# caveat on the recorded watts can never be answered from a
# different cycle's guard than the watts were.
self.arb_headroom_w = peakguard.charge_headroom_w(self.peak_guard)
if site_trade:
floor_pct, ceiling_pct = arbitrage.soc_bounds(self.o)
charge_w, discharge_w = self.trade_power_w(floor_pct, ceiling_pct)
self.arb = arbitrage.decide(
self.fresh_tariff_plan(now), now,
peak_guard_active=arbitrage.guard_state(self.peak_guard),
@@ -3767,12 +4030,17 @@ class Controller:
# apart from the hard limits, and it gives way to the peak guard's
# site floor (P_PEAK) only as far as the floor needs, so the two never
# meet in a `lo > hi` (arbiter._unit_modes()).
# P3: the range is the switch's, narrowed by the battery's role in
# the price runs this tick (`modetrade.decide()`); both hold 0 W.
if self.modes_on:
tick = self._mode_roles(now, maint_results)
for inv, result in zip(self.inverters, maint_results):
if result.phase != IDLE:
continue
pm = self.pack_mode(inv)
lo, hi = packmode.mode_range(pm, inv.max_w)
rlo, rhi = tick.ranges.get(inv.name, (lo, hi))
lo, hi = max(lo, rlo), min(hi, rhi)
claims.append(Claim.limit("mode", P_MODE, lo, hi, pm.key,
actuator=self.actuator_of("mode",
inv)))
@@ -3823,6 +4091,13 @@ class Controller:
_ctl.REFUSED,
actuator=self.actuator_of("refused",
inv)))
# --- OPT-11 P3: the site trade, after every unit claim --------------
# Sized from what the batteries may place with their modes and hard
# limits (the loop bound), so none is asked for watts it may not give.
mode_trade, trade_was = False, self._mode_trade_was
if self.modes_on:
mode_trade = self._mode_trade(claims, now)
self._mode_trade_was = mode_trade
if (self.auto or self.shadow) and self.inputs_ok:
# ⚠️ prev is the ARBITER's last output, not the loop's own last
# wish. If something outranked the loop, that is what the hardware
@@ -3840,6 +4115,9 @@ class Controller:
# stale period as power.
if self.loop_w is None or self.target != self.loop_w:
self.i_w = self.target
# OPT-11 P3: a run edge (a mode trade ended) reseeds the loop.
if trade_was and not mode_trade:
self._reseed_run_edge(claims)
# SAFETY-52: another writer overrode the loop, so a refusal learned
# against the loop's own ask no longer describes the wire. A gap
# (`loop_w` None: blind or stopped) keeps it: the stale-reading stop
@@ -4172,7 +4450,8 @@ class Controller:
else:
self.why_words = messages.why(
_why, resolution.target_w, resolution.bound_by,
contradiction=resolution.contradiction)
contradiction=resolution.contradiction,
raised=why_raised(resolution, claims))
# --- DEV-02: EACH INVERTER IS TOLD ITS OWN SHARE ---------------------
# ⚠️ ITS OWN SHARE, NEVER THE SITE TARGET. `resolution.shares` is the
# allocator's answer and `sum(shares) + unmet_w == round(target_w)` is
@@ -5934,8 +6213,11 @@ class Controller:
# instead of inferring it, and separates "trading is off" from "no
# decision was reached this cycle"; `cycle_at` and `arb_at` are what
# make the second one an observation rather than a guess.
# OPT-11 P3: with battery modes on the modes trade, whatever the
# switch says, so the row reports their decision.
("Arbitrage", arbitrage_row(
on=arbitrage.enabled(self.o), decision=self.arb,
on=arbitrage.enabled(self.o) or self.modes_on,
decision=self.arb,
decided_at=self.arb_at, cycle_at=self.cycle_at,
now=datetime.now(timezone.utc), why=self.why_words)),
]
@@ -5967,7 +6249,7 @@ class Controller:
async def handle_action(self, what: str, *, unit=None, value=None):
now = datetime.now(timezone.utc)
if what in ("pack_use", "pack_mode"):
if what in ("pack_use", "pack_mode", "pack_sell"):
return self._pack_action(what, unit, value, now)
if what == "auto_toggle":
self.auto = not self.auto
+11 -3
View File
@@ -839,17 +839,22 @@ WHY_LIMIT = {
}
def why(winner, target_w, bound_by=None, *, contradiction=False) -> str:
def why(winner, target_w, bound_by=None, *, contradiction=False,
raised=False) -> str:
"""`winner` is the arbiter's winning source, or its reason for the two
that share a source ("stopped", "inputs missing", "no claim was made").
`bound_by` is `Resolution.bound_by`, e.g. "peak(quarter peak)"."""
`bound_by` is `Resolution.bound_by`, e.g. "peak(quarter peak)".
`raised` is True when that limit pushed the power UP (the peak limit
asking for more): then it did not hold anything back (OPT-11 P1 test F2).
"""
if contradiction:
return "Two limits clash, so every inverter is set to 0 W"
out = f"{WHY_WINNER.get(winner, 'Other')}: {signed_watts(target_w)}"
if bound_by:
limit = WHY_LIMIT.get(bound_by) or WHY_LIMIT.get(
str(bound_by).split("(")[0], "another limit")
out += f", held back by {limit}"
out += f", raised for {limit}" if raised else \
f", held back by {limit}"
return out
@@ -922,6 +927,7 @@ HINTS = {
"pack_on": "{unit}: battery switched on",
"pack_off": "{unit}: battery switched off, it is asked for 0 W",
"pack_mode": "{unit}: battery mode changed",
"pack_sell": "{unit}: selling switched",
"maint_off": "{unit}: maintenance stopped, the battery is switched off",
"mode_fault": "{unit}: battery mode ignored because of a software fault",
}
@@ -932,6 +938,8 @@ PACK_REFUSED = {
"off": "Battery modes are off. Turn on “Battery modes” first.",
"unit": "There is no battery with that name.",
"value": "That choice is not possible for this battery.",
# OPT-11 P3, owner decision 2026-09-29: selling is off for now.
"sell_off": "Selling is off for now.",
# OPT-11 P2: a Home Assistant command dropped before it reaches a battery.
"retained": "An old saved command is not used.",
"size": "That command is too long.",
+231
View File
@@ -0,0 +1,231 @@
"""OPT-11 P3: each battery's role this tick, from its mode and the price runs.
Pure: no I/O, no clock. main.py hands in the run plan (`priceruns`), each
battery's state and the peak guard's answers, files each battery's range as
its `mode` claim (P_MODE), and files the site trade as the `arbitrage` claim
(P_ARBITRAGE), sized inside the modes (`size()`), only with battery modes on.
With them off, `arbitrage.decide()` runs exactly as before.
Roles (the page's Why line names them):
off Off: [0, 0].
house follows the house: charges from spare solar, gives power at any
hour (Solar first always; the other modes when there is no plan).
hold Leftover solar / Cheap hours outside a give run: may charge from
spare solar, gives nothing, so its energy is kept for the run.
give the same inside a give run: covers the house.
sell inside the sell run, with selling on: gives its rating, to the
house first and the rest to the grid.
charge inside the cheap run: charges from the grid.
rest another battery charges from the grid, so this one may not charge
(the owner's 2026-09-27 answer 3: while one battery charges from the
grid the others cannot give power either; the allocator never
splits a target both ways).
Rules that keep this safe (tests: test_modes_p3.py):
* every range holds 0 W (a mode may forbid a direction, never force one);
* a Solar first battery never charges from the grid (no forecast yet, P4);
* grid charging needs a measured peak guard that is clear and a measured
headroom (SAFETY-45); the trade is sized to that headroom, whose aim is
`peak_cap_w - peak_margin_w` (SAFETY-79). It starts only at START_W or
more, and once the guard or the headroom stops it, it stays stopped for
the rest of that 15-minute period (`blocked`): a charge makes the other
batteries stop giving power, so a few watts of charge flipping on and
off would swing the house between them and the grid (measured on the
2026-09-09 replay: 13 flips in 4 minutes at a 3.5 kW house);
* selling needs selling on for that battery, the sell run, a price >= 0
now, and the battery above `arbitrage_soc_floor`; it stops there. The
peak guard never stops a sell (section 9.1: discharge is never capped);
it must only have measured the quarter;
* a battery out of a run (full, or at the floor) stays out until the run
ends, so the ceiling and the floor do not re-arm within one run;
* a battery that refuses a direction (SAFETY-52/73) takes no part in a
run in that direction: the refusal wins over the run.
"""
from dataclasses import dataclass
from .packmode import (CHEAP_HOURS, GRID_CHARGE_MODES, LEFTOVER_SOLAR,
SOLAR_FIRST)
OFF, HOUSE, HOLD, GIVE, SELL, CHARGE, REST = (
"off", "house", "hold", "give", "sell", "charge", "rest")
PRICE_MODES = (LEFTOVER_SOLAR, CHEAP_HOURS)
# The smallest grid charge worth starting, in W. Under it the other batteries
# would stop covering the house for less than they give.
START_W = 500.0
@dataclass(frozen=True)
class Pack:
"""One battery as this module needs it."""
name: str
use: bool
mode: str
sell: bool
rating_w: float
soc: float | None
capacity_kwh: float | None
live: bool = True
idle: bool = True # not in a maintenance phase
refused: int = 0 # SAFETY-52/73: +1 refuses to give, -1 to charge
@dataclass(frozen=True)
class Tick:
ranges: dict # name -> (lo, hi), only for idle batteries
roles: dict # name -> role
action: str | None # "charge" | "sell" | None
ask_w: float # signed site ask before sizing (0 if none)
why: str | None # why no trade when a run is on, else None
def _room_kwh(p: Pack, pct: float) -> float:
"""kWh to `pct` (charging) or above it (selling); a battery with no
capacity set counts one hour at its rating, so a run is still sized."""
if p.capacity_kwh is None or p.soc is None:
return abs(p.rating_w) / 1000.0
return max(0.0, abs(pct - p.soc)) / 100.0 * float(p.capacity_kwh)
def chargers(packs, ceiling_pct, done=()) -> list:
"""Batteries that may charge from the grid in a cheap run: the modes in
`packmode.GRID_CHARGE_MODES` (Cheap hours only until P4's forecast)."""
return [p for p in packs
if p.use and p.idle and p.live and p.mode in GRID_CHARGE_MODES
and p.soc is not None and p.soc < ceiling_pct
and p.refused >= 0 and p.name not in done]
def sellers(packs, floor_pct, done=()) -> list:
"""Batteries that may sell in the sell run. Solar first never sells."""
return [p for p in packs
if p.use and p.idle and p.live and p.sell
and p.mode in PRICE_MODES
and p.soc is not None and p.soc > floor_pct
and p.refused <= 0 and p.name not in done]
def energy(packs, floor_pct, ceiling_pct, done_charge=(), done_sell=()):
"""`(charge_kwh, charge_kw, sell_kwh, sell_kw)` for `priceruns.plan()`."""
ch = chargers(packs, ceiling_pct, done_charge)
se = sellers(packs, floor_pct, done_sell)
return (sum(_room_kwh(p, ceiling_pct) for p in ch),
sum(abs(p.rating_w) for p in ch) / 1000.0,
sum(_room_kwh(p, floor_pct) for p in se),
sum(abs(p.rating_w) for p in se) / 1000.0)
def decide(runs, now, packs, *, charge_ok, sell_ok, trade_why, price_now,
headroom_w, floor_pct, ceiling_pct, house_w, charging_now=False,
blocked=False, done_charge=(), done_sell=()) -> Tick:
"""This tick's ranges, roles and site ask.
`runs` is the kept `priceruns.RunPlan` or None (no fresh prices).
`charge_ok` is True only when `TariffPlan.arbitrage()` allowed a trade
this tick with the MEASURED peak guard clear; `sell_ok` when it allowed
one with the guard measured (either state); else `trade_why` says why.
`headroom_w` is `peakguard.charge_headroom_w()` (None: unmeasured).
`house_w` is the measured house (grid + batteries), or None.
`charging_now` says last tick charged (then any headroom above 0 keeps
it going); `blocked` that the guard or the headroom stopped a charge in
this 15-minute period.
"""
plan_on = runs is not None and runs.on
in_cheap = plan_on and runs.cheap.holds(now) and runs.charge_useful()
in_give = plan_on and runs.give_at(now) is not None
in_sell = plan_on and runs.sell is not None and runs.sell.holds(now)
why = None
ch = chargers(packs, ceiling_pct, done_charge) if in_cheap else []
charging = False
if ch:
need = 0.0 if charging_now else START_W
if blocked:
why = "headroom"
elif not charge_ok:
why = trade_why
elif headroom_w is None or not headroom_w > need:
why = "headroom"
else:
charging = True
se = sellers(packs, floor_pct, done_sell) if in_sell and not charging \
else []
selling = False
if se:
if not sell_ok:
why = why or trade_why
elif price_now is None or price_now < 0.0 or house_w is None:
why = why or "negative"
else:
selling = True
ranges, roles = {}, {}
ch_names = {p.name for p in ch} if charging else set()
se_names = {p.name for p in se} if selling else set()
for p in packs:
r = abs(float(p.rating_w))
if not p.use:
role, lo, hi = OFF, 0.0, 0.0
elif p.name in ch_names:
role, lo, hi = CHARGE, -r, 0.0
elif charging:
# Another battery charges from the grid: this one may not charge
# (it would be grid charging too), and cannot give (one sign).
role, lo, hi = REST, 0.0, (r if p.mode == SOLAR_FIRST
or not plan_on or in_give else 0.0)
elif p.name in se_names:
role, lo, hi = SELL, -r, r
elif p.mode == SOLAR_FIRST or not plan_on:
role, lo, hi = HOUSE, -r, r
elif in_give:
role, lo, hi = GIVE, -r, r
else:
role, lo, hi = HOLD, -r, 0.0
roles[p.name] = role
if p.idle:
ranges[p.name] = (lo, hi)
if charging:
ask = -sum(abs(p.rating_w) for p in ch)
return Tick(ranges, roles, CHARGE, ask, None)
if selling:
ask = sum(abs(p.rating_w) for p in se) + max(0.0, float(house_w))
return Tick(ranges, roles, SELL, ask, None)
return Tick(ranges, roles, None, 0.0,
why if (in_cheap or in_sell) else None)
def size(tick: Tick, bound, headroom_w) -> float:
"""The site trade in W, inside what the batteries may place with their
modes (`bound` = main's loop bound) and, for a charge, the billed
quarter's headroom. 0.0 means no trade (a 0 W trade is a park order and
is never filed)."""
if tick.action is None or bound is None:
return 0.0
lo, hi = bound
if tick.action == CHARGE:
cap = min(-float(tick.ask_w), -float(lo),
float(headroom_w) if headroom_w is not None else 0.0)
return -max(0.0, cap)
return max(0.0, min(float(tick.ask_w), float(hi)))
def weights(roles: dict, base, target_w: float):
"""Who carries the site target first (design 3.5). Selling: the sellers.
Giving power: batteries in a give run before Solar first. Charging from
the grid: only the chargers. Otherwise `base` unchanged. Comfort only;
the ranges above are the permission."""
names = sorted(roles)
first = None
if target_w > 0 and any(r == SELL for r in roles.values()):
first = {n for n in names if roles[n] == SELL}
elif target_w > 0 and any(r == GIVE for r in roles.values()):
first = {n for n in names if roles[n] in (GIVE, SELL)}
elif target_w < 0 and any(r == CHARGE for r in roles.values()):
first = {n for n in names if roles[n] == CHARGE}
if not first:
return base
w = {n: (1.0 if base is None else float(base.get(n, 1.0))) for n in names}
out = {n: (w[n] if n in first else 0.0) for n in names}
return out if any(v > 0.0 for v in out.values()) else base
+2 -1
View File
@@ -453,7 +453,8 @@ def discovery_payloads(units=(), omit=(), include=(), modes=()) -> list:
#
# The select's options are these labels, in `packmode.MODES` order. A new mode
# (P3) is a new entry here and a new option at the end; no id changes.
MODE_LABELS = {"solar_first": "Solar first", "leftover_solar": "Leftover solar"}
MODE_LABELS = {"solar_first": "Solar first", "leftover_solar": "Leftover solar",
"cheap_hours": "Cheap hours only"} # OPT-11 P3
USE_ON, USE_OFF = "ON", "OFF"
# A longer command payload is refused before it is decoded.
MAX_COMMAND_BYTES = 64
+39 -22
View File
@@ -1,4 +1,4 @@
"""OPT-11 P1: each battery's mode and on/off switch, as limits and weights.
"""OPT-11: each battery's mode and on/off switch, as limits and weights.
Pure: no I/O, no clock. main.py reads the store, calls these, and files the
result as a `mode` claim (arbiter.P_MODE) and as the charge weights.
@@ -14,13 +14,11 @@ Two rules hold for every mode, and they are what keeps modes safe:
allocator._fill()'s last-resort pass still uses it. So "may not" is a range,
and weights only set the order among batteries that may.
P1 has no price runs and no forecast (P3, P4). So both modes give power to the
house at any hour, and the only mode that closes a direction is Off. What P1
changes is the solar order: while the site charges, Leftover solar batteries
weigh 0, so they get solar only once the Solar first batteries are full or at
their power limit. Selling (the owner's 2026-09-27 answer) is a per-battery
choice that arrives with the price runs in P3; `PackMode.sell` and the
`pack_sell` store key are reserved for it and read by nothing yet.
`mode_range()` is the switch alone (Off is [0, 0]). P3 adds the price runs:
`modetrade.decide()` narrows each battery's range by its role this tick
(hold for the expensive hours, charge from the grid, sell), and main.py files
the intersection. Selling is a per-battery choice (`pack_sell`, the owner's
2026-09-27 answer), for Leftover solar and Cheap hours only.
"""
from dataclasses import dataclass
@@ -29,17 +27,29 @@ from .store import unit_key
SOLAR_FIRST = "solar_first"
LEFTOVER_SOLAR = "leftover_solar"
CHEAP_HOURS = "cheap_hours" # reserved for P3; not accepted in P1
CHEAP_HOURS = "cheap_hours" # OPT-11 P3
# The modes a battery may be set to in this release, in the order the page
# lists them. The first is the default when the store has no value.
MODES = (SOLAR_FIRST, LEFTOVER_SOLAR)
# The modes a battery may be set to, in the order the page and the HA select
# list them. The first is the default when the store has no value. A new mode
# is appended, never inserted: automations send the HA labels (P2).
MODES = (SOLAR_FIRST, LEFTOVER_SOLAR, CHEAP_HOURS)
DEFAULT_MODE = SOLAR_FIRST
# Store keys, per battery: `unit/<name>/<key>` (store.unit_key()).
USE_KEY = "pack_use"
MODE_KEY = "pack_mode"
SELL_KEY = "pack_sell" # reserved for P3
SELL_KEY = "pack_sell" # OPT-11 P3: sell in the sell run
# Owner decision 2026-09-29 (option A): selling is off until it can be judged
# on the price the grid pays. While False, `read()` answers `sell=False`
# whatever the store holds, the page shows no switch and `pack_sell` is
# refused; the sell code in modetrade/priceruns stays, unreachable.
SELLING_ENABLED = False
# The modes that may charge from the grid in a cheap run. Leftover solar joins
# when the solar forecast exists (P4); until then it takes solar only (owner
# decision 2026-09-29, option A).
GRID_CHARGE_MODES = (CHEAP_HOURS,)
def enabled(opts) -> bool:
@@ -50,10 +60,16 @@ def enabled(opts) -> bool:
@dataclass(frozen=True)
class PackMode:
"""One battery's choice: on or off, its mode, and (P3) selling."""
"""One battery's choice: on or off, its mode, and selling."""
use: bool = True
mode: str = DEFAULT_MODE
sell: bool = False # reserved for P3, never read in P1
sell: bool = False
@property
def can_sell(self) -> bool:
"""Selling applies to the price modes only; Solar first never sells.
False for every battery while SELLING_ENABLED is False."""
return SELLING_ENABLED and self.mode in (LEFTOVER_SOLAR, CHEAP_HOURS)
@property
def key(self) -> str:
@@ -69,15 +85,15 @@ def read(data: dict, name: str) -> PackMode:
mode = data.get(unit_key(name, MODE_KEY), DEFAULT_MODE)
return PackMode(use=use is not False,
mode=mode if mode in MODES else DEFAULT_MODE,
sell=data.get(unit_key(name, SELL_KEY)) is True)
sell=SELLING_ENABLED
and data.get(unit_key(name, SELL_KEY)) is True)
def mode_range(pm: PackMode, rating_w: float) -> tuple:
"""`(lo, hi)` in W for this battery's mode claim. Always holds 0 W.
"""`(lo, hi)` in W for this battery's switch. Always holds 0 W.
Off asks for 0 W both ways. In P1 both modes may charge and give power up
to the battery's rating: Leftover solar holds back its power only in the
planned expensive hours, and there are none before P3.
Off asks for 0 W both ways; On allows the rating both ways. main.py files
this intersected with the role range from `modetrade.decide()`.
"""
if not pm.use:
return 0.0, 0.0
@@ -92,7 +108,7 @@ def charge_weights(base, modes: dict, target_w: float):
charges (`target_w < 0`), a Leftover solar battery weighs 0 and a Solar
first battery keeps its weight, so the fill gives the Solar first batteries
everything they can take before the last-resort pass reaches the others.
Giving power keeps `base` (P3 adds the expensive-run order).
Giving power keeps `base`; `modetrade.weights()` adds the run order.
When no battery that is on would keep a positive weight (all Leftover
solar, or every Solar first one at 0), `base` is returned unchanged, so the
@@ -102,7 +118,8 @@ def charge_weights(base, modes: dict, target_w: float):
return base
names = sorted(modes)
w = {n: (1.0 if base is None else float(base.get(n, 1.0))) for n in names}
out = {n: (0.0 if modes[n].mode == LEFTOVER_SOLAR else w[n])
out = {n: (0.0 if modes[n].mode in (LEFTOVER_SOLAR, CHEAP_HOURS)
else w[n])
for n in names}
if not any(out[n] > 0.0 for n in names if modes[n].use):
return base
+203
View File
@@ -0,0 +1,203 @@
"""OPT-11 P3: price runs. Which consecutive firm price slots are cheap, which
are expensive, and which one a battery may sell in.
Pure: prices, the time and a few energy figures in, a `RunPlan` out. No I/O,
no clock. `modetrade.py` turns the plan into each battery's range and the
site's trade; main.py rebuilds the plan at most once per 15 minutes and keeps
a run that has started until its end (`keep_started()`).
The owner's SAFETY-46 answer (2026-09-25): a trade is a run of consecutive
slots, not one 15-minute interval. OPT-11-design section 2 ("Prices") and
DECISIONS/OPT-11-battery-modes.md have the record.
* The CHEAP run is the cheapest consecutive firm window long enough to put
the planned energy in at the chargers' combined rating.
* A slot is EXPENSIVE when its price is above the cheap run's mean by at
least the required spread (`cycle_cost_threshold_eur`, widened as
`TariffPlan.arbitrage()` widens it). Give runs are the maximal consecutive
expensive slots. Holding batteries cover the house in them. The mean is
the lower of the next cheap run's and `ref_eur_kwh`, the cheap run the
batteries last filled in (main.py keeps it for 12 h): energy bought at
03:00 is still worth that price at 08:00, when the next cheap run in sight
is today's dearer midday (measured: without it a Cheap hours battery spent
its night charge on flat morning prices, then had nothing at 19:00).
* The SELL run is the dearest consecutive window inside one give run long
enough to sell the sellers' energy at their rating, with no slot below 0
(section 17 of the spec: never export at a negative price).
* A provisional (carried) price is never part of any run (section 8.3).
"""
import math
from dataclasses import dataclass
from datetime import datetime, timedelta
from .price import _cover_table
# How far ahead runs are looked for. A day: tonight's cheap hours and the next
# evening's expensive ones, without holding a battery for a peak two days out.
HORIZON = timedelta(hours=24)
# No plan, and why (the page words are in homeview / ui_text).
NO_PRICES = "no_prices" # fewer than two firm slots ahead
NOT_WORTH = "not_worth" # no slot clears the required spread
@dataclass(frozen=True)
class Run:
start: datetime
end: datetime
mean_eur_kwh: float
def holds(self, when: datetime) -> bool:
return self.start <= when < self.end
@dataclass(frozen=True)
class RunPlan:
"""What the prices say for the next day. `reason` is None when there is
a plan (a cheap run and at least one give run), else why there is none."""
cheap: Run | None
give: tuple
sell: Run | None
level_eur_kwh: float | None
reason: str | None
# A slot after the cheap run clears the spread over the CHEAP RUN'S OWN
# price (not the reference): only then is buying in it worth a cycle.
charge_worth: bool = False
@property
def on(self) -> bool:
return self.reason is None
def give_at(self, when):
return next((g for g in self.give if g.holds(when)), None)
def next_give(self, when):
"""The give run under way at `when`, else the next one, else None."""
return next((g for g in self.give if g.end > when), None)
def charge_useful(self) -> bool:
"""Charging in the cheap run pays only if, after it, a slot clears
the spread over the cheap run's own price."""
return self.cheap is not None and self.charge_worth
def firm_slots(points, now: datetime, horizon: timedelta = HORIZON) -> list:
"""`[(start, end, eur_kwh)]`, ascending: the firm slots that have not
ended and start within the horizon. A slot's span is TARIFF-14's rule
(`price._cover_table()`), so 15-minute and hourly series both work."""
out = []
for start, end, p in _cover_table(points):
if p.provisional or end <= now or start >= now + horizon:
continue
if end > start:
out.append((start, end, float(p.eur_per_kwh)))
return out
def _segments(slots) -> list:
"""Split slots into runs of consecutive ones (each starts where the last
ended). A gap or a provisional slot breaks a run."""
segs, cur = [], []
for s in slots:
if cur and s[0] != cur[-1][1]:
segs.append(cur)
cur = []
cur.append(s)
if cur:
segs.append(cur)
return segs
def _hours(s) -> float:
return (s[1] - s[0]).total_seconds() / 3600.0
def _mean(win) -> float:
h = sum(_hours(s) for s in win)
return sum(_hours(s) * s[2] for s in win) / h
def _windows(segs, need_h):
"""Every consecutive window that lasts at least `need_h` hours, shortest
such from each start. If no segment is that long, each whole segment."""
out = []
for seg in segs:
for i in range(len(seg)):
h, j = 0.0, i
while j < len(seg) and h < need_h - 1e-9:
h += _hours(seg[j])
j += 1
if h >= need_h - 1e-9:
out.append(seg[i:j])
return out or [seg for seg in segs]
def need_hours(kwh, kw, slots) -> float:
"""How long a run must last: `kwh` at `kw`, and at least one slot."""
one = min((_hours(s) for s in slots), default=1.0)
if not (kwh and kw) or kwh <= 0 or kw <= 0:
return one
return max(one, float(kwh) / float(kw))
def plan(points, now: datetime, *, required_eur_kwh: float, charge_kwh,
charge_kw, sell_kwh, sell_kw, horizon: timedelta = HORIZON,
ref_eur_kwh=None) -> RunPlan:
"""The run plan for the next `horizon`. Ties go to the earliest window."""
if not (isinstance(required_eur_kwh, (int, float))
and math.isfinite(required_eur_kwh) and required_eur_kwh >= 0):
raise ValueError("required_eur_kwh must be a finite number >= 0")
slots = firm_slots(points, now, horizon)
if len(slots) < 2:
return RunPlan(None, (), None, None, NO_PRICES)
segs = _segments(slots)
wins = _windows(segs, need_hours(charge_kwh, charge_kw, slots))
best = min(wins, key=lambda w: (round(_mean(w), 9), w[0][0]))
cheap = Run(best[0][0], best[-1][1], _mean(best))
base = cheap.mean_eur_kwh if ref_eur_kwh is None else min(
cheap.mean_eur_kwh, float(ref_eur_kwh))
level = base + float(required_eur_kwh)
# Expensive: clears the spread, strictly dearer than the cheap mean, and
# not part of the cheap run itself (a threshold of 0 must not overlap).
dear = [s for s in slots
if s[2] >= level and s[2] > cheap.mean_eur_kwh
and not (cheap.start <= s[0] < cheap.end)]
give_segs = _segments(dear)
give = tuple(Run(g[0][0], g[-1][1], _mean(g)) for g in give_segs)
worth = any(s[0] >= cheap.end and s[2] >= cheap.mean_eur_kwh
+ float(required_eur_kwh) for s in dear)
if not give:
return RunPlan(cheap, (), None, level, NOT_WORTH)
# Sell: inside one give run, never over a slot priced below 0.
sellable = [part for g in give_segs
for part in _segments([s for s in g if s[2] >= 0.0])]
sell = None
if sellable:
sw = _windows(sellable, need_hours(sell_kwh, sell_kw, slots))
top = max(sw, key=lambda w: (round(_mean(w), 9),
-w[0][0].timestamp()))
sell = Run(top[0][0], top[-1][1], _mean(top))
return RunPlan(cheap, give, sell, level, None, worth)
def keep_started(old: RunPlan | None, new: RunPlan, now: datetime) -> RunPlan:
"""A run under way keeps going to its end (no flapping on a new plan): the
cheap run, the give run and the sell run each come from `old` while `old`
holds `now` in it. The rest is `new`. A safety or peak rule can still stop
a run at any tick; that is not decided here."""
if old is None or not old.on:
return new
cheap = old.cheap if old.cheap and old.cheap.holds(now) else new.cheap
sell = old.sell if old.sell and old.sell.holds(now) else new.sell
g_old = old.give_at(now)
give = new.give
if g_old is not None:
give = (g_old,) + tuple(g for g in new.give
if g.start >= g_old.end)
if cheap is new.cheap and sell is new.sell and g_old is None:
return new
worth = old.charge_worth if cheap is old.cheap else new.charge_worth
return RunPlan(cheap, give, sell, new.level_eur_kwh if new.on
else old.level_eur_kwh, None if give else new.reason,
worth)
+5 -14
View File
@@ -314,11 +314,11 @@
/* ---------------- Batteries (UX-13 P3) ---------------- */
// One card per battery. With battery modes on (`mode_key` in /status) each
// card has an On/Off switch and a mode list, sent as `pack_use` and
// `pack_mode` through /action (OPT-11 P1). Selling and cheap-hours charging
// are shown greyed out until OPT-11 P3. The Now, Today and Why lines are
// `pack_mode` through /action (OPT-11 P1). Selling has no switch while it
// is off (owner decision 2026-09-29). The Now, Today and Why lines are
// homeview.py's sentences. The Simple settings view draws the same switch
// and list into its own card (simpleHost).
const MODES = ["solar_first", "leftover_solar"];
const MODES = ["solar_first", "leftover_solar", "cheap_hours"];
const say = (text, vars) => Object.keys(vars).reduce((x, k) => x.replace("{" + k + "}", vars[k]), text);
let packBusy = false;
let simpleHost = null;
@@ -345,15 +345,7 @@
function modeSelect(b, i, where, lid) {
return h("select", { class: "sel", id: "bsel-" + where + i, "aria-labelledby": lid, disabled: !b.use,
onchange: (e) => packAct("pack_mode", b, e.target.value) },
...MODES.map((k) => h("option", { value: k, selected: b.mode_choice === k }, t("batteries.choice." + k + ".name"))),
h("option", { value: "", disabled: true }, t("batteries.later")));
}
function sellRow(i, where) {
const lid = "bsl-" + where + i;
return h("div", { class: "f inline" },
h("div", {}, h("div", { class: "lbl", id: lid }, t("batteries.sell.name")), h("div", { class: "help" }, t("batteries.sell.description"))),
h("div", { class: "ctl" }, h("button", { type: "button", class: "switch", role: "switch", "aria-checked": "false",
"aria-labelledby": lid, disabled: true })));
...MODES.map((k) => h("option", { value: k, selected: b.mode_choice === k }, t("batteries.choice." + k + ".name"))));
}
function todayLines(b) {
if (!b.today || !b.today.length) return b.today_line || "–";
@@ -372,8 +364,7 @@
modes ? h("div", { class: "modebox" },
h("label", { class: "lbl small", for: "bsel-c" + i }, t("batteries.mode")),
modeSelect(b, i, "c", null),
h("div", { class: "help" }, t("batteries.choice." + b.mode_key + ".description")),
sellRow(i, "c")) : null,
h("div", { class: "help" }, t("batteries.choice." + b.mode_key + ".description"))) : null,
levelBar(b.level_pct, b.name),
h("dl", { class: "kv" },
h("dt", {}, t("batteries.level")), h("dd", {}, lvl + stored),
+21 -8
View File
@@ -87,18 +87,18 @@
"on": "On", "off": "Off",
"choice": {
"solar_first": {"name": "Solar first", "description": "Charges from solar before the other batteries and covers the house."},
"leftover_solar": {"name": "Leftover solar", "description": "Charges from solar the other batteries cannot take, and covers the house."},
"leftover_solar": {"name": "Leftover solar", "description": "Gets solar the others cannot take, and gives power in expensive hours."},
"cheap_hours": {"name": "Cheap hours only", "description": "Charges in the cheapest hours and gives power in the expensive ones."},
"off": {"name": "Off", "description": "Rests, and helps only when the peak limit needs it."}
},
"later": "Cheap hours only, coming later",
"today": "Today", "why": "Why",
"sell": {"name": "Sell in expensive hours", "description": "Selling comes in a later update."},
"modes_off": {"name": "Battery modes are off"},
"how": {"name": "How the modes work", "description": "Which battery gets solar first, and when each gives power.",
"solar_first": {"description": "Solar first gets spare solar before the other batteries, and covers the house at any hour."},
"leftover_solar": {"description": "Leftover solar gets the solar the Solar first batteries cannot take, and also covers the house."},
"leftover_solar": {"description": "Leftover solar gets the solar the Solar first batteries cannot take, and saves its power for expensive hours."},
"cheap_hours": {"description": "Cheap hours only charges in the cheapest hours and gives power in the expensive ones."},
"off": {"description": "Off rests the battery and starts no monthly cycle, but the peak limit may still use it."},
"rules": {"description": "The peak limit and the safety stops always come first, and charging from the grid and selling come later."}},
"rules": {"description": "The peak limit and the safety stops come first, and while one battery charges from the grid the others rest."}},
"toast": {"on": "{name} is on.", "off": "{name} is off.", "mode": "{name}: {mode}."}},
"settings": {"name": "Settings", "description": "The settings are grouped by task below.", "open": "Open the settings page"}
},
@@ -209,7 +209,7 @@
"5": {"name": "Peak limit", "description": "The grid stays under this limit, averaged over each 15-minute period."},
"6": {"name": "Check and start", "description": "The add-on checks everything with its own start checks, then restarts."}
},
"modes_later": {"description": "Choosing a mode for each battery comes in a later update."},
"modes_later": {"description": "After setup, turn on battery modes to choose a mode for each battery on the Batteries screen."},
"kettle": {
"name": "Check the direction",
"description": "Read the meter, switch on a kettle, then read the meter again.",
@@ -250,7 +250,7 @@
},
"home_words": {
"mode_house": "Follows the house",
"modes": {"solar_first": "Solar first", "leftover_solar": "Leftover solar", "off": "Off"},
"modes": {"solar_first": "Solar first", "leftover_solar": "Leftover solar", "cheap_hours": "Cheap hours only", "off": "Off"},
"attn_shadow": "Shadow mode is on",
"attn_shadow_body": "The add-on decides everything, but every inverter stays at 0 W.",
"zero_held": "Held at 0 W",
@@ -326,10 +326,19 @@
"leftover": "It gets the solar that Solar first batteries cannot take.",
"leftover_alone": "No Solar first battery is on, so it gets solar first.",
"covers_house": "It gives power to cover the house.",
"other_refuses": "{name} refuses to give power, so this battery takes over."
"other_refuses": "{name} refuses to give power, so this battery takes over.",
"grid_charge": "Prices are low, so it charges from the grid.",
"rests_charge": "Another battery charges from the grid, so it rests.",
"selling": "Prices are high, so it sells to the grid.",
"expensive": "Prices are high, so it covers the house.",
"saves": "Saves its power for {at}.",
"saves_later": "Saves its power for higher prices.",
"no_prices": "No prices, so it follows the house.",
"no_spread": "Prices differ too little, so it follows the house."
},
"today": {
"not_planned": "Not planned yet.",
"no_runs": "No cheap or expensive hours left today.",
"off": "Off, so no monthly cycle today.",
"house": "Follows the house.",
"house_would": "Would follow the house.",
@@ -356,6 +365,10 @@
"give_what": "Gives up to {kwh}, to the house first.",
"give_what_would": "Would give up to {kwh}, to the house first.",
"give_why": "Highest published price, {eur} per kWh.",
"sell_what": "Sells up to {kwh}, to the house first.",
"sell_what_would": "Would sell up to {kwh}, to the house first.",
"cheap_why": "Cheapest hours, about {eur} per kWh.",
"dear_why": "Expensive hours, about {eur} per kWh.",
"give_amount": "Worth up to {eur} at that price.",
"cycle_start": "Starts the monthly battery cycle.",
"cycle_start_would": "Would start the monthly battery cycle.",
+21 -8
View File
@@ -84,18 +84,18 @@
"on": "Aan", "off": "Uit",
"choice": {
"solar_first": {"name": "Eerst zonne-energie", "description": "Laadt met zonne-energie vóór de andere batterijen en dekt het huis."},
"leftover_solar": {"name": "Overschot zonne-energie", "description": "Laadt met zonne-energie die de andere batterijen niet opnemen, en dekt het huis."},
"leftover_solar": {"name": "Overschot zonne-energie", "description": "Krijgt zonne-energie die de andere niet opnemen, en levert in dure uren."},
"cheap_hours": {"name": "Alleen goedkope uren", "description": "Laadt in de goedkoopste uren en levert in de dure uren."},
"off": {"name": "Uit", "description": "Rust, en helpt alleen als de piekgrens haar nodig heeft."}
},
"later": "Alleen goedkope uren, volgt later",
"today": "Vandaag", "why": "Waarom",
"sell": {"name": "Verkopen in dure uren", "description": "Verkopen komt in een latere update."},
"modes_off": {"name": "Batterijmodi staan uit"},
"how": {"name": "Zo werken de modi", "description": "Welke batterij eerst zonne-energie krijgt, en wanneer elke batterij levert.",
"solar_first": {"description": "Eerst zonne-energie krijgt het overschot vóór de andere batterijen, en dekt het huis op elk uur."},
"leftover_solar": {"description": "Overschot zonne-energie krijgt wat de batterijen met Eerst zonne-energie niet opnemen, en dekt ook het huis."},
"leftover_solar": {"description": "Overschot zonne-energie krijgt wat de batterijen met Eerst zonne-energie niet opnemen, en spaart haar stroom voor dure uren."},
"cheap_hours": {"description": "Alleen goedkope uren laadt in de goedkoopste uren en levert in de dure uren."},
"off": {"description": "Uit laat de batterij rusten en start geen maandelijkse cyclus, maar de piekgrens mag haar nog gebruiken."},
"rules": {"description": "De piekgrens en de veiligheidsstops gaan altijd voor, en laden van het net en verkopen volgen later."}},
"rules": {"description": "De piekgrens en de veiligheidsstops gaan voor, en terwijl een batterij van het net laadt rusten de andere."}},
"toast": {"on": "{name} staat aan.", "off": "{name} staat uit.", "mode": "{name}: {mode}."}},
"settings": {"name": "Instellingen", "description": "De instellingen staan hieronder per taak.", "open": "Instellingen openen"}
},
@@ -190,7 +190,7 @@
"5": {"name": "Piekgrens", "description": "Het net blijft onder deze grens, gemiddeld over elk kwartier."},
"6": {"name": "Nakijken en starten", "description": "De add-on kijkt alles na met zijn eigen startcontroles en herstart dan."}
},
"modes_later": {"description": "Een modus kiezen per batterij komt in een latere update."},
"modes_later": {"description": "Zet na de installatie batterijmodi aan om per batterij een modus te kiezen op het scherm Batterijen."},
"kettle": {
"name": "Richting nakijken",
"description": "Lees de meter af, zet een waterkoker aan en lees de meter opnieuw af.",
@@ -224,7 +224,7 @@
},
"home_words": {
"mode_house": "Volgt het huis",
"modes": {"solar_first": "Eerst zonne-energie", "leftover_solar": "Overschot zonne-energie", "off": "Uit"},
"modes": {"solar_first": "Eerst zonne-energie", "leftover_solar": "Overschot zonne-energie", "cheap_hours": "Alleen goedkope uren", "off": "Uit"},
"attn_shadow": "Schaduwmodus staat aan",
"attn_shadow_body": "De add-on beslist alles, maar elke omvormer blijft op 0 W.",
"zero_held": "Op 0 W gehouden",
@@ -268,10 +268,19 @@
"leftover": "Ze krijgt de zonne-energie die de batterijen met Eerst zonne-energie niet opnemen.",
"leftover_alone": "Geen batterij met Eerst zonne-energie staat aan, dus zij krijgt eerst zonne-energie.",
"covers_house": "Ze levert stroom om het huis te dekken.",
"other_refuses": "{name} weigert te leveren, dus deze batterij neemt het over."
"other_refuses": "{name} weigert te leveren, dus deze batterij neemt het over.",
"grid_charge": "De prijs is laag, dus ze laadt van het net.",
"rests_charge": "Een andere batterij laadt van het net, dus deze rust.",
"selling": "De prijs is hoog, dus ze verkoopt aan het net.",
"expensive": "De prijs is hoog, dus ze dekt het huis.",
"saves": "Spaart haar stroom voor {at}.",
"saves_later": "Spaart haar stroom voor hogere prijzen.",
"no_prices": "Geen prijzen, dus ze volgt het huis.",
"no_spread": "De prijzen verschillen te weinig, dus ze volgt het huis."
},
"today": {
"not_planned": "Nog niet gepland.",
"no_runs": "Vandaag geen goedkope of dure uren meer.",
"off": "Uit, dus vandaag geen maandelijkse cyclus.",
"house": "Volgt het huis.",
"house_would": "Zou het huis volgen.",
@@ -298,6 +307,10 @@
"give_what": "Levert tot {kwh}, eerst aan het huis.",
"give_what_would": "Zou tot {kwh} leveren, eerst aan het huis.",
"give_why": "Hoogste gepubliceerde prijs, {eur} per kWh.",
"sell_what": "Verkoopt tot {kwh}, eerst aan het huis.",
"sell_what_would": "Zou tot {kwh} verkopen, eerst aan het huis.",
"cheap_why": "Goedkoopste uren, ongeveer {eur} per kWh.",
"dear_why": "Dure uren, ongeveer {eur} per kWh.",
"give_amount": "Tot {eur} waard aan die prijs.",
"cycle_start": "Start de maandelijkse batterijcyclus.",
"cycle_start_would": "Zou de maandelijkse batterijcyclus starten.",
+3 -2
View File
@@ -34,7 +34,8 @@ succeeds, so the page stops opening the wizard by itself (uipage.setup_needed).
UX-15's setup-only mode serves these routes from `setuponly.SetupOnly`, which
commands nothing; there `/action` answers 409.
OPT-11 P1 adds two `/action` values, `pack_use` and `pack_mode`, which set one
battery's on/off switch or mode in the store (Controller._pack_action). They
battery's on/off switch or mode in the store (Controller._pack_action); P3 adds
`pack_sell`, that battery's selling switch. They
are the only actions that name a battery; the configuration routes still reach
none of them.
"""
@@ -58,7 +59,7 @@ from .uimodel import DeriveError, options_to_ui
_LOG = logging.getLogger("goodwe.web")
# OPT-11: the per-battery actions, which name a battery and a value.
PACK_ACTIONS = ("pack_use", "pack_mode")
PACK_ACTIONS = ("pack_use", "pack_mode", "pack_sell")
# UX-13 P1: the page lives in app/static/ as three files - index.html,
# app.css, app.js - plus the page words from app/ui_text/<lang>.json (English
+67 -22
View File
@@ -36,7 +36,7 @@ import aiohttp # noqa: E402
from aiohttp import web as aioweb # noqa: E402
import ux13_demo as D # noqa: E402
from app import homeview, packmode, uilayout as L, uipage, web # noqa: E402
from app import homeview, messages, packmode, uilayout as L, uipage, web # noqa
from app.maintenance import IDLE, DRAIN # noqa: E402
fails = []
@@ -120,13 +120,15 @@ for st, want in WANT.items():
for e, n in zip(en, nl)))
_modes = ROWS[("modes", "en")]
check("modes on: each row carries its switch and mode, and Today says the "
"mode's plan is not made yet (OPT-11 P3/P4), without inventing one",
check("modes on: each row carries its switch and mode; Today says Solar "
"first follows the house and, before any cycle has made the price "
"runs (OPT-11 P3), that Leftover solar is not planned yet",
_modes["modes_on"] is True
and [(b["use"], b["mode_key"]) for b in _modes["batteries"]]
== [(True, "solar_first"), (True, "leftover_solar")]
and all(b["today_line"] == EN["home_words"]["today"]["not_planned"]
for b in _modes["batteries"]))
and [b["today_line"] for b in _modes["batteries"]]
== [EN["home_words"]["today"]["house"],
EN["home_words"]["today"]["not_planned"]])
_off = ROWS[("running", "en")]
check("modes off: no switch, no mode, and Today and Why say it follows the "
"house", _off["modes_on"] is False
@@ -212,16 +214,25 @@ print(f" {_kinds}")
check("Off: no 'follows the house', no trade, no monthly cycle for B",
not _kinds[1] and "house" in _kinds[0])
_p = build("off", now=MORNING)
_m = build("modes", now=MORNING)
# OPT-11 P3: with modes on the plan shows each battery's price runs, which a
# control cycle makes; make them as the cycle does (`_mode_roles()`).
_mc = D.controller("modes")
_mc._mode_roles(MORNING, None)
_m = build("modes", c=_mc, now=MORNING)
_kinds = {i: {r["kind"] for r in _p["plan"]["runs"] if r["battery"] == i}
for i in (0, 1)}
print(f" morning: {_kinds}, lines {_p['plan']['lines']}")
check("in the morning, with trades and a cycle ahead: A has them, Off B has "
"no block and waits for no cycle (with B on, it does)",
{"house", "grid", "give", "cycle"} <= _kinds[0] and not _kinds[1]
check("in the morning, with trades and a cycle ahead: Solar first A follows "
"the house and has its cycle but no price block (OPT-11 P3: Solar first "
"never trades), Off B has no block and waits for no cycle; with B on "
"(Leftover solar) B has its expensive run but no grid block (it takes "
"solar only until the forecast, owner decision 2026-09-29) and waits",
{"house", "cycle"} == _kinds[0] and not _kinds[1]
and not any(B_NAME in x for x in _p["plan"]["lines"])
and {"grid", "give"} <= {r["kind"] for r in _m["plan"]["runs"]
if r["battery"] == 1}
and "give" in {r["kind"] for r in _m["plan"]["runs"]
if r["battery"] == 1}
and "grid" not in {r["kind"] for r in _m["plan"]["runs"]
if r["battery"] == 1}
and any(B_NAME in x for x in _m["plan"]["lines"]))
_p = build("off_peak")
check("...but the peak block stays on it while the peak limit needs it",
@@ -290,6 +301,22 @@ _cases = [
(dict(mode=packmode.LEFTOVER_SOLAR, sf_ahead=True), "leftover"),
(dict(mode=packmode.LEFTOVER_SOLAR), "leftover_alone"),
({}, "solar_first"),
# OPT-11 P3: the battery's role in the price runs, and no plan.
(dict(mode=packmode.CHEAP_HOURS, role="charge"), "grid_charge"),
(dict(role="rest"), "rests_charge"),
(dict(mode=packmode.CHEAP_HOURS, role="sell", giving=True), "selling"),
(dict(mode=packmode.LEFTOVER_SOLAR, role="give", giving=True),
"expensive"),
(dict(mode=packmode.LEFTOVER_SOLAR, role="hold", has_at=True), "saves"),
(dict(mode=packmode.CHEAP_HOURS, role="hold"), "saves_later"),
(dict(mode=packmode.CHEAP_HOURS, role="house", plan="no_prices"),
"no_prices"),
(dict(mode=packmode.LEFTOVER_SOLAR, role="house", plan="no_spread",
giving=True), "no_spread"),
# ...and the urgent reasons still come first.
(dict(role="charge", refused=-1), "refuses_charge"),
(dict(role="sell", peak=True), "peak"),
(dict(role="hold", off=True), "off"),
]
_wrong = [(kw, want, why(**kw)) for kw, want in _cases if why(**kw) != want]
print(f" {_wrong}")
@@ -336,9 +363,13 @@ _r = asyncio.run(_serve("modes", [
"nl"),
({"action": "pack_use", "unit": B, "value": True}, "en"),
({"action": "pack_mode", "unit": A, "value": "cheap_hours"}, "en"),
({"action": "pack_sell", "unit": A, "value": True}, "en"),
({"action": "pack_sell", "unit": A, "value": "yes"}, "en"),
({"action": "pack_mode", "unit": A, "value": "sell_only"}, "en"),
]))
(_s1, _j1), (_, _st1), (_s2, _j2), (_, _st2), (_s3, _), (_, _st3), \
(_s4, _j4), (_, _st4) = _r
(_s4, _j4), (_, _st4), (_s5, _j5), (_, _st5), (_s6, _j6), (_, _st6), \
(_s7, _j7), (_, _st7) = _r
check("switch B off: 200, and /status shows B Off with 'You switched it off.'",
_s1 == 200 and _j1["result"]["use"] is False
and _st1["batteries"][1]["mode_key"] == "off"
@@ -351,9 +382,18 @@ check("set A to Leftover solar: 200, and /status?lang=nl shows it in Dutch",
check("switch B back on: its stored mode returns",
_s3 == 200 and _st3["batteries"][1]["mode_key"]
== packmode.LEFTOVER_SOLAR)
check("a mode the list only shows greyed out is refused with a plain 400, and "
"nothing changes", _s4 == 400 and _j4["ok"] is False and _j4["error"]
and _st4["batteries"][0]["mode_choice"] == packmode.LEFTOVER_SOLAR)
check("OPT-11 P3: Cheap hours only is a mode now: 200, and /status shows it",
_s4 == 200 and _st4["batteries"][0]["mode_choice"]
== packmode.CHEAP_HOURS)
check("owner decision 2026-09-29 (option A): selling is off, so a posted "
"pack_sell is refused with the plain reason 'Selling is off for now.', "
"and /status says no battery sells or can",
_s5 == 400 and _j5["error"] == messages.pack_refused("sell_off")
and _s6 == 400 and all(b["sell"] is False and b["can_sell"] is False
for b in _st5["batteries"] + _st6["batteries"]))
check("...a mode that does not exist is refused with a plain 400, and "
"nothing changes", _s7 == 400 and _j7["error"]
and _st7["batteries"][0]["mode_choice"] == packmode.CHEAP_HOURS)
_r = asyncio.run(_serve("running", [
({"action": "pack_use", "unit": B, "value": False}, "en")]))
check("with battery modes off the switch is refused with the plain reason, "
@@ -376,14 +416,19 @@ check("Settings has a Simple / Installer choice as a radio group, and Simple "
and PAGE.count('role="radio"') >= 2 and 'id="simple"' in PAGE)
_acts = set(re.findall(r'packAct\("(\w+)"', JS))
check(f"the cards post only the existing actions ({sorted(_acts)}), as "
f"{{action, unit, value}}", _acts == set(web.PACK_ACTIONS)
f"{{action, unit, value}}; selling has no control while it is off",
_acts == set(web.PACK_ACTIONS) - {"pack_sell"}
and packmode.SELLING_ENABLED is False
and "action: what, unit: b.name, value: value" in JS)
check("the switch is a real switch and the list names cheap hours as coming "
"later, not choosable", 'role: "switch"' in JS and '"aria-checked"' in JS
and 'h("option", { value: "", disabled: true }, t("batteries.later"))'
in JS)
check("the selling switch is greyed out until OPT-11 P3 (no pack_sell is "
"posted)", "pack_sell" not in JS and "disabled: true })" in JS)
check("the switch is a real switch and the list offers the three modes, "
"cheap hours included, with nothing greyed out",
'role: "switch"' in JS and '"aria-checked"' in JS
and 'const MODES = ["solar_first", "leftover_solar", "cheap_hours"];'
in JS and "batteries.later" not in JS)
check("no selling switch anywhere on the page or in Settings, and no words "
"for one", "pack_sell" not in JS and "sell" not in SJS
and "batteries.sell" not in JS
and all("sell" not in W["home"]["batteries"] for W in LANGS.values()))
_T = uipage.ui_text("en")
_lay = uipage.layout(D.schema_ui(D.CONFIG), {"en": {}})
_simple = [f for g in _lay["groups"] for p in g["parts"] for f in p["fields"]
+2 -2
View File
@@ -335,11 +335,11 @@ _bad = [asyncio.run(_c.handle_action("pack_use", unit="inverter-z",
asyncio.run(_c.handle_action("pack_use", unit=A, value=0)),
asyncio.run(_c.handle_action("pack_use", unit=A, value=1)),
asyncio.run(_c.handle_action("pack_mode", unit=A,
value=packmode.CHEAP_HOURS)),
value="sell_only")),
asyncio.run(_c.handle_action("pack_mode", unit=A, value=None))]
check("an unknown battery, a switch that is not true or false (0 and 1 "
"included), and a mode "
"this release does not offer (Cheap hours arrives in P3) are refused, "
"that does not exist are refused, "
"with a plain reason, and store nothing",
all(isinstance(x, dict) and set(x) == {"error"} for x in _bad)
and unit_key(A, packmode.MODE_KEY) not in _c.store.data
+722
View File
@@ -0,0 +1,722 @@
"""Runnable check for OPT-11 P3, price runs per battery. `python3 test_modes_p3.py`
Drives the pure modules (priceruns, modetrade) and the REAL Controller
(tools/opt11p3_replay.py: read_inputs + cycle on a P1 meter, the arbiter, the
allocator and the peak guard, with two lagged model batteries), in simulated
time. DECISIONS/OPT-11-battery-modes.md has the record.
1. With battery modes off, every command and trade is byte-identical to
release/1.0 7a00f12 on the trade path P3 changed (digests measured there),
and the harness would see a change there.
2. Price runs: consecutive, firm, sized by energy; give runs clear the spread;
the sell run sits in a give run with no slot below 0; a started run is kept.
3. Roles: a fuzz of modetrade.decide() and size() against the invariants.
4. The real controller over a day: no opposite-direction split, no grid
charge on Solar first, the quarter never above the aim while charging, no
sell below the floor or the spread, holding outside the expensive hours.
5. A refusal wins over a run.
6. No prices or too small a gap: the price modes follow the house; Solar first
alone never trades.
7. A run edge reseeds the loop (handover measured with and without).
8. The site Why row: a limit that pushes power up does not "hold it back" (F2).
9. Selling is off (owner, 2026-09-29): `pack_sell` refused, read as false.
10. The Batteries screen: Why and Today lines from the roles and runs.
11. Review additions: modes on ignore the site trade switch, a battery in
maintenance takes no part, the headroom stop latches for the period.
12. Option A: no sale in a day; Leftover solar never grid-charges (P4).
Mutations (MUTATION.md, purged copies, graded over this file and the named
tests) are listed in the pull request.
"""
import asyncio
import logging
import os
import random
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)
from named_absence import nth, present # noqa: E402
import opt11p3_identity as ident # noqa: E402
import opt11p3_replay as rp # noqa: E402
from app import homeview, messages, modetrade, packmode, priceruns # noqa
from app import arbitrage # noqa: E402
from app.arbiter import Claim, P_ARBITRAGE, P_LOOP, P_PEAK, Resolution # noqa
from app.main import Controller, why_raised # noqa: E402
from app.price import PricePoint, brussels # noqa: E402
from app.store import unit_key # noqa: E402
fails = []
def check(name, cond):
print(f" {'ok ' if cond else 'FAIL'} {name}")
if not cond:
fails.append(name)
L = timezone(timedelta(hours=2))
A, B = "inverter-a", "inverter-b"
# ============================================================================ #
print("1. modes off: byte-identical to 7a00f12 on the trade path")
# Printed by tools/opt11p3_identity.py on release/1.0 7a00f12 (2026-09-29),
# goodwe-addon-checks:1. 1080 cycles each; the trade CHARGEs and DISCHARGEs.
AT_7A00F12 = {
"n1": "71245b227ca40bada4f71f57ea6154475f4b0bd60239c9ad4f80bffb0094514c",
"n2": "56cf28010b06765c29e5479abb815d7503d570b39240d82dfc818efa90c93722",
"n1_off": "71245b227ca40bada4f71f57ea6154475f4b0bd60239c9ad4f80bffb0094514c",
"n2_off": "56cf28010b06765c29e5479abb815d7503d570b39240d82dfc818efa90c93722",
}
_rows = {k: ident.run(**v)[0] for k, v in ident.SCENARIOS.items()}
_got = {k: ident.digest(v) for k, v in _rows.items()}
for k in AT_7A00F12:
print(f" {k} {'same' if _got[k] == AT_7A00F12[k] else 'DIFFERS'} "
f"{_got[k][:16]}")
check("with battery modes off (option absent or false), every command, "
"integrator, allocation and trade sentence equals 7a00f12's, at one "
"inverter and at two", _got == AT_7A00F12)
_acts = {r[-1][0] for r in _rows["n2"] if r[-1]}
check("...and the runs reach the trade P3 now skips with modes on: a charge "
"and a sale", {arbitrage.CHARGE, arbitrage.DISCHARGE} <= _acts)
def _moved(scn, patch):
"""Does `patch` (applied before each cycle) move scenario `scn`?"""
real = ident._feed
def feed(c, t):
patch(c)
real(c, t)
ident._feed = feed
try:
return ident.digest(ident.run(**ident.SCENARIOS[scn])[0]) \
!= AT_7A00F12[scn]
finally:
ident._feed = real
def _as_if_trade_ended(c):
c._mode_trade_was = True # the run-edge reseed, leaking to modes off
def _moved_headroom(scn):
"""Does an unmeasured headroom (the expression P3 hoisted) move it?"""
from app import peakguard
real = peakguard.charge_headroom_w
peakguard.charge_headroom_w = lambda g: None
try:
return ident.digest(ident.run(**ident.SCENARIOS[scn])[0]) \
!= AT_7A00F12[scn]
finally:
peakguard.charge_headroom_w = real
check("non-vacuous: a run-edge reseed leaking into modes off moves n2, and so "
"does the headroom P3 hoisted, if it went unmeasured",
_moved("n2", _as_if_trade_ended) and _moved_headroom("n1"))
# ============================================================================ #
print("\n2. price runs")
D0 = datetime(2026, 9, 9, 0, 0, tzinfo=L)
def pts(prices, start=D0, minutes=15, provisional=()):
return [PricePoint(start=brussels(start + timedelta(minutes=minutes * i)),
eur_per_kwh=p, provisional=i in provisional)
for i, p in enumerate(prices)]
# 24 h of quarters: cheap 02:00-04:00 (and one cheaper single quarter at
# 05:00), dear 18:00-20:00 with a peak quarter at 19:00, flat otherwise.
DAY = [0.30] * 96
for i in range(8, 16):
DAY[i] = 0.10
DAY[20] = 0.05
for i in range(72, 80):
DAY[i] = 0.50
DAY[76] = 0.70
P = pts(DAY)
NOW = D0 + timedelta(minutes=5)
r1 = priceruns.plan(P, NOW, required_eur_kwh=0.25, charge_kwh=5.0,
charge_kw=2.5, sell_kwh=0.0, sell_kw=0.0)
r2 = priceruns.plan(P, NOW, required_eur_kwh=0.25, charge_kwh=0.0,
charge_kw=0.0, sell_kwh=2.5, sell_kw=2.5)
print(f" 2 h need: {r1.cheap}; give {r1.give}")
check("the cheap run lasts what the energy needs at the rating (5 kWh at "
"2.5 kW = 2 h) and is the cheapest such consecutive window, not the "
"cheapest single quarter",
r1.cheap.start == D0 + timedelta(hours=2)
and r1.cheap.end == D0 + timedelta(hours=4)
and abs(r1.cheap.mean_eur_kwh - 0.10) < 1e-9)
check("...with nothing to charge, one quarter: the cheapest one",
r2.cheap.start == D0 + timedelta(hours=5)
and r2.cheap.end == D0 + timedelta(hours=5, minutes=15))
check("give runs are the consecutive quarters at least the required spread "
"above the cheap run (0.10 + 0.25): 18:00-20:00, one run",
r1.on and len(r1.give) == 1
and r1.give[0].start == D0 + timedelta(hours=18)
and r1.give[0].end == D0 + timedelta(hours=20))
check("the sell run is the dearest window inside a give run, sized by the "
"sellers' energy (2.5 kWh at 2.5 kW = 1 h)",
r2.sell.end - r2.sell.start == timedelta(hours=1)
and r1.give[0].start <= r2.sell.start < r2.sell.end <= r1.give[0].end
and r2.sell.start <= D0 + timedelta(hours=19) < r2.sell.end)
_flat = priceruns.plan(pts([0.3] * 96), NOW, required_eur_kwh=0.1,
charge_kwh=5, charge_kw=2.5, sell_kwh=0, sell_kw=0)
_thr0 = priceruns.plan(pts([0.3] * 96), NOW, required_eur_kwh=0.0,
charge_kwh=5, charge_kw=2.5, sell_kwh=0, sell_kw=0)
check("a flat day has no plan (`not_worth`), even with a threshold of 0: the "
"cheap run never counts as its own expensive run",
_flat.reason == priceruns.NOT_WORTH and not _flat.give
and _thr0.reason == priceruns.NOT_WORTH)
_prov = priceruns.plan(pts(DAY, provisional=set(range(74, 96))), NOW,
required_eur_kwh=0.25, charge_kwh=5, charge_kw=2.5,
sell_kwh=2.5, sell_kw=2.5)
check("a provisional (carried) price is never part of a run (section 8.3)",
all(g.end <= D0 + timedelta(hours=18, minutes=30) for g in _prov.give)
and (_prov.sell is None
or _prov.sell.end <= D0 + timedelta(hours=18, minutes=30)))
_neg = list(DAY)
_neg[76] = -0.05
_rn = priceruns.plan(pts(_neg), NOW, required_eur_kwh=0.25, charge_kwh=0,
charge_kw=0, sell_kwh=10.0, sell_kw=2.5)
check("the sell run never spans a quarter priced below 0 (section 17)",
_rn.sell is not None and not (_rn.sell.start <= D0 + timedelta(
hours=19) < _rn.sell.end))
# All prices below 0 but one: the give run holds negative quarters (they
# clear the spread over an even lower cheap run), and the sale must skip them.
_allneg = [-0.30] * 96
for i in range(8, 16):
_allneg[i] = -0.40
for i in range(72, 80):
_allneg[i] = -0.05
_allneg[76] = 0.10
_rneg = priceruns.plan(pts(_allneg), NOW, required_eur_kwh=0.15,
charge_kwh=0, charge_kw=0, sell_kwh=2.5, sell_kw=2.5)
check("...even when the whole give run is below 0 but one quarter: the sale "
"is that quarter only",
_rneg.on and _rneg.sell is not None
and _rneg.sell.start == D0 + timedelta(hours=19)
and _rneg.sell.end == D0 + timedelta(hours=19, minutes=15))
# A spike inside the cheap run clears the spread over the run's own mean:
# it must not also be a give run (charge and give would overlap).
_sp = [0.50] * 96
for i in range(8, 16):
_sp[i] = 0.02
_sp[15] = 0.20
_rsp = priceruns.plan(pts(_sp), NOW, required_eur_kwh=0.05, charge_kwh=5.0,
charge_kw=2.5, sell_kwh=0, sell_kw=0)
check("no give run overlaps the cheap run, even a quarter inside it that "
"clears the spread over the run's mean",
_rsp.cheap.start == D0 + timedelta(hours=2)
and all(g.end <= _rsp.cheap.start or g.start >= _rsp.cheap.end
for g in _rsp.give))
_few = priceruns.plan(pts([0.1]), NOW, required_eur_kwh=0.1, charge_kwh=1,
charge_kw=1, sell_kwh=0, sell_kw=0)
check("fewer than two firm slots: no plan (`no_prices`)",
_few.reason == priceruns.NO_PRICES)
_mid = D0 + timedelta(hours=3)
_new = priceruns.plan(P, _mid, required_eur_kwh=0.25, charge_kwh=0.0,
charge_kw=0.0, sell_kwh=0.0, sell_kw=0.0)
_kept = priceruns.keep_started(r1, _new, _mid)
check("a run under way is kept to its end when a new plan would move it "
"(no flapping), and a run not yet started is not",
_new.cheap != r1.cheap and _kept.cheap == r1.cheap
and priceruns.keep_started(r1, _new, NOW).cheap == _new.cheap)
_late = priceruns.plan(P, D0 + timedelta(hours=20, minutes=5),
required_eur_kwh=0.25, charge_kwh=5, charge_kw=2.5,
sell_kwh=0, sell_kw=0)
check("a run that has ended is not planned again, and nothing beyond the "
"firm prices is invented",
all(g.start >= D0 + timedelta(hours=20) for g in _late.give)
and not _late.on)
# ============================================================================ #
print("\n3. roles: modetrade.decide() fuzz")
rnd = random.Random(311)
_bad = []
_seen = set()
for _ in range(20000):
n = rnd.choice((1, 2, 3))
packs = []
for k in range(n):
packs.append(modetrade.Pack(
f"inv-{k}", rnd.random() > 0.15,
rnd.choice(packmode.MODES), rnd.random() > 0.5,
rnd.choice((1000.0, 2500.0, 5000.0)),
rnd.choice((None, 5.0, 19.0, 20.0, 21.0, 60.0, 95.0, 99.0)),
rnd.choice((None, 5.0, 10.0)), live=rnd.random() > 0.1,
idle=rnd.random() > 0.1, refused=rnd.choice((0, 0, 0, 1, -1))))
t = D0 + timedelta(minutes=rnd.randrange(0, 24 * 60))
runs = rnd.choice((None, r1, r2, _flat))
ok_s = rnd.random() > 0.3
ok_c = ok_s and rnd.random() > 0.3
hr = rnd.choice((None, 0.0, 100.0, 499.0, 501.0, 3000.0))
price = rnd.choice((None, -0.1, 0.0, 0.5))
house = rnd.choice((None, -500.0, 800.0))
was = rnd.random() > 0.7
blocked = rnd.random() > 0.8
tk = modetrade.decide(runs, t, packs, charge_ok=ok_c, sell_ok=ok_s,
trade_why="x", price_now=price, headroom_w=hr,
floor_pct=20.0, ceiling_pct=95.0, house_w=house,
charging_now=was, blocked=blocked)
_seen.add(tk.action)
by = {p.name: p for p in packs}
for name, (lo, hi) in tk.ranges.items():
if not lo <= 0.0 <= hi:
_bad.append(("range without 0", name, lo, hi))
for name, role in tk.roles.items():
p = by[name]
if role == modetrade.CHARGE and (p.mode != packmode.CHEAP_HOURS
or p.refused < 0 or not p.use
or p.soc is None or p.soc >= 95.0):
_bad.append(("bad charger", p))
if role == modetrade.SELL and (not p.sell or p.refused > 0
or p.mode == packmode.SOLAR_FIRST
or p.soc is None or p.soc <= 20.0):
_bad.append(("bad seller", p))
if not p.use and tk.ranges.get(name, (0.0, 0.0)) != (0.0, 0.0):
_bad.append(("off not 0", p))
if tk.action == modetrade.CHARGE:
need = 0.0 if was else modetrade.START_W
if not (ok_c and hr is not None and hr > need and not blocked
and runs is not None and runs.cheap.holds(t)):
_bad.append(("charge without leave", ok_c, hr, blocked))
for name, (lo, hi) in tk.ranges.items():
if tk.roles[name] != modetrade.CHARGE and lo < 0.0:
_bad.append(("grid charge on a non-charger", by[name]))
s = modetrade.size(tk, (-7000.0, 7000.0), hr)
if not (s < 0 and -s <= min(-tk.ask_w, hr) + 1e-9):
_bad.append(("charge size", s, tk.ask_w, hr))
if modetrade.size(tk, (-300.0, 300.0), hr) < -300.0:
_bad.append(("charge over the bound",))
if tk.action == modetrade.SELL:
if not (ok_s and price is not None and price >= 0.0 and runs
is not None and runs.sell.holds(t) and house is not None):
_bad.append(("sell without leave", ok_s, price))
s = modetrade.size(tk, (-7000.0, 2000.0), hr)
if not 0.0 < s <= 2000.0:
_bad.append(("sell size", s))
if tk.action is None and modetrade.size(tk, (-1e4, 1e4), hr) != 0.0:
_bad.append(("no action but a size",))
print(f" actions seen {sorted(map(str, _seen))}; {len(_bad)} bad: {_bad[:3]}")
check("20000 random ticks: every range holds 0 W; Off is [0, 0]; no Solar "
"first, refusing, full or unknown battery charges; no Solar first, "
"refusing or at-floor battery sells; while one charges no other may "
"charge; a charge needs the guard, the headroom (500 W to start) and "
"no stop this period, and is sized under the headroom; a sale needs a "
"price >= 0 in the sell run", not _bad
and {modetrade.CHARGE, modetrade.SELL, None} <= _seen)
# ============================================================================ #
print("\n4. the real controller over a day (A Solar first, B Cheap hours + sell)")
# Wholesale; all-in (x + 0.12) x 1.06: cheap 0.148, flat 0.339, dear 0.551.
# With a cycle cost of 0.20 (0.30 widened) only 19:00-20:00 clears it.
DAYW = [0.20] * 96 * 2
for i in list(range(4, 12)) + list(range(100, 108)):
DAYW[i] = 0.02 # 01:00-03:00 wholesale
for i in list(range(76, 80)) + list(range(172, 176)):
DAYW[i] = 0.40 # 19:00-20:00
UTC0 = D0.astimezone(timezone.utc)
WPTS = [(UTC0 + timedelta(minutes=15 * i), 900, p) for i, p in enumerate(DAYW)]
def house_flat(w):
return [(UTC0.timestamp() - 10, float(w))]
AIM = rp.PEAK_CAP - 50
def run_day(variant, *, house_w=2600.0, start_h=0.75, hours=21.0,
soc0=None, choice=None, refuse=(), n=2, check_cb=None,
patch=None, house=None, limits=None):
rows = []
def cb(c, h, now, pr):
sh = dict(c.allocation.shares) if c.allocation is not None and \
c.allocation.shares else {}
rows.append(dict(
now=now, grid=h.grid(), shares=sh,
action=getattr(c.arb, "action", None),
winner=c.reason.split(" ->")[0],
roles=dict(c.mode_tick.roles) if c.mode_tick else {},
soc={u: p.soc for u, p in h.plants.items()},
p={u: p.p for u, p in h.plants.items()},
runs=c.mode_runs, price=pr, why=c.why_words,
unmet=getattr(c.allocation, "unmet_w", 0) or 0))
if check_cb:
check_cb(c, h, now, pr)
st, c = rp.run(variant, house or house_flat(house_w), WPTS,
UTC0 + timedelta(hours=start_h), 3600 * hours, check=cb,
soc0=soc0, choice=choice, refuse=refuse, n=n, patch=patch,
extra={"cycle_cost_threshold_eur": 0.20}, limits=limits)
return rows, st, c
# Selling is off (owner, 2026-09-29, option A). Its code is kept behind
# packmode.SELLING_ENABLED, and these checks turn it on to hold that code to
# its invariants; part 12 checks it is unreachable with the flag as shipped.
_SELL_SHIPPED = packmode.SELLING_ENABLED
packmode.SELLING_ENABLED = True
ROWS, ST, _c4 = run_day("p3_ch_sell", soc0={"a": 60.0, "b": 30.0})
charge = [r for r in ROWS if r["action"] == arbitrage.CHARGE]
sell = [r for r in ROWS if r["action"] == arbitrage.DISCHARGE]
print(f" {len(ROWS)} cycles, {len(charge)} charging, {len(sell)} selling; "
f"{rp.summary(ST)}")
check("B charged from the grid in the cheap run and sold in the dear one, "
"and A never took part in either",
charge and sell
and all(r["roles"][A] in (modetrade.REST, modetrade.HOUSE)
for r in charge + sell)
and all(r["now"].astimezone(L).hour in (0, 1, 2, 3) for r in charge)
and all(r["runs"].cheap.holds(r["now"]) for r in charge)
and all(r["now"].astimezone(L).hour == 19 for r in sell))
_mixed = [r for r in ROWS if any(v > 0 for v in r["shares"].values())
and any(v < 0 for v in r["shares"].values())]
check("no opposite-direction split on any cycle (every share one sign)",
not _mixed)
check("no grid charge on Solar first: while the site charges from the grid, "
"A's share is never below 0 W",
all(r["shares"].get(A, 0) >= 0 for r in charge))
# The quarters the add-on itself banked (the P1 blocks, as the bill reads).
_chq = {int(r["now"].timestamp() // 900) for r in charge}
_qc = [b.offtake_avg_w for b in _c4.p1.blocks
if int(b.start.timestamp() // 900) in _chq]
print(f" banked quarters with a grid charge: {len(_qc)}, highest "
f"{max(_qc):.2f} W, aim {AIM} W")
check("never above the SAFETY-79 aim while grid charging: every banked "
"15-minute period with a grid charge is at most cap - margin",
_qc and max(_qc) <= AIM + 0.5)
check("...and the charge is sized to the headroom, not the rating: under a "
"2.6 kW house B charges well under its 2.5 kW",
max(-r["shares"].get(B, 0) for r in charge) < 1200)
_sold = [r for r in sell]
check("no sale below the floor or the spread: every sale cycle has B above "
"arbitrage_soc_floor (20 %) and a price at least cheap + required",
all(r["soc"]["b"] > 19.9 for r in _sold)
and all(r["price"] is not None and rp.all_in(r["price"])
>= r["runs"].cheap.mean_eur_kwh + 0.30 - 1e-9 for r in _sold))
_hold = [r for r in ROWS if r["roles"].get(B) == modetrade.HOLD]
check("outside the expensive hours B holds its power: never gives while "
"holding", _hold and all(r["shares"].get(B, 0) <= 0 for r in _hold))
_give = [r for r in ROWS if r["roles"].get(B) in (modetrade.GIVE,
modetrade.SELL)]
check("...and gives only in the expensive hours (19:00-20:00)",
all(r["now"].astimezone(L).hour == 19 for r in _give))
def starts(rows, kind):
return sum(1 for a, b in zip(rows, rows[1:])
if a["action"] != kind and b["action"] == kind)
check("the sale starts once and runs to its end: no on-off while the two "
"batteries' 5 kW discharge makes the peak guard read 'engaged'",
starts(ROWS, arbitrage.DISCHARGE) == 1)
# B's inverter can only take 1000 W: the trade asks for that, not its rating.
ROWS_U, _, _ = run_day("p3_ch_sell", hours=1.0, house_w=800.0,
soc0={"a": 60.0, "b": 30.0},
limits={"b": (-1000.0, 1000.0)})
_cu = [r for r in ROWS_U if r["action"] == arbitrage.CHARGE]
print(f" B limited to 1 kW: charge {min(r['shares'].get(B, 0) for r in _cu)}"
f" W, unmet {max(abs(r['unmet']) for r in _cu)} W")
check("SAFETY-45 sizing inside the modes: the charge is sized to what the "
"batteries may place, so none of it is left unmet (P1's gap)",
_cu and all(abs(r["unmet"]) < 1 for r in _cu)
and min(r["shares"].get(B, 0) for r in _cu) >= -1000)
# A house that swings over the limit every minute during the cheap run.
_sw = [(UTC0.timestamp() + 60 * k, 2000.0 if k % 2 else 3900.0)
for k in range(0, 200)]
ROWS_W, _, _ = run_day("p3_ch_sell", hours=1.5, house=_sw,
soc0={"a": 60.0, "b": 30.0})
_per_q = {}
for a_, b_ in zip(ROWS_W, ROWS_W[1:]):
if a_["action"] != arbitrage.CHARGE and b_["action"] == arbitrage.CHARGE:
q = int(b_["now"].timestamp() // 900)
_per_q[q] = _per_q.get(q, 0) + 1
print(f" house swinging 2.0/3.9 kW each minute: charge starts per period "
f"{sorted(_per_q.values())}")
check("a charge the peak guard or the headroom stopped stays off for the "
"rest of that 15-minute period (at most one start per period)",
_per_q and max(_per_q.values()) == 1)
# At the floor: B starts the sale just above it.
ROWS_F, _, _ = run_day("p3_ch_sell", start_h=18.75, hours=1.5,
soc0={"a": 60.0, "b": 24.0})
_sf = [r for r in ROWS_F if r["action"] == arbitrage.DISCHARGE]
_after = [r for r in ROWS_F if r["now"].astimezone(L).hour == 19
and r["soc"]["b"] <= 20.0]
_lastf = nth(ROWS_F, -1, "the last cycle")
print(f" from 24 %: sold {len(_sf)} cycles, B ends at "
f"{_lastf['soc']['b'] if present(_lastf) else _lastf} %")
check("the sale stops at arbitrage_soc_floor and does not re-arm in the run",
_sf and _after and all(r["action"] != arbitrage.DISCHARGE
for r in _after))
packmode.SELLING_ENABLED = _SELL_SHIPPED
# N=1: one Cheap hours battery.
ROWS1, _, _ = run_day("p3_ch_sell", n=1, soc0={"a": 30.0},
choice={"a": (True, "cheap_hours", False)},
hours=3.0)
_c1 = [r for r in ROWS1 if r["action"] == arbitrage.CHARGE]
check("at one inverter the same rules: it charges in the cheap run, sized "
"under the aim", _c1 and max(r["grid"] for r in _c1) < AIM + 400)
# ============================================================================ #
print("\n5. a refusal wins over a run")
ROWS_R, _, _ = run_day("p3_ch_sell", hours=1.0, soc0={"a": 60.0, "b": 30.0},
refuse=("b",))
_first = nth(ROWS_R, 0, "the first cycle")
_ref = [r for r in ROWS_R if present(_first)
and r["now"] > _first["now"] + timedelta(minutes=3)]
print(f" B refuses to charge; after 3 min: trades "
f"{sorted({str(r['action']) for r in _ref})}, B roles "
f"{sorted({r['roles'].get(B) for r in _ref})}")
check("B refusing to charge: no grid trade is filed after the refusal is "
"learned, B's role is no longer 'charge', and A covers the house",
_ref and all(r["action"] is None for r in _ref)
and all(r["roles"].get(B) != modetrade.CHARGE for r in _ref)
and sum(r["shares"].get(A, 0) > 0 for r in _ref) > len(_ref) / 2)
# ============================================================================ #
print("\n6. no prices, no gap, Solar first alone")
def _no_plan(c):
c.tariff_plan = None
c.fresh_tariff_plan = lambda now: None
ROWS_N, _, _ = run_day("p3_ch_sell", hours=1.0, patch=_no_plan,
soc0={"a": 60.0, "b": 60.0}, house_w=1500.0)
check("no prices: no trade, and Leftover solar / Cheap hours follow the "
"house (they give power, not held forever)",
all(r["action"] is None for r in ROWS_N)
and all(r["roles"].get(B) == modetrade.HOUSE for r in ROWS_N)
and any(r["shares"].get(B, 0) > 0 for r in ROWS_N))
ROWS_S, _, _ = run_day("p3_ch_sell", hours=21.0,
choice={"a": (True, "solar_first", False),
"b": (True, "solar_first", True)})
check("both batteries Solar first: never a grid charge or a sale all day, "
"even with selling switched on",
all(r["action"] is None for r in ROWS_S))
# ============================================================================ #
print("\n7. a run edge reseeds the loop")
def _no_reseed(c):
c._reseed_run_edge = lambda claims: None
def edge(rows, kind):
for i in range(1, len(rows)):
if rows[i - 1]["action"] == kind and rows[i]["action"] != kind:
w = [r["grid"] for r in rows[i:i + 36]]
return min(w), max(w)
return None
_e_on = edge(run_day("p3_ch_sell", hours=3.0, house_w=800.0,
soc0={"a": 60.0, "b": 70.0})[0], arbitrage.CHARGE)
_e_off = edge(run_day("p3_ch_sell", hours=3.0, house_w=800.0,
soc0={"a": 60.0, "b": 70.0}, patch=_no_reseed)[0],
arbitrage.CHARGE)
print(f" grid in the 3 min after a 2.5 kW charge ends, 800 W house: "
f"reseeded {_e_on}, not reseeded {_e_off}")
check("reseeded, the loop takes over without the import spike the trade's "
"watts leave behind (and exports no more than without)",
_e_on and _e_off and _e_on[1] < 1000 and _e_off[1] > 2000
and _e_on[0] > _e_off[0] - 100)
# ============================================================================ #
print("\n8. the site Why row (OPT-11 P1 test, F2)")
_cl = [Claim.set("loop", P_LOOP, 1200.0), Claim.limit("peak", P_PEAK,
lo=2500.0)]
_up = Resolution(2500.0, "loop", bound_by="peak(quarter peak)")
_dn = Resolution(800.0, "loop", bound_by="peak(quarter peak)")
check("a peak floor that raised the power reads 'raised for the peak limit', "
"never 'held back'",
why_raised(_up, _cl) is True and "held back" not in messages.why(
"loop", 2500.0, "peak(quarter peak)", raised=True)
and "raised for the peak limit" in messages.why(
"loop", 2500.0, "peak(quarter peak)", raised=True))
check("...a limit that lowered it still reads 'held back'",
why_raised(_dn, _cl) is False
and "held back by" in messages.why("loop", 800.0, "pack"))
check("...and a trade the peak floor raised reads the same way",
why_raised(Resolution(2000.0, "arbitrage", bound_by="peak"),
[Claim.set("arbitrage", P_ARBITRAGE, -1000.0)]) is True)
# ============================================================================ #
print("\n9. the selling switch is off (owner decision 2026-09-29, option A)")
_st = rp._Store()
_st.data[unit_key(B, packmode.SELL_KEY)] = True # an old stored "on"
_st.data[unit_key(B, packmode.MODE_KEY)] = packmode.CHEAP_HOURS
_ctl = Controller(dict(rp.options(True, False)), rp.Hass(("a", "b"),
lambda t: 0.0), _st, rp._Pub(), stop=asyncio.Event())
_r9 = [asyncio.run(_ctl.handle_action("pack_sell", unit=B, value=v))
for v in (True, False, "yes")]
check("pack_sell is refused with the plain reason 'Selling is off for now.' "
"and stores nothing",
all(x == {"error": messages.pack_refused("sell_off")} for x in _r9)
and _st.data[unit_key(B, packmode.SELL_KEY)] is True
and not any("selling switched" in str(e) for e in _ctl.events))
check("...a stored 'on' reads as false: no battery sells or can, whatever "
"the store holds", packmode.SELLING_ENABLED is False
and packmode.read(_st.data, B).sell is False
and packmode.read(_st.data, B).can_sell is False
and all(p.sell is False for p in _ctl._mode_packs()))
# ============================================================================ #
print("\n10. the Batteries screen")
_keys = set(homeview.ROLE_WHY.values()) | {"saves_later", "no_prices",
"no_spread"}
_en, _nl = homeview.uitext.words("en"), homeview.uitext.words("nl")
check("every role's Why line has words in English and Dutch, and they differ",
all(k in _en["home_words"]["why"] and k in _nl["home_words"]["why"]
and _en["home_words"]["why"][k] != _nl["home_words"]["why"][k]
for k in _keys))
packmode.SELLING_ENABLED = True # the kept sell code, as in part 4
_c10 = run_day("p3_ch_sell", start_h=18.0, hours=0.5,
soc0={"a": 60.0, "b": 80.0})[2]
_now10 = D0 + timedelta(hours=18, minutes=30)
_plan = homeview.mode_runs(_c10, *homeview.local_day(_now10.astimezone(
timezone.utc)), _now10.astimezone(timezone.utc), False)
packmode.SELLING_ENABLED = _SELL_SHIPPED
print(f" plan blocks for B at 18:30: "
f"{[(r['from'], r['to'], r['kind'], r['what']) for r in _plan[0]]}")
check("the plan draws B's expensive run, the sale inside it, and nothing "
"for Solar first A", _plan[0]
and {r["battery"] for r in _plan[0]} == {1}
and any(r["kind"] == "give" for r in _plan[0]))
_sale = [r for r in _plan[0] if r["what"].startswith("Sells")]
check("a sale block shows no money: the price shown is what the house pays, "
"and a sale earns the lower injection price",
_sale and all(r["amount"] is None and r["eur"] is None for r in _sale))
_why = homeview.battery_why(
modes_on=True, mode=packmode.CHEAP_HOURS, off=False, reading=True,
phase="idle", shadow=False, running=True, held=False, refused=0,
peak=False, sf_ahead=True, role="hold", has_at=True)
check("a holding battery's Why names when it gives power",
_why == "saves" and "18:00 - 19:00" in homeview.why_words(
_why, at="18:00 - 19:00"))
# ============================================================================ #
print("\n11. review additions (PR #153): gaps four mutants survived")
import types # noqa: E402
import app.main as _main # noqa: E402
# Modes on with the site switch on: only the modes trade (R13).
ROWS_X, _, _ = run_day("p2_arb", hours=3.0, soc0={"a": 30.0, "b": 30.0},
choice={"a": (True, "solar_first", False),
"b": (True, "solar_first", False)})
check("modes on, 'Trade on the price spread' on, both batteries Solar first: "
"no trade all through the cheap hours (the site trade is not run)",
ROWS_X and all(r["action"] is None for r in ROWS_X)
and all(v >= 0 for r in ROWS_X for v in r["shares"].values())
and not any(r["winner"].startswith("arbitrage") for r in ROWS_X))
# A battery in a maintenance phase takes no part in a run (R10).
ROWS_M, _, _cm = run_day("p3_ch_sell", hours=0.6, house_w=800.0,
soc0={"a": 60.0, "b": 30.0})
_lm = nth(ROWS_M, -1, "the last cycle")
_nm = _lm["now"] if present(_lm) else None
_tm = _cm._mode_roles(_nm, [types.SimpleNamespace(phase=_main.IDLE),
types.SimpleNamespace(phase="CHARGE")]) \
if _nm is not None else None
check("a battery in its monthly cycle is not a charger: no range, no charge "
"role, and no grid charge when it is the only price battery",
_tm is not None and B not in _tm.ranges
and _tm.roles.get(B) != modetrade.CHARGE and _tm.action is None)
# The headroom stopping a charge latches it off for the period (R8).
_cm2 = run_day("p3_ch_sell", hours=0.6, house_w=800.0,
soc0={"a": 60.0, "b": 30.0})[2]
_ok8 = False
if _nm is not None and _cm2.mode_tick is not None \
and _cm2.mode_tick.action == modetrade.CHARGE:
_q = int(_nm.timestamp() // 900)
_cm2.arb_headroom_w = 0.0
_t1 = _cm2._mode_roles(_nm, None)
_cm2.arb_headroom_w = 3000.0
_t2 = _cm2._mode_roles(_nm + timedelta(seconds=5), None)
_nq = datetime.fromtimestamp((_q + 1) * 900 + 5, timezone.utc)
_cm2.tariff_at = _nq
_t3 = _cm2._mode_roles(_nq, None)
_ok8 = (_t1.why == "headroom" and _cm2._charge_blocked_q in (_q, _q + 1)
and _t2.action is None and _t3.action == modetrade.CHARGE)
check("a charge the headroom stopped stays off for the rest of the period, "
"even when the headroom comes back, and may start in the next one", _ok8)
# ============================================================================ #
print("\n12. option A: no sale, and Leftover solar takes solar only (P4)")
ROWS_A, _, _ = run_day("p3_ch_sell", soc0={"a": 60.0, "b": 30.0},
choice={"a": (True, "solar_first", False),
"b": (True, "cheap_hours", True)})
print(f" Cheap hours B with a stored sell 'on': "
f"{sum(r['action'] == arbitrage.CHARGE for r in ROWS_A)} charging, "
f"{sum(r['action'] == arbitrage.DISCHARGE for r in ROWS_A)} selling")
check("with selling on in the store, a whole day has no sale and no sell "
"role, while Cheap hours only still charges from the grid",
all(r["action"] != arbitrage.DISCHARGE for r in ROWS_A)
and all(modetrade.SELL not in r["roles"].values() for r in ROWS_A)
and any(r["action"] == arbitrage.CHARGE for r in ROWS_A))
ROWS_L, _, _c12 = run_day("p3_ch_sell", soc0={"a": 60.0, "b": 30.0},
choice={"a": (True, "solar_first", False),
"b": (True, "leftover_solar", True)})
_lh = [r for r in ROWS_L if r["roles"].get(B) == modetrade.HOLD]
_lg = [r for r in ROWS_L if r["roles"].get(B) == modetrade.GIVE]
print(f" Leftover solar B: {len(_lh)} holding, {len(_lg)} giving, "
f"grid trades {sorted({str(r['action']) for r in ROWS_L})}")
check("Leftover solar never charges from the grid before the forecast: no "
"grid trade and no charge role all day, in the same cheap run",
all(r["action"] is None for r in ROWS_L)
and all(r["roles"].get(B) != modetrade.CHARGE for r in ROWS_L))
check("...it still saves its power and gives it in the expensive hours",
_lh and _lg and all(r["now"].astimezone(L).hour == 19 for r in _lg))
_pk = [modetrade.Pack("x", True, m, False, 2500.0, 30.0, 10.0)
for m in packmode.MODES]
check("...only Cheap hours only is a grid charger (packmode.GRID_CHARGE_MODES)",
[p.mode for p in modetrade.chargers(_pk, 95.0)]
== [packmode.CHEAP_HOURS])
_c12b = run_day("p3_ch_sell", start_h=0.75, hours=0.25,
soc0={"a": 60.0, "b": 30.0},
choice={"a": (True, "solar_first", False),
"b": (True, "leftover_solar", True)})[2]
_n12 = UTC0 + timedelta(hours=1)
_plan12 = homeview.mode_runs(_c12b, *homeview.local_day(_n12), _n12,
False)[0]
print(f" Leftover solar plan at 01:00: "
f"{[(r['from'], r['to'], r['kind']) for r in _plan12]}")
check("...and the plan draws no grid block for it, only its expensive run",
_plan12 and all(r["kind"] != "grid" for r in _plan12))
_enw, _nlw = homeview.uitext.words("en"), homeview.uitext.words("nl")
check("...and its help text promises no grid charging (EN and NL)",
all("cheap" not in t.lower() and "goedkop" not in t.lower()
for t in (_enw["home"]["batteries"]["choice"]["leftover_solar"]
["description"],
_nlw["home"]["batteries"]["choice"]["leftover_solar"]
["description"],
_enw["home"]["batteries"]["how"]["leftover_solar"]
["description"],
_nlw["home"]["batteries"]["how"]["leftover_solar"]
["description"])))
print()
if fails:
print(f"{len(fails)} FAILED:")
for f in fails:
print(f" - {f}")
sys.exit(1)
print("all checks passed")
+12 -8
View File
@@ -996,7 +996,8 @@ check("modes off: every configuration announces byte-for-byte what ed739d4 "
MODES = packmode.MODES
ON1, SUB1 = announce_p2(["inverter-a"], modes=MODES)
m, c, a = report("roster of one, modes on", ON1)
_opt = '["Solar first", "Leftover solar"]'
# P2 announced the first two; OPT-11 P3 appended "Cheap hours only".
_opt = '["Solar first", "Leftover solar", "Cheap hours only"]'
MODE_N1 = [
("homeassistant/select/goodwe_ctl_unit_inverter_a_mode/config",
'{"name": "Inverter A Mode", "object_id": "goodwe_inverter_a_mode", '
@@ -1062,16 +1063,19 @@ check("each HA mode option is the page's English name for that mode, in "
"both places the page names it",
all(mqtt.MODE_LABELS[k] == _en["home_words"]["modes"][k]
== _en["home"]["batteries"]["choice"][k]["name"] for k in MODES))
_later = dict(mqtt.MODE_LABELS, cheap_hours="Cheap hours only")
# OPT-11 P3 added `cheap_hours`: against P2's two modes (7a00f12), only the
# options list changes, by one entry at its end; no topic, id or label moves.
_p3 = mqtt.mode_payloads(NAMES2, MODES)
_orig = mqtt.MODE_LABELS
mqtt.MODE_LABELS = _later
mqtt.MODE_LABELS = {k: v for k, v in _orig.items() if k != "cheap_hours"}
try:
_p3 = mqtt.mode_payloads(NAMES2, MODES + ("cheap_hours",))
_p2 = mqtt.mode_payloads(NAMES2, MODES[:2])
finally:
mqtt.MODE_LABELS = _orig
_p2 = mqtt.mode_payloads(NAMES2, MODES)
check("...(simulated P3: adding `cheap_hours` changes only the options list, "
"by one entry at its end)",
check("...P3 appended `cheap_hours`: against P2's list only the options "
"change, by one entry at the end, and P2's two labels are unchanged",
MODES[:2] == ("solar_first", "leftover_solar")
and _p2[0][1]["options"] == ["Solar first", "Leftover solar"] and
[t for t, _ in _p3] == [t for t, _ in _p2]
and all({k: v for k, v in x.items() if k != "options"}
== {k: v for k, v in y.items() if k != "options"}
@@ -1261,7 +1265,7 @@ BAD = [
(f"{T_A}/use/set", b"true"), (f"{T_A}/use/set", b"1"),
(f"{T_A}/use/set", b""),
(f"{T_A}/mode/set", b"solar_first"), # a key, not the option
(f"{T_A}/mode/set", b"Cheap hours only"), # not offered before P3
(f"{T_A}/mode/set", b"Sell only"), # not an offered mode
(f"{T_A}/mode/set", b"solar first"),
(f"{T_A}/mode/set", "Sοlar first".encode()), # Greek omicron
(f"{T_A}/mode/set", b"Solar first\x00"),
+2 -2
View File
@@ -76,8 +76,8 @@ check("the mockup's six steps: meter, inverters, solar, prices, peak limit, "
and [WS["steps"][str(n)]["name"] for n in range(1, 7)]
== ["Your meter", "Inverters and batteries", "Solar", "Prices",
"Peak limit", "Check and start"])
check("battery modes are not a step yet (OPT-11): step 5 says they come later",
"later update" in WS["modes_later"]["description"]
check("battery modes are not a wizard step (OPT-11): step 5 points to the Batteries screen",
"Batteries screen" in WS["modes_later"]["description"]
and not any("mode" in k for k in BY_STEP[5]))
SPEC = uipage.setup_spec()
check("the spec the page reads is the table's",
+110
View File
@@ -0,0 +1,110 @@
#!/usr/bin/env python3
"""OPT-11 P3: with battery modes off, is every command byte-identical to
release/1.0 7a00f12, on the paths P3 changed?
python3 tools/opt11p3_identity.py # prints one digest per scenario
tools/safety79_identity.py's harness (the REAL Controller on a P1 meter, so
the peak guard runs) plus a day-ahead price series and `arbitrage_enabled`,
so the site trade P3 now skips with modes on (`arbitrage.decide()`, its
claim, the sizing to the guard's headroom) runs on every cycle, with a buy
period, a sell period and periods in between, and a house that goes over the
limit. Each row hashes safety79's fields plus the trade's action, watts and
sentence. Uses only the API 7a00f12 has, so the same file runs on both trees;
test_modes_p3.py part 1 holds this tree to the digests printed on 7a00f12.
"""
import functools
import hashlib
import os
import sys
import tempfile
from datetime import timedelta
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
import opt11_identity as rig # noqa: E402
import safety79_identity as s79 # noqa: E402
from app.price import PricePoint, PriceProduct, PriceRead, brussels # noqa
from app.static_tariff import TariffPlan # noqa: E402
DUR = 5400.0
def prices():
"""15-minute firm prices: the cheapest period 15-30 min into the run,
the dearest 75-90 min in, and ones in between."""
pts = []
for i in range(-8, 40):
s = rig.BASE + timedelta(minutes=15 * i)
p = 0.08 if i == 1 else 0.52 if i == 5 else 0.20 + 0.003 * (i % 7)
pts.append(PricePoint(start=brussels(s), eur_per_kwh=p,
provisional=False))
return TariffPlan(PriceRead.series(pts, PriceProduct.DAY_AHEAD,
"generic_dayahead", rig.BASE), None,
"provider")
PLAN = prices()
def _feed(c, t):
c.tariff_plan, c.tariff_at = PLAN, rig.BASE + timedelta(seconds=t)
s79._feed(c, t)
def load(t):
"""Under the limit, over it, a surplus, a big evening load."""
k = int(t // 600) % 5
return (900.0, 3200.0, -2200.0, 1500.0, 4200.0)[k]
def soc(u, t):
return 55.0 if u == "a" else 45.0
ARB = {"arbitrage_enabled": True, "cycle_cost_threshold_eur": 0.05}
SCENARIOS = {
"n1": dict(n=1),
"n2": dict(n=2),
# The option present and false: an install that saved it off.
"n1_off": dict(n=1, extra={"battery_modes_enabled": False}),
"n2_off": dict(n=2, extra={"battery_modes_enabled": False}),
}
def run(n, *, extra=None, dur=DUR):
opts = dict(s79.P1, **ARB, **(extra or {}))
trade = []
def on_cycle(c, t):
a = c.arb
trade.append(None if a is None else (a.action, a.power_w, a.reason))
db = os.path.join(tempfile.mkdtemp(prefix="p3id-"), "tariff.db")
real = rig.Controller
rig.Controller = functools.partial(real, peak_db_path=db)
try:
rows, c = rig.run(n, extra=opts, dur=dur, load=load, soc=soc,
cut=lambda u, t: False, before=_feed,
on_cycle=on_cycle)
finally:
rig.Controller = real
return [r + (x,) for r, x in zip(rows, trade)], c
def digest(rows) -> str:
return hashlib.sha256(repr(rows).encode()).hexdigest()
def digests() -> dict:
return {k: digest(run(**v)[0]) for k, v in SCENARIOS.items()}
if __name__ == "__main__":
if "--summary" in sys.argv:
for k, v in SCENARIOS.items():
rows = run(**v)[0]
acts = sorted({str(r[-1][0]) if r[-1] else "None" for r in rows})
print(f"{k}: {len(rows)} cycles, trade actions {acts}")
for k, v in digests().items():
print(f"{k} {v}")
+394
View File
@@ -0,0 +1,394 @@
#!/usr/bin/env python3
"""OPT-11 P3: replay days of recorded house load and real prices through the
REAL Controller (read_inputs + cycle, the arbiter, the allocator, the peak
guard on a P1 meter), with two modelled batteries, and report energy, cycles
and money, plus the P3 invariants on every tick.
python3 tools/opt11p3_replay.py --days 16 --variant p3_ls
python3 tools/opt11p3_replay.py --captured # 2026-09-09/10
The same file runs on release/1.0 7a00f12 (the P2 baseline) and on this
branch: it uses only the Controller API 7a00f12 has, and sets the batteries'
choices through the store keys P1 defined (`pack_sell` was reserved there).
REAL: the house (grid + battery power recorded 2026-09-02..17 on the live
install, scenarios/loop-2026-09-02_2026-09-17: the house's own net demand,
solar included), the prices (EPEX BE day-ahead, hourly, same weekday one year
earlier (364 days), or the captured quarter-hourly energy-charts bodies for
2026-09-09/10), the decision code.
MODELLED: the batteries (a first-order lag, 3 s, 800 W/s ramp, SoC
integrated, 95 % each way, a reading that jitters by a few watts, and
optionally a device limit or a refusal to charge), the all-in price (the EPEX file's own stated
assumption: (wholesale + 0.02 + 0.10) x 1.06), export paid at the wholesale
price, tomorrow's prices known from 13:00 local. The absolute money is
therefore an estimate; the comparison between variants is the point.
"""
import argparse
import asyncio
import gzip
import json
import math
import os
import sys
import time
from datetime import datetime, timedelta, timezone
HERE = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
sys.path.insert(0, HERE)
import app.main as m # noqa: E402
from app.main import Controller # noqa: E402
from app.p1 import make_sample, split_signed # noqa: E402
from app.price import PricePoint, PriceProduct, PriceRead, brussels # noqa
from app.static_tariff import TariffPlan # noqa: E402
from app.store import unit_key # noqa: E402
LOOP = os.path.join(HERE, "scenarios", "loop-2026-09-02_2026-09-17")
EPEX = os.path.join(HERE, "scenarios",
"epex-be-dayahead-2025-09-01_2026-09-01.json")
QH = os.path.join(HERE, "scenarios",
"energy-charts-be-price-2026-09-09_2026-09-10-twoday.json")
TICK = 5.0
CAP_KWH = 10.0
RATING = 2500.0
PEAK_CAP = 3500
def all_in(wholesale_kwh):
return round((wholesale_kwh + 0.02 + 0.10) * 1.06, 6)
def load_house(days):
"""[(t_utc_epoch, house_w)] from the recording: grid + battery power."""
out = []
for d in days:
def rd(kind):
f = os.path.join(LOOP, f"{d}-{kind}.jsonl.gz")
rows = []
with gzip.open(f, "rt") as fh:
for line in fh:
r = json.loads(line)
try:
v = float(r["v"])
except ValueError:
continue
rows.append((datetime.fromisoformat(r["t"]).timestamp(),
v))
return rows
g, b = rd("grid"), rd("batt")
j = 0
for t, gw in g:
while j + 1 < len(b) and b[j + 1][0] <= t:
j += 1
bw = b[j][1] if b and b[j][0] <= t else 0.0
out.append((t, gw + bw))
out.sort()
return out
def epex_points(shift_days=364):
d = json.load(open(EPEX))["prices"]
pts = []
for r in d:
s = datetime.fromisoformat(r["start"]) + timedelta(days=shift_days)
pts.append((s.astimezone(timezone.utc), 3600,
float(r["eur_per_kwh"])))
return pts
def qh_points():
d = json.load(open(QH))
return [(datetime.fromtimestamp(t, timezone.utc), 900, p / 1000.0)
for t, p in zip(d["unix_seconds"], d["price"])
if p is not None]
class Plant:
def __init__(self, soc, refuse_charge=False):
self.cmd, self.p, self.soc = 0.0, 0.0, float(soc)
self.refuse_charge = refuse_charge
def step(self, dt, floor):
c = self.cmd
if c > 0 and self.soc <= floor:
c = 0.0
if c < 0 and (self.soc >= 100.0 or self.refuse_charge):
c = 0.0
want = self.p + (c - self.p) * (1 - math.exp(-dt / 3.0))
dp = max(-800.0 * dt, min(800.0 * dt, want - self.p))
self.p += dp
e = self.p * dt / 3600.0 / 1000.0 # kWh, + = out
eff = 1 / 0.95 if e > 0 else 0.95
self.soc -= e * eff / CAP_KWH * 100.0
self.soc = max(0.0, min(100.0, self.soc))
class Hass:
def __init__(self, units, house_at, soc0=None, refuse=(), limits=None):
self.units = units
self.lim = limits or {}
self.plants = {u: Plant((soc0 or {}).get(u, 50.0), u in refuse)
for u in units}
self.house_at = house_at
self.t = 0.0
def grid(self):
return self.house_at(self.t) - sum(p.p for p in self.plants.values())
async def limits(self, e):
lo, hi = self.lim.get(e[-1], (-5000.0, 5000.0))
return (lo, hi, 1.0)
async def call(self, *a):
return True
async def number(self, e, invert=False):
if e == "sensor.grid":
return self.grid()
u = e[-1]
if e.startswith("sensor.soc"):
return round(self.plants[u].soc, 1)
if e.startswith("sensor.batt"):
# A real reading jitters by a few watts; a dead-flat one is a
# stale register to the add-on (SAFETY-37), which is not modelled.
j = (ord(u) * 31 + int(self.t)) % 7 - 3
return float(round(self.plants[u].p) + j)
return None
async def set_number(self, e, v):
self.plants[e[-1]].cmd = float(v)
return True
class _Store:
def __init__(self):
from app.store import DEFAULTS
self.data = dict(DEFAULTS)
def set(self, k, v):
self.data[k] = v
def get_time(self, k):
v = self.data.get(k)
return datetime.fromisoformat(v) if v else None
def set_time(self, k, when):
self.data[k] = when.isoformat() if when else None
class _Pub:
enabled = False
def publish(self, *a, **k):
pass
def close(self):
pass
VARIANTS = {
# (modes on, arbitrage_enabled, {unit: (use, mode, sell)})
"modes_off": (False, False, {}),
"arb_only": (False, True, {}),
"p2_sf_ls": (True, False, {"a": (True, "solar_first", False),
"b": (True, "leftover_solar", False)}),
"p2_arb": (True, True, {"a": (True, "solar_first", False),
"b": (True, "leftover_solar", False)}),
"p3_ls": (True, False, {"a": (True, "solar_first", False),
"b": (True, "leftover_solar", False)}),
"p3_ch_sell": (True, False, {"a": (True, "solar_first", False),
"b": (True, "cheap_hours", True)}),
"p3_both_ch": (True, False, {"a": (True, "cheap_hours", False),
"b": (True, "cheap_hours", True)}),
}
def options(modes, arb, n=2):
o = {"meter_entity": "sensor.grid", "meter_source": "ha_signed",
"p1_net_entity": "sensor.net", "meter_phases": 1,
"peak_cap_w": PEAK_CAP, "gain": 0.6, "slew_w": 400,
"target_grid_w": -10, "deadband_w": 15, "step_w": 10,
"heartbeat_s": 10, "stale_input_s": 15, "auto_start": True,
"battery_modes_enabled": modes, "arbitrage_enabled": arb,
"cycle_cost_threshold_eur": 0.10,
"inverters": [{"name": f"inverter-{u}",
"setpoint_entity": f"number.sp_{u}",
"soc_entity": f"sensor.soc_{u}",
"batt_entity": f"sensor.batt_{u}",
"max_w": int(RATING), "bms_soc_floor": 10,
"capacity_kwh": CAP_KWH} for u in "ab"[:n]]}
return o
def plan_at(now, pts):
"""The firm day-ahead points a site holds at `now`: today, and tomorrow
once published (13:00 local)."""
loc = brussels(now)
day0 = loc.replace(hour=0, minute=0, second=0, microsecond=0)
end = day0 + timedelta(days=2 if loc.hour >= 13 else 1)
sel = [PricePoint(start=brussels(s), eur_per_kwh=all_in(p),
provisional=False)
for s, span, p in pts if day0 <= s < end]
if len(sel) < 2:
return None
return TariffPlan(PriceRead.series(sel, PriceProduct.DAY_AHEAD,
"generic_dayahead", now), None,
"provider")
def run(variant, house, pts, start, dur, *, check=None, stride=1, n=2,
soc0=None, refuse=(), extra=None, choice=None, patch=None,
limits=None):
modes, arb, var_choice = VARIANTS[variant]
choice = var_choice if choice is None else choice
units = tuple("ab"[:n])
import bisect
ts = [h[0] for h in house]
def house_at(t):
i = bisect.bisect_right(ts, start.timestamp() + t) - 1
return house[max(0, i)][1]
wholesale = {s: (span, p) for s, span, p in pts}
starts = sorted(wholesale)
def price_at(now):
i = bisect.bisect_right(starts, now) - 1
if i < 0:
return None
s = starts[i]
span, p = wholesale[s]
return p if now < s + timedelta(seconds=span) else None
store = _Store()
for u, (use, mode, sell) in choice.items():
store.data[unit_key(f"inverter-{u}", "pack_use")] = use
store.data[unit_key(f"inverter-{u}", "pack_mode")] = mode
store.data[unit_key(f"inverter-{u}", "pack_sell")] = sell
stats = dict(imp_kwh=0.0, exp_kwh=0.0, cost=0.0, exp_rev=0.0,
out_kwh={u: 0.0 for u in "ab"[:n]},
in_kwh={u: 0.0 for u in "ab"[:n]},
ticks=0, quarters={}, trade_ticks=0)
async def go():
hass = Hass(units, house_at, soc0, refuse, limits)
import tempfile
db = os.path.join(tempfile.mkdtemp(prefix="p3r-"), "tariff.db")
c = Controller(dict(options(modes, arb, n), **(extra or {})), hass,
store, _Pub(), stop=asyncio.Event(), peak_db_path=db)
for i in c.inverters:
i.dev_min, i.dev_max = (limits or {}).get(i.name[-1],
(-5000.0, 5000.0))
if patch is not None:
patch(c)
c.auto = True
real_dt = m.datetime
clock = [start]
class _DT(real_dt):
@classmethod
def now(cls, tz=None):
return clock[0] if tz is None else clock[0].astimezone(tz)
m.datetime = _DT
plan_hour = None
try:
t = 0.0
while t < dur:
now = start + timedelta(seconds=t)
hass.t = t
clock[0] = now
if plan_hour != now.replace(minute=0, second=0,
microsecond=0):
plan_hour = now.replace(minute=0, second=0,
microsecond=0)
c.tariff_plan, c.tariff_at = plan_at(now, pts), now
else:
c.tariff_at = now
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()
for p in hass.plants.values():
p.step(TICK, 10.0)
g = hass.grid()
pr = price_at(now)
dt_h = TICK / 3600.0
if g > 0:
stats["imp_kwh"] += g / 1000 * dt_h
stats["cost"] += g / 1000 * dt_h * (all_in(pr) if pr
is not None else 0)
else:
stats["exp_kwh"] += -g / 1000 * dt_h
stats["exp_rev"] += -g / 1000 * dt_h * (pr or 0.0)
for u, p in hass.plants.items():
if p.p > 0:
stats["out_kwh"][u] += p.p / 1000 * dt_h
else:
stats["in_kwh"][u] -= p.p / 1000 * dt_h
q = int(now.timestamp() // 900)
acc = stats["quarters"].setdefault(q, [0.0, 0, False])
acc[0] += max(0.0, g)
acc[1] += 1
if (getattr(c.arb, "action", None) or "").endswith("CHARGE"):
acc[2] = True
stats["trade_ticks"] += 1
stats["ticks"] += 1
if check is not None:
check(c, hass, now, pr)
t += TICK
finally:
m.datetime = real_dt
return c
c = asyncio.run(go())
return stats, c
def summary(stats):
qs = stats["quarters"]
worst = max((a[0] / a[1] for a in qs.values()), default=0.0)
worst_trade = max((a[0] / a[1] for a in qs.values() if a[2]),
default=0.0)
cyc = {u: round(v / CAP_KWH, 2) for u, v in stats["out_kwh"].items()}
return {"import_kwh": round(stats["imp_kwh"], 2),
"export_kwh": round(stats["exp_kwh"], 2),
"import_eur": round(stats["cost"], 2),
"export_eur": round(stats["exp_rev"], 2),
"net_eur": round(stats["cost"] - stats["exp_rev"], 2),
"cycles": cyc, "worst_quarter_w": round(worst, 1),
"worst_charge_quarter_w": round(worst_trade, 1),
"charge_ticks": stats["trade_ticks"]}
LOOP_DAYS = [f"2026-09-{d:02d}" for d in range(2, 18)]
def main():
ap = argparse.ArgumentParser()
ap.add_argument("--variant", default="p3_ls")
ap.add_argument("--days", type=int, default=16)
ap.add_argument("--captured", action="store_true")
a = ap.parse_args()
if a.captured:
days = ["2026-09-09", "2026-09-10"]
pts = qh_points()
else:
days = LOOP_DAYS[:a.days]
pts = epex_points()
house = load_house(days)
start = datetime.fromisoformat(days[0] + "T00:00:00+02:00") \
.astimezone(timezone.utc)
dur = 86400.0 * len(days)
t0 = time.time()
stats, _c = run(a.variant, house, pts, start, dur)
out = summary(stats)
out["variant"], out["days"] = a.variant, len(days)
out["wall_s"] = round(time.time() - t0, 1)
print(json.dumps(out))
if __name__ == "__main__":
main()