Compare commits

..
12 Commits
Author SHA1 Message Date
glenn schrooyenandClaude a66a7f1a0f SAFETY-126 round 2: state what the 10 W tolerance costs; test the latch reset
checks / unit (pull_request) Successful in 18m11s
infra-pin / fresh (pull_request) Successful in 2s
Text only (DECISIONS, DOCS, CHANGELOG): a step load alarms on the same cycle
as before; a steadily rising load alarms a little later; with peak_margin_w 0
a period can end up to 10 W over the limit without this alarm, as for 'Load
too high' (SAFETY-81). Owner accepted 2026-10-31.

test_safety126.py part 2: after a real overrun ends, _peak_landing()'s latch
is reset and the night's later landings are quiet (review F3).

Co-Authored-By: Claude <noreply@anthropic.com>
2026-10-31 11:52:27 +01:00
glenn schrooyenandClaude 8e2463bc32 SAFETY-126: test reads through named_absence; points at MUTATION.md
checks / unit (pull_request) Successful in 19m25s
infra-pin / fresh (pull_request) Successful in 1s
Co-Authored-By: Claude <noreply@anthropic.com>
2026-10-31 01:54:16 +01:00
glenn schrooyenandClaude 40595e42ae SAFETY-126: a landing error on the aim is not an overrun
During the night grid charge under the peak reserve the guard alarmed
(ERROR PEAK_GUARD_FLOOR_LIMITED, "Battery stopped limiting the peak") and the
Batteries "Why" said "The peak limit needs power" at quarter ends, while every
quarter landed on the aim. The charge lands the period 0-7 W over the aim and
excess_w()'s 900 / remaining turns that into tens of kW in the last second;
with nothing able to give, check_peak_floor_limited() alarmed on any
wanted > 0.

Each cycle the guard wants power, _peak_landing() asks misses_aim(): with the
batteries at 0 W, does the period end more than ALARM_MISS_W over the aim? If
not, the decision is within_aim: no reserve alarm, no red peak row, no "Why"
peak. The claim and the commands are unchanged. A real overrun under the
reserve alarms on the same cycle as before.

Co-Authored-By: Claude <noreply@anthropic.com>
2026-10-31 01:26:02 +01:00
admin cc9bbeb73e Merge pull request 'OPT-27: the night cheap run is judged with the last fresh forecast' (#219) from OPT-27 into release/1.0
infra-pin / fresh (push) Successful in 2s
2026-10-30 22:16:32 +01:00
glenn schrooyenandClaude b9dea6d90d OPT-27 round 3: the kept-forecast wording states the 12 h limit (text only)
checks / unit (pull_request) Successful in 15m27s
infra-pin / fresh (pull_request) Successful in 1s
Co-Authored-By: Claude <noreply@anthropic.com>
2026-10-30 21:29:29 +01:00
glenn schrooyenandClaude 13b61d586a OPT-27 round 3: a kept figure is at most 12 h old; DST-safe midnight bound; pruning test
checks / unit (pull_request) Successful in 15m37s
infra-pin / fresh (pull_request) Successful in 1s
F1: the evening rule has an age limit (KEPT_MAX_AGE_S).
F2: the evening bound is the next local midnight (_bounds), right on 25 h and 23 h days.
F5: test that old days are dropped from the stored hold.

Co-Authored-By: Claude <noreply@anthropic.com>
2026-10-30 20:41:34 +01:00
glenn schrooyenandClaude Opus 5.5 307cb47caf OPT-27 review: the kept forecast line says "kept for tonight"
checks / unit (pull_request) Successful in 17m20s
infra-pin / fresh (pull_request) Successful in 1s
With battery modes off the forecast is still read and shown (Settings,
Simple, and the Forecast tab), but no plan uses it, so "used for tonight's
plan" was not true there (UX-53). "Kept for tonight" is true in every
shape. NL "vanavond" read wrong after midnight; "vannacht" covers the
whole night. DECISIONS' "modes off: nothing reads the sensors" corrected.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Bdqr7PiaHcBnzPituu2yTC
2026-10-30 19:59:29 +01:00
glenn schrooyenandClaude 60aa7f310c OPT-27: a kept figure sizes the night under OPT-28's shortfall, as a fresh one does
checks / unit (pull_request) Successful in 17m22s
infra-pin / fresh (pull_request) Successful in 1s
Co-Authored-By: Claude <noreply@anthropic.com>
2026-10-30 18:57:37 +01:00
glenn schrooyenandClaude bf842997c1 OPT-27 round 2: a kept tomorrow figure is bounded on both sides
Hold.stand_in(): day0 <= now < day0 + 3 h. A tomorrow sensor is asked with
day0 = the next midnight, so the upper bound alone was true all day and a
figure stood in at any hour. test_opt27.py covers daytime tomorrow outages.

Co-Authored-By: Claude <noreply@anthropic.com>
2026-10-30 18:54:41 +01:00
glenn schrooyenandClaude 4fde0b177c OPT-27: test_modes_p4 says what a covered sensor does; test_opt27 guards
test_modes_p4.py: 'the forecast is lost mid-run' now loses it for good (a hold
that never covers); a second scenario is the same loss with the hold, which
keeps Leftover solar charging. test_opt27.py: a failing check no longer hides
the ones after it.

Co-Authored-By: Claude <noreply@anthropic.com>
2026-10-30 18:54:41 +01:00
glenn schrooyenandClaude 50b567d6d1 OPT-27: the night plan keeps the last fresh solar forecast
Forecast.Solar's sensors stop updating in the evening, so at night every such
site had no forecast: the night cheap run needed 1.5x the cycle cost and
Leftover solar planned nothing. One missed poll also dropped it at once.

solarfc.Hold keeps the last fresh reading of each sensor, for the day it is
for. A missing or stale sensor is covered by it for up to FRESH_S (a short
outage), and from the end of the bell (20:00) until local midnight + 3 h
however old. A tomorrow figure is tomorrow's until midnight and then today's;
a today figure is only today's. Stored in the controller's store
(solar_fc_hold), so an evening restart keeps it. The forecast line says
'Solar forecast read 17:05, used for tonight's plan.' (EN, NL).

test_tariff24_weather.py runs with a hold that never covers: it tests what the
plan does when the forecast IS lost.

Co-Authored-By: Claude <noreply@anthropic.com>
2026-10-30 18:54:41 +01:00
admin f1b72443ff Merge pull request 'Sync 0.19.x into release/1.0: TARIFF-24 and SAFETY-125, with the forward-port fixes' (#220) from sync-0.19 into release/1.0
infra-pin / fresh (push) Successful in 2s
2026-10-30 18:51:53 +01:00
16 changed files with 1520 additions and 30 deletions
+2 -2
View File
@@ -42,12 +42,12 @@
SERIAL="test_loop.py test_freeze.py test_safety87.py"
#
# LONGEST FIRST (SAFETY-125's CI run, job 2580; then a local parallel run, seconds:
# modes_p4 758, safety125 508 on hotfix-0.19.2 only, modes_p3 364, safety119 328,
# modes_p4 758, safety126 634 (SAFETY-126, local, one core), safety125 508 on hotfix-0.19.2 only, modes_p3 364, safety119 328,
# price_run_fixes 184, safety90 160, safety103 134, safety94 114, peakguard 113,
# opt20 102, safety97 101, safety81 100). With JOBS workers pulling from one
# queue, starting the longest files first keeps the tail short: the batch takes
# about as long as the longest file. A sort order, not a selection.
HEAVY="test_modes_p4.py test_safety125.py test_modes_p3.py test_safety119.py test_price_run_fixes.py test_safety90.py test_safety103.py test_safety94.py test_peakguard.py test_opt20.py test_safety97.py test_safety81.py"
HEAVY="test_modes_p4.py test_safety126.py test_safety125.py test_modes_p3.py test_safety119.py test_price_run_fixes.py test_safety90.py test_safety103.py test_safety94.py test_peakguard.py test_opt20.py test_safety97.py test_safety81.py"
#
JOBS=${CI_JOBS:-6}
cd "${TESTS_DIR:-/w/goodwe-addon/goodwe_controller}"
+4
View File
@@ -2,6 +2,10 @@
## 0.19.2
### Not in a release yet
- The night grid charge no longer raises "Battery stopped limiting the peak" when the 15-minute period stays on the aim. A steadily rising load now alarms a little later, and with a peak margin of 0 a period can end up to 10 W over the limit without this alarm ([[SAFETY-126]]).
**A patch bump: one fix for charging from the grid under the peak limit, and no option changes.**
No option was added, removed or renamed, and no default changed (86 options).
**Existing installs need no option change:** options that start 0.19.1 also
@@ -0,0 +1,44 @@
# OPT-27 - the night plan keeps the last fresh forecast
**Status: done on `release/1.0`, not released.** Tests: `test_opt27.py`.
**The defect.** A today sensor is stale after 3 h (`FRESH_S`) or when it is from
before local midnight. Forecast.Solar stops updating in the evening, so at night
every Forecast.Solar site had no forecast: a price mode's night cheap run needed
1.5x the cycle cost (TARIFF-24 round 2) and Leftover solar planned nothing.
Separately, one missed poll (`unavailable`) dropped the forecast at once.
**Where: `solarfc`, not the plan.** The last fresh reading of each sensor is
kept in `solarfc.Hold` and `read_sum()` uses it when a sensor has none. A hold
at the plan (`_wx_track`/`weather_held`) can only keep the 1.5x gap off. The
day plan, Leftover solar's levels, Home and the Forecast tab all read
`Controller.solar_fc`, so one kept forecast makes control and every screen agree
(UX-53) with no second rule to keep in step. `weather_held()` is unchanged: it
still keeps a run under way when a read is lost.
**The rule** (`Hold.stand_in()`). A kept reading is for one day. It covers a
sensor that is missing or stale, never one with a bad value or unit, and:
- for up to `FRESH_S` when the sensor is missing: a short outage, any time of day;
- from the end of the bell (20:00 local, `BELL_TO_H`) until local midnight + 3 h
(`KEPT_UNTIL_H`), if the reading is at most `KEPT_MAX_AGE_S` (12 h) old when used;
- the day it is for: a today figure only on that day, a tomorrow figure as
tomorrow until midnight and, if the today sensors are still stale, as today's
until midnight + 3 h (all sensors of the sum are needed, as for any sum);
- until the next fresh reading, which replaces it at once.
So a figure is never used for another day and never older than 12 hours (`KEPT_MAX_AGE_S`): a 17:05 reading serves until 03:00 (9 h 55 min), a 09:05 reading only until 21:05.
Daytime is as before: a sensor that stopped is stale after 3 h, and no forecast
all day is still no forecast (the night run still needs 1.5x).
**Where it is stored.** The controller's store, key `solar_fc_hold`, so an
evening restart keeps it. It is written when a reading's figure changes or
every 15 min (`SAVE_EVERY_S`), not on every poll. A damaged entry is dropped.
**The screens.** `Forecast.kept` is true when a reading older than `FRESH_S` is
in use; the forecast line then reads "Solar forecast read 17:05, kept for
tonight." (`home_words.solar.kept`, EN and NL). The event log says
"read, kept from 17:05".
**Not changed.** Control with modes off: it does not use the forecast. The
screens still show the line, so it says "kept", not "used for the plan". A
forecast that stays fresh. The hold does not look at the weather, only at the clock and the day.
@@ -0,0 +1,47 @@
# SAFETY-126 - a landing error on the aim is not an overrun
**Status: FIXED on release/1.0 (after 0.19.2).** Tests: `test_safety126.py`.
**Cause.** A grid charge under the peak limit is sized to land the 15-minute
period on the aim (SAFETY-125), so the period ends 0-7 W over it. In the last
second `peakguard.excess_w()` multiplies that by 900 / remaining (3 000 at
0.3 s left, 90 000 at 0.01 s), so `decide()` wants tens of kW
(`wanted > 0`, peakguard.py `decide()`). With every battery under the peak
reserve that is floor-limited at 0 W, and `check_peak_floor_limited()`
alarmed on it at once: SAFETY-87 graded only the case where another battery
could give. With one battery able, the guard asked it for its rating for that
cycle and the Batteries "Why" said "The peak limit needs power". The
add-on's own charge is not the cause: `wanted` already counts it as
stoppable.
**Rule.** Each cycle the guard wants power, `Controller._peak_landing()`
asks `peakguard.misses_aim()`: with the batteries at 0 W from now on, does
the period end more than `ALARM_MISS_W` (10 W) over the aim? This is
`load_too_high()` with nothing to give, without `decide()`'s gain. If not, and
no real miss was seen since the guard took hold, the decision is
`within_aim`. Such a decision raises no reserve alarm (ERROR,
notification, red row), no red "Load too high" row, no "Why" peak and
no "Battery stopped limiting the peak" event hint. The claim, the floor and the
commands are unchanged.
**Still alarms.** House load the batteries cannot cover under the reserve:
the period would end over the aim with the batteries at 0 W. A step load
alarms on the same cycle as before. SAFETY-87's case and SAFETY-81's 'Load too
high' notification are unchanged.
**What the 10 W tolerance costs** (review, owner accepted 2026-10-31):
- A steadily rising load alarms a little later, once the projected miss passes
10 W (3 000 -> 4 000 W over 45 min: 56 s later). With the default 50 W
margin the alarm still comes before the limit.
- With `peak_margin_w` 0, a period can end up to 10 W over the limit without
this alarm, the tolerance 'Load too high' already accepts (SAFETY-81).
`misses_aim()` takes the house as meter plus batteries, so an export nets
off (`decide()`'s projection counts offtake only). A re-alarm while the loop's
draw on held batteries covers the house is gone too (test_safety126.py part 2,
modes off: 2 notifications, was 4).
**Not changed.** The one-cycle floor itself: at a period's last instant the
guard still asks an able battery for its rating, or stops a charge, for a
landing error. That is control, outside the alarm path. Raise a card if it
should go.
+2 -1
View File
@@ -557,6 +557,7 @@ With battery modes on, these choices replace the *Charging and selling on price*
- The plan is checked every 15 minutes. A planned level moves only by 1 kWh or 10 %, so a wobbling forecast does not flip it.
- No forecast, one older than 3 hours or from before midnight, or no house use yet: *Leftover solar* takes solar only and *Cheap hours only* fills to the top level. The Batteries screen says so.
- A battery needs its capacity (`inverters.capacity_kwh`) for a plan.
- **At night.** Some forecasts stop updating in the evening (Forecast.Solar does). The add-on keeps the last reading of each sensor and uses it for that night's plan: from 20:00 until 03:00, or until a newer reading comes. A sensor that is missing for up to 3 hours is covered the same way. A figure is never used for another day, and never when it is more than 12 hours old. The Batteries screen says "Solar forecast read 17:05, kept for tonight." With no reading all day, the night run still needs the wider gap.
**What always comes first.**
- The peak limit, the safety stops and the monthly cycle come before a battery's mode.
@@ -3457,7 +3458,7 @@ Charging from the grid stays blocked meanwhile. With several inverters the
notification names the battery at its limit (SAFETY-32), each graded against
its own cut-out level. That battery is held (weight 0 while the site gives)
and the guard's need goes to the others, up to their rating; the alarm fires
only if they miss the limit at full power for `ALARM_HOLD_S` (SAFETY-87). ⚠️ **The reserve binds the peak guard only
only if they miss the limit at full power for `ALARM_HOLD_S` (SAFETY-87). With no battery able to give it fires at once, but only if the period would end more than 10 W over the aim with the batteries at 0 W. A grid charge the add-on runs counts as one it can stop, and the last second's landing error on the aim is not an alarm. A steadily rising load therefore alarms a little later. With `peak_margin_w` 0 a period can end up to 10 W over the limit without this alarm, as for “Load too high” (SAFETY-126). ⚠️ **The reserve binds the peak guard only
(SAFETY-25).** Self-consumption has no such floor. See *What `peak_soc_floor`
does not protect*.
+2 -1
View File
@@ -371,7 +371,8 @@ def _day(c, k, d0, d1, now, fc, house, trust, levels, blocks, rows, shadow,
if fc is not None:
if k == 0:
line = hv.line_solar(fc.today.kwh, fc.today.kwh * trust,
hv.hhmm(fc.at))
hv.hhmm(fc.at),
kept=getattr(fc, "kept", False))
elif fc.tomorrow is not None:
line = hv.day_note("tomorrow_total",
kwh=hv._kwh(fc.tomorrow.kwh),
+11 -4
View File
@@ -568,7 +568,8 @@ def solar_block(c, now):
out["line"] = _t("solar." + st)
elif fc is not None:
out["line"] = line_solar(fc.today.kwh, out["count_kwh"],
out["updated"])
out["updated"],
kept=getattr(fc, "kept", False))
dp = getattr(c, "day_plan", None)
if st == "plan" and dp is not None and dp.uncovered_kwh >= MIN_KWH:
out["line"] += " " + line_uncovered(dp.uncovered_kwh)
@@ -817,9 +818,11 @@ def refusing(c, inv) -> int:
def peak_engaged(c) -> bool:
"""The peak guard asks the batteries for power this cycle. Not when it is
blind or floor-limited: those are engaged with a 0 W floor (no charging
only), so no battery gives power for it."""
only), so no battery gives power for it. Nor when the quarter ends on
the aim anyway (SAFETY-126, `within_aim`)."""
pg = getattr(c, "peak_guard", None)
return bool(pg is not None and pg.engaged and (pg.floor_w or 0.0) > 0.0)
return bool(pg is not None and pg.engaged and (pg.floor_w or 0.0) > 0.0
and not pg.within_aim)
def battery_rows(c, shadow, running, held=False):
@@ -1446,7 +1449,11 @@ def day_note(key, **values) -> str:
return _t("day." + key, **values)
def line_solar(fc_kwh, count_kwh, at) -> str:
def line_solar(fc_kwh, count_kwh, at, kept=False) -> str:
"""The forecast line. OPT-27: a reading kept past three hours (the
evening's, used at night) says so instead of "Updated"."""
if kept:
return _t("solar.kept", at=at)
return _t("solar.line", kwh=_kwh(fc_kwh), count=_kwh(count_kwh), at=at)
+50 -7
View File
@@ -208,6 +208,9 @@ EDGE_HOLD_S = _ctl.PLANT_SETTLE_S
# OPT-11 P4: the store key of the house profile (dayplan.HouseProfile), and
# how often the forecast sensors are read (they change hourly at most).
HOUSE_PROFILE_KEY = "house_profile"
# OPT-27: the forecast sensors' last fresh readings (`solarfc.Hold`), so an
# evening restart keeps the figure the night plan uses.
FC_HOLD_KEY = "solar_fc_hold"
FORECAST_READ_S = 60.0
@@ -947,6 +950,9 @@ class Controller:
# SAFETY-87: the reserve alarm's hold, as SAFETY-81's above.
self._reserve_batt_prev = None
self._reserve_high_since = None
# SAFETY-126: `_peak_landing()`'s last battery reading and latch.
self._land_batt_prev = None
self._peak_needed = False
self._maint_shape_suspended = False
# OPT-11. Battery modes, off by default so an install that has not
# turned them on behaves exactly as before (SAFETY-10 section 1). Each
@@ -995,6 +1001,8 @@ class Controller:
self.solar_fc_why = (solarfc.NOT_SET if self.fc_entity is None
else solarfc.MISSING)
self._fc_read_mono = None
self.fc_hold = solarfc.Hold(
(getattr(store, "data", None) or {}).get(FC_HOLD_KEY))
self.day_plan = None
self._plan_targets = None
self.house_profile = None
@@ -2984,6 +2992,27 @@ class Controller:
soc=min(float(p.soc) for p in able),
soc_floor=peakguard.binding_soc_floor(able, floor))
def _peak_landing(self, decision):
"""SAFETY-126: `decision` marked `within_aim` while the guard wants
power but the batteries at 0 W would end the quarter within
`ALARM_MISS_W` of the aim (`peakguard.misses_aim()`). Once a real miss
is seen, the need stands until the guard lets go. The claim and the
commands are unchanged; only the alarms and the "Why" read it.
DECISIONS/SAFETY-126-landing-error-is-not-an-overrun.md."""
batt, prev = self.site_batt_w, self._land_batt_prev
self._land_batt_prev = batt
if not (decision.engaged and decision.blind is None
and (decision.wanted_w or 0.0) > 0.0):
self._peak_needed = False
return decision
if batt is not None and prev is not None:
batt = min(float(batt), float(prev))
self._peak_needed = self._peak_needed or (
self._peak_proj is not None and peakguard.misses_aim(
self._peak_proj, batt_w=batt, target_w=decision.target_w))
return (decision if self._peak_needed
else dataclasses.replace(decision, within_aim=True))
def _peak_reserve(self) -> frozenset:
"""SAFETY-87: the batteries that give nothing for the peak this
cycle: under their reserve while the others carry a peak floor."""
@@ -3410,17 +3439,25 @@ class Controller:
now = datetime.now(timezone.utc)
fc, why, gaps = solarfc.read_sum(
list(zip(self.fc_entities, states[:n])),
list(zip(self.fc_tomorrow_entities, states[n:])), now)
list(zip(self.fc_tomorrow_entities, states[n:])), now,
hold=self.fc_hold)
if self.fc_hold.changed:
self.fc_hold.changed = False
self.store.set(FC_HOLD_KEY, self.fc_hold.dump())
# A member is named only in a sum of several; one entity reads as
# before. Logged once per episode: on a change, never every read.
gaps = tuple(g for g in gaps if len(
self.fc_entities if g[0] == "today"
else self.fc_tomorrow_entities) > 1)
was_kept = bool(self.solar_fc is not None and self.solar_fc.kept)
if (fc is None) != (self.solar_fc is None) \
or why != self.solar_fc_why or gaps != self.solar_fc_gaps:
or why != self.solar_fc_why or gaps != self.solar_fc_gaps \
or bool(fc is not None and fc.kept) != was_kept:
self.log_event(
"solar forecast: " + ("read" if fc is not None else
f"none ({why})")
"solar forecast: " + (
"none (" + str(why) + ")" if fc is None else
f"read, kept from {brussels(fc.at):%H:%M}" if fc.kept
else "read")
+ "".join(f", {d} {e} not usable ({w})"
for d, _i, e, w in gaps),
hint=messages.hint(
@@ -4832,6 +4869,7 @@ class Controller:
if self.peak_guard is not None:
self.peak_guard = self._peak_on_able(self.peak_guard,
maint_results)
self.peak_guard = self._peak_landing(self.peak_guard)
# SITE, and peakguard.claim() says so on the object it builds. The
# quantity is the 15-minute average OFFTAKE AT THE GRID CONNECTION,
# which is what the capacity tariff bills; there is one quarter-hour
@@ -4856,6 +4894,7 @@ class Controller:
self.log_event(_pg.explain(), hint=messages.hint(
"peak_off" if not _pg.engaged else
"peak_blind" if _pg.blind else
"peak_on" if _pg.within_aim else
"peak_reserve" if _pg.floor_limited else
"peak_saturated" if _pg.saturated else "peak_on"))
@@ -6153,6 +6192,9 @@ class Controller:
limited = self._reserve_short(g)
else:
self._reserve_batt_prev = self._reserve_high_since = None
# SAFETY-126: nothing can give, so it alarms at once, but only
# when the quarter really misses the aim (`_peak_landing()`).
limited = limited and not g.within_aim
was, self._peak_floor_limited = self._peak_floor_limited, limited
if not limited or was:
return
@@ -7030,12 +7072,13 @@ class Controller:
if _pg is None:
out.append({"ok": True, "warn": True, "id": "peak_guard",
"text": messages.row_peak_guard_waiting()})
elif _pg.saturated:
elif _pg.saturated and not _pg.within_aim:
out.append({"ok": False, "warn": False, "id": "peak_guard",
"runs": True,
"text": messages.row_peak_saturated(_pg.ceiling_w)})
elif _pg.floor_limited and (not (_pg.floor_w or 0.0) > 0.0
or self._peak_floor_limited):
elif _pg.floor_limited and (
not ((_pg.floor_w or 0.0) > 0.0 or _pg.within_aim)
or self._peak_floor_limited):
# SAFETY-24. NOT ok, like saturation: in both states the money
# is going to be spent and a human is the only actuator left. A
# separate row from saturation because the two ask for opposite
+17
View File
@@ -504,6 +504,11 @@ class PeakDecision:
# when they are not asked). See `place_on_able()`.
held: tuple = ()
able_w: float | None = None
# SAFETY-126: the guard wants power, but with the batteries at 0 W the
# quarter still ends within ALARM_MISS_W of the aim (`misses_aim()`), as
# for a landing error in its last second. Set by main.py. No alarm and no
# "Why" peak while it holds; the floor and the claim are unchanged.
within_aim: bool = False
def explain(self) -> str:
"""Idle: the status row's plain words (UX-08). Engaged: `reason`, the
@@ -997,6 +1002,18 @@ def load_too_high(projection, *, batt_w, ceiling_w, achievable_w) -> bool:
return ends_w - float(ceiling_w) > ALARM_MISS_W
def misses_aim(projection, *, batt_w, target_w) -> bool:
"""SAFETY-126. With the batteries at 0 W from now on, does the quarter
end more than `ALARM_MISS_W` over the aim (`target_w`)? `decide()`'s
`wanted > 0` asks the same question with no margin and with its gain, so a
1 W landing error in a quarter's last second reads as tens of kW. A grid
charge the add-on runs counts as stoppable: it is in `batt_w`. The caller
passes the lower of two cycles' battery readings, as for
`load_too_high()`. DECISIONS/SAFETY-126-landing-error-is-not-an-overrun.md."""
return load_too_high(projection, batt_w=batt_w, ceiling_w=target_w,
achievable_w=0.0)
def charge_headroom_w(decision) -> float | None:
"""SAFETY-45. The hardest this SITE may charge without putting the billed
quarter over the ceiling, in W, positive - or None when nothing measured it.
+182 -11
View File
@@ -28,6 +28,18 @@ A reading older than `FRESH_S`, or from before local midnight, is stale,
and a stale or missing forecast is no forecast: the day plan is off and
every mode acts as in P3 option A.
OPT-27: but the last fresh reading of each sensor is kept (`Hold`), so the
night plan is not made blind. Forecast.Solar's sensors stop changing in the
evening and api.forecast.solar is sometimes out for a poll. A kept reading
stands in for its sensor (a) for up to `FRESH_S` when the sensor is missing
or unavailable, as a sensor that merely stopped updating is already
tolerated, and (b) from the end of the bell (`BELL_TO_H`) until local
midnight + `KEPT_UNTIL_H`, if it is at most `KEPT_MAX_AGE_S` old when used.
Never beyond that, and never
for another day: a today figure is only today's, a tomorrow figure is
tomorrow's until midnight and then, until midnight + 3 h, today's.
DECISIONS/OPT-27-last-fresh-forecast-for-the-night.md.
OPT-17: each day may instead name a list (`pv_forecast_entities`,
`pv_forecast_tomorrow_entities`), summed by `read_sum()`. One member that is
not usable makes that day's sum unusable, never a partial sum.
@@ -50,6 +62,14 @@ FRESH_S = 3 * 3600.0
# The bell a state-only forecast is spread over, in local hours.
BELL_FROM_H, BELL_TO_H = 6.0, 20.0
# OPT-27: a kept reading serves the night plan from the end of the bell until
# this many hours after local midnight (the card: "local midnight + 3 h").
KEPT_UNTIL_H = 3.0
# OPT-27: and a reading older than this when used is not kept (a morning
# figure must not drive the night).
KEPT_MAX_AGE_S = 12 * 3600.0
SAVE_EVERY_S = 900.0
# Why there is no forecast (the page words are in ui_text `home_words`).
NOT_SET, MISSING, UNIT, STALE, BAD = (
"not_set", "missing", "unit", "stale", "bad")
@@ -75,6 +95,10 @@ class Forecast:
today: Day
tomorrow: Day | None
at: datetime # when the sensor last reported
# OPT-27: a reading kept past `FRESH_S` is in it (the evening's figure at
# night). The screens say so. Last, so `Forecast(today, tomorrow, at)` is
# as before.
kept: bool = False
def between(self, t0: datetime, t1: datetime,
since: datetime | None = None) -> float | None:
@@ -246,6 +270,110 @@ def _reported(state):
return max(ts) if ts else None
class Hold:
"""OPT-27: the last fresh reading of each forecast sensor, per day it is
for. `read_sum()` remembers every fresh reading here and, when a sensor
has none now, asks it (`stand_in()`).
Keyed `(kind, entity, day0)`: `kind` is "today" (energy still to come on
`day0`) or "tomorrow" (all of `day0`), and `day0` is the day the figure is
FOR. A today figure therefore never answers for another day, and a
tomorrow figure answers for its own day only: tomorrow until midnight,
then today until midnight + `KEPT_UNTIL_H`. Only the newest reading of a
key is kept, so the next fresh reading ends the hold at once.
`dump()` / `load()` are what the controller keeps in its store, so an
evening restart does not lose it (`load()` drops what is bad).
`changed` is set when `dump()` would differ."""
def __init__(self, data=None):
self._r = {}
self._saved = {} # key -> the `at` last flagged for saving
self.changed = False
if isinstance(data, dict):
self.load(data)
def remember(self, kind, entity, day: "Day", at: datetime) -> None:
key = (kind, entity, day.day0)
old = self._r.get(key)
if old is not None and at <= old[1]:
return
self._r[key] = (day, at)
# A sensor that reports every minute with the same figure must not
# rewrite the store every minute: the saved age may lag by
# `SAVE_EVERY_S` (the bounds are hours).
mark = self._saved.get(key)
if old is None or old[0].kwh != day.kwh or mark is None \
or (at - mark).total_seconds() >= SAVE_EVERY_S:
self._saved[key] = at
self.changed = True
def forget_before(self, day0: datetime) -> None:
"""Drop what is for a day before `day0` (the local midnight now)."""
gone = [k for k in self._r if k[2] < day0]
for k in gone:
del self._r[k]
self._saved.pop(k, None)
if gone:
self.changed = True
def stand_in(self, kind, entity, day0, now, *, as_today=False):
"""`(Day, at, kept)` for a sensor that has no fresh reading, or None.
The kept reading must be for `day0`, and either (a) at most `FRESH_S`
old, or (b) used at night: from the end of the bell to local midnight,
or within `KEPT_UNTIL_H` after it, and then at most `KEPT_MAX_AGE_S`
old. `kept` is true when (b) was needed,
or `as_today` (a tomorrow figure standing in for today, after
midnight)."""
rec = self._r.get((kind, entity, day0))
if rec is None:
return None
day, at = rec
if (now - at).total_seconds() <= FRESH_S and not as_today:
return day, at, False
if (now - at).total_seconds() > KEPT_MAX_AGE_S:
return None
# Both sides: a tomorrow sensor is asked with day0 = the next
# midnight, which is in the future all day (round 2).
after_midnight = day0 <= now < day0 + timedelta(hours=KEPT_UNTIL_H)
evening = _local_hour(now) >= BELL_TO_H and now < _bounds(day0)[1]
if as_today:
return (day, at, True) if after_midnight else None
return (day, at, True) if (evening or after_midnight) else None
def dump(self) -> dict:
return {"v": 1, "r": [
[kind, ent, d.day0.isoformat(), d.kwh, at.isoformat(),
[[s.isoformat(), e.isoformat(), k] for s, e, k in d.periods]]
for (kind, ent, _d0), (d, at) in sorted(
self._r.items(), key=lambda kv: (kv[0][0], kv[0][1],
kv[0][2]))]}
def load(self, data: dict) -> None:
rows = data.get("r") if isinstance(data, dict) else None
if not isinstance(rows, list):
return
for row in rows:
try:
kind, ent, d0, kwh, at, per = row
day0, at = _parse_time(d0), _parse_time(at)
kwh = float(kwh)
periods = tuple((_parse_time(s), _parse_time(e), float(k))
for s, e, k in per)
ok = (kind in ("today", "tomorrow") and isinstance(ent, str)
and day0 is not None and at is not None
and math.isfinite(kwh) and kwh >= 0
and all(s is not None and e is not None and e > s
and math.isfinite(k) and k >= 0
for s, e, k in periods))
except (TypeError, ValueError):
continue
if ok:
self._r[(kind, ent, day0)] = (Day(day0, kwh, periods), at)
self._saved[(kind, ent, day0)] = at
def entities(opts: dict, *, tomorrow=False) -> tuple:
"""The forecast entities one day names, in configured order: the list
if it has any entry, else the single key, else (). A repeated id is
@@ -282,14 +410,14 @@ def _today(state, day0, now):
def _tomorrow(state, day0, day1, now):
"""`(Day, None)` or `(None, why)` for one tomorrow sensor."""
"""`(Day, at, None)` or `(None, None, why)` for one tomorrow sensor."""
t, why = _day(state, day1, now, today=False)
if t is None:
return None, why
return None, None, why
t_at = _reported(state)
if t_at is None or t_at < day0 or (now - t_at).total_seconds() > FRESH_S:
return None, STALE
return t, None
return None, None, STALE
return t, t_at, None
def read(today_state, tomorrow_state, now: datetime, *,
@@ -305,7 +433,7 @@ def read(today_state, tomorrow_state, now: datetime, *,
return None, why
tomorrow = None
if tomorrow_state is not None:
tomorrow, _w = _tomorrow(tomorrow_state, day0, day1, now)
tomorrow, _at, _w = _tomorrow(tomorrow_state, day0, day1, now)
return Forecast(today, tomorrow, at), None
@@ -321,7 +449,12 @@ def _sum(days):
return Day(days[0].day0, sum(d.kwh for d in days), periods)
def read_sum(today, tomorrow, now: datetime):
# Why a sensor may be covered by a kept reading: it has none now, or it is
# too old. A bad value or a wrong unit is a set-up fault and is never covered.
_COVERABLE = (MISSING, STALE)
def read_sum(today, tomorrow, now: datetime, hold: Hold | None = None):
"""OPT-17: `(Forecast or None, why, gaps)` over lists of
`(entity, state)`, one per configured member, in configured order.
@@ -334,29 +467,67 @@ def read_sum(today, tomorrow, now: datetime):
`gaps` is `((day, position, entity, why), ...)` for every unusable
member (`day` "today" or "tomorrow", `position` 1-based), so the page
and the log can name it. With one member per day the forecast is what
`read()` gives (test_opt17.py proves it)."""
`read()` gives (test_opt17.py proves it).
OPT-27: with a `hold`, every fresh reading is remembered in it, and a
member that is missing or stale is covered by its last fresh reading
where `Hold.stand_in()` allows it (the forecast then has `kept` set when
that took a reading older than `FRESH_S`). After midnight, if the today
side is still short, the tomorrow figures read before midnight are
today's, until midnight + `KEPT_UNTIL_H`. Without a `hold` this is
exactly as before."""
if not today:
return None, NOT_SET, ()
day0, day1 = _bounds(now)
gaps, days, ats, why0 = [], [], [], None
if hold is not None:
hold.forget_before(day0)
gaps, days, ats, why0, kept = [], [], [], None, False
for i, (ent, st) in enumerate(today, 1):
d, at, why = _today(st, day0, now)
if d is not None and hold is not None:
hold.remember("today", ent, d, at)
elif hold is not None and why in _COVERABLE:
r = hold.stand_in("today", ent, day0, now)
if r is not None:
d, at, k = r
kept = kept or k
if d is None:
gaps.append(("today", i, ent, why))
why0 = why0 or why
else:
days.append(d)
ats.append(at)
tdays = []
tdays, tats, tkept = [], [], False
for i, (ent, st) in enumerate(tomorrow, 1):
t, why = _tomorrow(st, day0, day1, now)
t, t_at, why = _tomorrow(st, day0, day1, now)
if t is not None and hold is not None:
hold.remember("tomorrow", ent, t, t_at)
elif hold is not None and why in _COVERABLE:
r = hold.stand_in("tomorrow", ent, day1, now)
if r is not None:
t, t_at, k = r
tkept = tkept or k
if t is None:
gaps.append(("tomorrow", i, ent, why))
else:
tdays.append(t)
tats.append(t_at)
if why0 is not None and hold is not None and tomorrow and all(
g[3] in _COVERABLE for g in gaps if g[0] == "today"):
# Past midnight the sensors may still show yesterday's "left today".
# The figure each tomorrow sensor gave before midnight is for this
# day; all of them are needed, as for any sum.
subs = [hold.stand_in("tomorrow", ent, day0, now, as_today=True)
for ent, _st in tomorrow]
if all(r is not None for r in subs):
days = [r[0] for r in subs]
ats = [r[1] for r in subs]
gaps = [g for g in gaps if g[0] != "today"]
why0, kept = None, True
gaps = tuple(gaps)
if why0 is not None:
return None, why0, gaps
tday = _sum(tdays) if tomorrow and len(tdays) == len(tomorrow) else None
kept = kept or (tday is not None and tkept)
# The oldest member's report is when the sum was last fully known.
return Forecast(_sum(days), tday, min(ats)), None, gaps
return Forecast(_sum(days), tday, min(ats), kept), None, gaps
+1 -1
View File
@@ -360,7 +360,7 @@
"tops_up": "It charges from the grid to {pct}. That covers {kwh} of expensive hours the sun will not.",
"no_forecast": "No solar forecast, so it charges from solar only."
},
"solar": {"line": "Solar still to come today: {kwh}, we count on {count}. Updated {at}.", "uncovered": "The expensive hours need more than Leftover solar can hold. The rest, {kwh}, comes from another battery or the grid.", "no_forecast": "No solar forecast now, so Leftover solar charges from solar only.", "learning": "The add-on is learning your house's use. The plan starts after one full day."},
"solar": {"line": "Solar still to come today: {kwh}, we count on {count}. Updated {at}.", "uncovered": "The expensive hours need more than Leftover solar can hold. The rest, {kwh}, comes from another battery or the grid.", "kept": "Solar forecast read {at}, kept for tonight.", "no_forecast": "No solar forecast now, so Leftover solar charges from solar only.", "learning": "The add-on is learning your house's use. The plan starts after one full day."},
"today": {
"not_planned": "Not planned yet.",
"no_runs": "No cheap or expensive hours left today.",
+1 -1
View File
@@ -300,7 +300,7 @@
"tops_up": "Ze laadt van het net tot {pct}. Dat dekt {kwh} dat de zon in dure uren niet levert.",
"no_forecast": "Geen zonneprognose, dus ze laadt alleen met zonne-energie."
},
"solar": {"line": "Nog verwachte zon vandaag: {kwh}, we rekenen op {count}. Bijgewerkt om {at}.", "uncovered": "De dure uren vragen meer dan Overschot zonne-energie kan opslaan. De rest, {kwh}, komt van een andere batterij of het net.", "no_forecast": "Nu geen zonneprognose, dus Overschot zonne-energie laadt alleen met zon.", "learning": "De add-on leert eerst het verbruik van uw huis. Het plan start na één volle dag."},
"solar": {"line": "Nog verwachte zon vandaag: {kwh}, we rekenen op {count}. Bijgewerkt om {at}.", "uncovered": "De dure uren vragen meer dan Overschot zonne-energie kan opslaan. De rest, {kwh}, komt van een andere batterij of het net.", "kept": "Zonneprognose van {at}, bewaard voor vannacht.", "no_forecast": "Nu geen zonneprognose, dus Overschot zonne-energie laadt alleen met zon.", "learning": "De add-on leert eerst het verbruik van uw huis. Het plan start na één volle dag."},
"today": {
"not_planned": "Nog niet gepland.",
"no_runs": "Vandaag geen goedkope of dure uren meer.",
+25 -1
View File
@@ -672,9 +672,24 @@ check("Cheap hours with a sun forecast charges as with none (P3)",
and top > 80.0)
# The forecast goes away mid-run: Leftover solar stops at once (P3 option A).
# OPT-27: a sensor that is unavailable is covered by its last fresh reading
# for up to 3 h (`solarfc.Hold`), so "lost" here is a hold that never covers;
# the next scenario is the same loss with the hold (test_opt27.py).
DROP = UTC0 + timedelta(hours=1, minutes=30)
class _NoHold(solarfc.Hold):
def stand_in(self, *a, **k):
return None
def seed_profile_no_hold(c):
seed_profile(c)
c.fc_hold = _NoHold()
rows, st_, c = run_day(SF_LS, fc=lambda now: 1.0 if now < DROP else None,
soc0={"a": 60.0, "b": 10.0})
soc0={"a": 60.0, "b": 10.0}, patch=seed_profile_no_hold)
invariants("forecast lost mid-run", rows, c)
before = [r for r in rows if r["now"] < DROP and r["action"]
== arbitrage.CHARGE]
@@ -682,6 +697,15 @@ late = [r for r in rows if r["now"] > DROP + timedelta(
seconds=2 * 60 + 10) and r["action"] == arbitrage.CHARGE]
check("forecast lost mid-run: Leftover solar was charging, and stops within "
"a forecast read (60 s) and a cycle", before and not late)
rows, st_, c = run_day(SF_LS, fc=lambda now: 1.0 if now < DROP else None,
soc0={"a": 60.0, "b": 10.0})
invariants("forecast unavailable mid-run, covered", rows, c)
check("forecast unavailable mid-run (OPT-27): the last reading covers it, so "
"Leftover solar goes on charging past the point where it was lost",
[r for r in rows if r["now"] < DROP and r["action"]
== arbitrage.CHARGE]
and [r for r in rows if r["now"] > DROP + timedelta(
seconds=2 * 60 + 10) and r["action"] == arbitrage.CHARGE])
# The forecast is far too high (40 kWh, and there is no sun at all): no grid
# charge, so B covers less of the evening - a money risk, never a safety one.
+646
View File
@@ -0,0 +1,646 @@
"""Runnable check for OPT-27. `python3 test_opt27.py`
Forecast.Solar's sensors stop changing in the evening, so at night every
Forecast.Solar site was "no forecast": a price mode's night cheap run needed
1.5x the cycle cost and Leftover solar bought nothing. And one missed poll
(`unavailable`) dropped the forecast at once. `solarfc.Hold` now keeps the last
fresh reading of each sensor, for the day it is FOR:
* up to `FRESH_S` when the sensor is missing or unavailable (a short outage);
* from the end of the bell (20:00) until local midnight + 3 h, if the reading
is at most `KEPT_MAX_AGE_S` (12 h) old when used (the night plan); never
beyond, never for another day.
DECISIONS/OPT-27-last-fresh-forecast-for-the-night.md.
1. `solarfc` alone: the evening, the outage, the bounds, the day a figure is
for, restart (dump/load), and no hold = as before.
2. The real Controller on a day/night tariff (0.34/0.26, gap 0.08, cycle cost
0.07), forecast sensors that stop updating at 17:05 local:
a. the 22:00 night run is planned with the cycle-cost gap and charges;
b. no forecast all day: the night run still needs 1.5x and does not charge;
c. an evening restart keeps it (a new Controller on the saved store);
d. a one-hour outage at 18:47 does not drop it;
e. past midnight + 3 h the figure is not used, before it it is.
3. The screens: the forecast line says when it was read and that it is kept
for tonight (EN and NL), and Home agrees with control.
Mutations (MUTATION.md, purged copies, graded over this file): see the
DECISIONS note and the dev report.
"""
import asyncio
import json
import logging
import os
import sys
import tempfile
import time
from datetime import datetime, timedelta, timezone
HERE = os.path.dirname(os.path.abspath(__file__))
sys.path.insert(0, HERE)
sys.path.insert(0, os.path.join(HERE, "tools"))
logging.disable(logging.WARNING)
import opt11p3_replay as rp # noqa: E402
from named_absence import first, nth, present # noqa: E402
import app.main as m # noqa: E402
from app import homeview, modetrade, solarfc # noqa: E402
from app.main import Controller # noqa: E402
from app.p1 import make_sample, split_signed # 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=1)) # Brussels in winter (CET)
def at(day, hh, mm=0):
"""Local time on 2026-11-`day` (CET) as UTC."""
return datetime(2026, 11, day, hh, mm, tzinfo=L).astimezone(timezone.utc)
def st(kwh, when, *, unit="kWh"):
return {"state": f"{kwh}", "attributes": {"unit_of_measurement": unit},
"last_updated": when.isoformat(),
"last_reported": when.isoformat()}
GONE = {"state": "unavailable", "attributes": {}}
T_ENT, M_ENT = "sensor.fc_today", "sensor.fc_tomorrow"
# --------------------------------------------------------------------------- #
print("1. solarfc: the hold")
D1_READ = at(3, 17, 5) # the last fresh reading, 17:05
EVE = at(3, 22, 0)
def rs(today, tomorrow, now, hold):
return solarfc.read_sum([(T_ENT, today)], [(M_ENT, tomorrow)], now, hold)
h = solarfc.Hold()
fc, why, gaps = rs(st(0.4, D1_READ), st(12.0, D1_READ), at(3, 17, 10), h)
check("a fresh reading is a forecast, not kept", fc is not None
and not fc.kept and fc.tomorrow.kwh == 12.0 and fc.today.kwh == 0.4)
check("it is remembered (the hold is flagged for saving)", h.changed)
fc_no, why_no, _g = rs(st(0.4, D1_READ), st(12.0, D1_READ), EVE, None)
check("with no hold, 22:00 is no forecast (as before)",
fc_no is None and why_no == solarfc.STALE)
fc, why, gaps = rs(st(0.4, D1_READ), st(12.0, D1_READ), EVE, h)
print(f" 22:00, sensors still show 17:05: {fc and fc.kept=}, {gaps=}")
check("22:00 with the sensors stale since 17:05: the 17:05 figures are used",
fc is not None and fc.kept and fc.tomorrow.kwh == 12.0
and fc.at == D1_READ and gaps == ())
check("... today has no sun left in it, tomorrow's is counted from 06:00",
fc is not None and fc.between(EVE, at(4, 7, 0)) is not None
and 0.0 < fc.between(EVE, at(4, 7, 0)) < 0.5
and fc.today.kwh == 0.4)
fc2, _w, _g = rs(GONE, GONE, EVE, h)
check("the same when the sensors are unavailable at 22:00",
fc2 is not None and fc2.kept and fc2.tomorrow.kwh == 12.0)
# A new fresh reading replaces the old one at once.
fc3, _w, _g = rs(st(0.0, at(3, 22, 20)), st(14.5, at(3, 22, 20)),
at(3, 22, 30), h)
check("a newer fresh reading is used and is not 'kept'",
fc3 is not None and not fc3.kept and fc3.tomorrow.kwh == 14.5)
fc3b, _w, _g = rs(st(0.4, D1_READ), st(12.0, D1_READ), at(3, 23, 0), h)
check("... and later the newest reading is the one kept, not the old one",
fc3b is not None and fc3b.tomorrow.kwh == 14.5)
# A short outage: unavailable for one or two polls, any time of day.
h2 = solarfc.Hold()
rs(st(1.1, at(3, 18, 5)), st(12.0, at(3, 18, 5)), at(3, 18, 6), h2)
for label, now in (("18:47", at(3, 18, 47)), ("19:20", at(3, 19, 20)),
("19:47", at(3, 19, 47)), ("20:55", at(3, 20, 55))):
fo, _w, _g = rs(GONE, GONE, now, h2)
check(f"unavailable at {label}: the 18:05 forecast is still used",
fo is not None and fo.tomorrow.kwh == 12.0)
fo, _w, _g = rs(GONE, GONE, at(3, 20, 55), solarfc.Hold())
check("(nothing was ever read: still no forecast)", fo is None)
# Daytime is as before: an outage longer than FRESH_S drops it.
h3 = solarfc.Hold()
rs(st(6.0, at(4, 9, 5)), st(12.0, at(4, 9, 5)), at(4, 9, 6), h3)
d_in = rs(GONE, GONE, at(4, 11, 50), h3)[0]
d_out = rs(GONE, GONE, at(4, 12, 10), h3)[0]
d_out2 = rs(st(6.0, at(4, 9, 5)), st(12.0, at(4, 9, 5)), at(4, 14, 0), h3)[0]
check("daytime: an outage under 3 h is covered, one over 3 h is not",
d_in is not None and not d_in.kept and d_out is None)
check("daytime: a sensor that stopped is stale after 3 h, as before "
"(the hold does not start before the bell ends)", d_out2 is None)
# Round 2: today fresh, tomorrow's sensor stopped or down in the daytime.
h9 = solarfc.Hold()
rs(st(5.0, at(4, 9, 5)), st(12.0, at(4, 9, 5)), at(4, 9, 6), h9)
for label, tm in (("stale", st(12.0, at(4, 9, 5))), ("unavailable", GONE)):
fd = rs(st(3.0, at(4, 13, 55)), tm, at(4, 14, 0), h9)[0]
check(f"14:00, today fresh, tomorrow {label} since 09:05 (over 3 h): "
"no tomorrow, nothing kept",
fd is not None and fd.tomorrow is None and not fd.kept)
fn = rs(st(0.1, at(4, 20, 55)), tm, at(4, 21, 0), h9)[0]
check(f"21:00, tomorrow {label}: the 09:05 figure covers it, kept",
fn is not None and fn.tomorrow is not None
and fn.tomorrow.kwh == 12.0 and fn.kept)
fs = rs(st(3.0, at(4, 10, 55)), tm, at(4, 11, 0), h9)[0]
check(f"11:00, tomorrow {label} for under 3 h: still covered, not kept",
fs is not None and fs.tomorrow is not None and not fs.kept)
# The bound: midnight + 3 h, and the day a figure is for.
fm = rs(st(0.0, at(3, 23, 0)), st(0.0, at(3, 23, 0)), at(4, 1, 0), h2)[0]
check("00:30 is covered by the 18:05 tomorrow figure, as TODAY's",
rs(GONE, GONE, at(4, 0, 30), h2)[0] is not None)
fmid = rs(GONE, GONE, at(4, 2, 59), h2)[0]
print(f" 02:59: {fmid and (fmid.today.kwh, fmid.tomorrow, fmid.kept)}")
check("02:59: tomorrow's 12.0 kWh is now today's, there is no tomorrow",
fmid is not None and fmid.today.kwh == 12.0 and fmid.tomorrow is None
and fmid.kept and fmid.today.day0 == at(4, 0, 0))
check("... its sun is counted on the day it is for (06:00-20:00 on the 4th)",
fmid is not None and fmid.between(at(4, 3, 0), at(4, 7, 0)) is not None
and fmid.between(at(4, 3, 0), at(4, 7, 0)) > 0.0)
check("03:01 (midnight + 3 h): not used any more",
rs(GONE, GONE, at(4, 3, 1), h2)[0] is None)
check("... and not the next evening either (past midnight + 3 h)",
rs(GONE, GONE, at(4, 22, 0), h2)[0] is None)
fe = rs(st(0.1, at(4, 21, 50)), GONE, at(4, 22, 0), h2)[0]
check("... the next evening, with today fresh and tomorrow's sensor down, "
"there is no tomorrow: the 4th's figure is not the 5th's",
fe is not None and fe.today.kwh == 0.1 and fe.tomorrow is None)
# Never today's figure for tomorrow.
h4 = solarfc.Hold()
rs(st(3.0, at(3, 17, 0)), st(12.0, at(3, 17, 0)), at(3, 17, 5), h4)
# Next day 00:30: today's sensor is fresh again (4.0), tomorrow's is gone.
fx = rs(st(4.0, at(4, 0, 20)), GONE, at(4, 0, 30), h4)[0]
check("00:30: today is the fresh 4.0, and there is NO tomorrow "
"(the 3rd's 'today' 3.0 and 'tomorrow' 12.0 are not the 5th's)",
fx is not None and fx.today.kwh == 4.0 and fx.tomorrow is None
and not fx.kept)
fy = rs(st(4.0, at(4, 0, 20)), st(9.0, at(4, 0, 20)), at(4, 0, 30), h4)[0]
check("00:30 with both fresh: tomorrow is the fresh 9.0, not the old 12.0",
fy is not None and fy.tomorrow.kwh == 9.0 and not fy.kept)
# A tomorrow sensor that is down while today's is fresh, in the evening: the
# hold gives the tomorrow figure for tomorrow, never today's figure.
h5 = solarfc.Hold()
rs(st(3.0, at(3, 17, 0)), st(12.0, at(3, 17, 0)), at(3, 17, 5), h5)
fz = rs(st(0.2, at(3, 21, 30)), GONE, at(3, 21, 40), h5)[0]
check("21:40: today fresh, tomorrow down: tomorrow is the kept 12.0 (not "
"today's 0.2 or 3.0)", fz is not None and fz.tomorrow.kwh == 12.0
and fz.today.kwh == 0.2 and fz.kept)
# A set-up fault is never covered.
h6 = solarfc.Hold()
rs(st(3.0, at(3, 17, 0)), st(12.0, at(3, 17, 0)), at(3, 17, 5), h6)
fb = rs(st("lots", at(3, 21, 30)), st(12.0, at(3, 21, 30)),
at(3, 21, 40), h6)
fu = rs(st(0.2, at(3, 21, 30), unit="W"), st(12.0, at(3, 21, 30)),
at(3, 21, 40), h6)
check("a bad value or a wrong unit is not covered by the hold",
fb[0] is None and fb[1] == solarfc.BAD and fu[0] is None
and fu[1] == solarfc.UNIT)
# Several sensors: all or nothing.
h7 = solarfc.Hold()
two_t = [("sensor.t1", st(1.0, at(3, 17, 0))),
("sensor.t2", st(2.0, at(3, 17, 0)))]
two_m = [("sensor.m1", st(5.0, at(3, 17, 0))),
("sensor.m2", st(6.0, at(3, 17, 0)))]
solarfc.read_sum(two_t, two_m, at(3, 17, 5), h7)
gone2 = [("sensor.t1", GONE), ("sensor.t2", GONE)]
f2 = solarfc.read_sum(gone2, [("sensor.m1", GONE), ("sensor.m2", GONE)],
at(3, 22, 0), h7)
check("three or two sensors: the kept figures are summed",
f2[0] is not None and f2[0].tomorrow.kwh == 11.0
and f2[0].today.kwh == 3.0 and f2[2] == ())
h8 = solarfc.Hold()
solarfc.read_sum(two_t[:1] + [("sensor.t2", GONE)], two_m, at(3, 17, 5), h8)
f3 = solarfc.read_sum(gone2, [("sensor.m1", GONE), ("sensor.m2", GONE)],
at(3, 22, 0), h8)
check("one sensor never read fresh: no partial sum, the gap is named",
f3[0] is None and any(g[:3] == ("today", 2, "sensor.t2") for g in f3[2]))
# Restart: what the store holds comes back.
saved = json.loads(json.dumps(h.dump()))
hr = solarfc.Hold(saved)
fr = rs(GONE, GONE, at(3, 23, 0), hr)[0]
check("dump/load through JSON: the kept figures come back (40 min old: "
"inside FRESH_S, so not 'kept' in the screens' sense)",
fr is not None and not fr.kept and fr.tomorrow.kwh == 14.5
and fr.at == at(3, 22, 20))
junk = solarfc.Hold({"r": [["today", 7, "x", "y", "z", []],
["tomorrow", T_ENT, "2026-11-04T00:00:00+00:00",
-1, "2026-11-03T16:00:00+00:00", []],
"bad", None]})
check("a damaged store entry is dropped, not raised",
rs(GONE, GONE, at(3, 22, 0), junk)[0] is None
and solarfc.Hold(None).dump()["r"] == [] and solarfc.Hold(5).changed
is False)
hp = solarfc.Hold()
rs(st(8.0, at(3, 9, 0), ), st(12.0, at(3, 9, 0)), at(3, 9, 5), hp)
periods = [{"period_start": f"2026-11-04T{hh:02d}:00:00+01:00",
"pv_estimate": 1.0} for hh in range(8, 16)]
s_per = st(10.0, at(3, 17, 0))
s_per["attributes"]["detailedForecast"] = periods
rs(st(0.0, at(3, 17, 0)), s_per, at(3, 17, 5), hp)
hq = solarfc.Hold(json.loads(json.dumps(hp.dump())))
fq = rs(GONE, GONE, at(3, 22, 0), hq)[0]
check("an hourly split (Solcast shape) survives the store",
fq is not None and fq.tomorrow.periods
and abs(fq.between(at(4, 8, 0), at(4, 16, 0)) - 7.5) < 1e-6)
# Round 3. F1: a kept reading is at most KEPT_MAX_AGE_S old when used.
hz = solarfc.Hold()
rs(st(0.4, at(3, 0, 5)), st(12.0, at(3, 0, 5)), at(3, 0, 10), hz)
print(f" 00:05 reading, sensors dead: 19:59 "
f"{rs(GONE, GONE, at(3, 19, 59), hz)[0]}, 20:00 "
f"{rs(GONE, GONE, at(3, 20, 0), hz)[0]}")
check("F1: a 00:05 reading with the sensors dead all day is not kept at "
"20:00 (20 h old): no forecast",
rs(GONE, GONE, at(3, 20, 0), hz)[0] is None
and rs(GONE, GONE, at(3, 23, 0), hz)[0] is None)
check("... nor after midnight until 03:00 (as today's, 26 h old)",
rs(GONE, GONE, at(4, 0, 30), hz)[0] is None
and rs(GONE, GONE, at(4, 2, 30), hz)[0] is None)
hy = solarfc.Hold()
rs(st(0.4, at(3, 17, 5)), st(12.0, at(3, 17, 5)), at(3, 17, 10), hy)
check("F1: a 17:05 reading is kept at 22:00 (5 h) ...",
(lambda f: f is not None and f.kept)(rs(GONE, GONE, at(3, 22, 0), hy)[0]))
check("... and at 02:59 next day (9 h 54 min), the midnight + 3 h bound ...",
(lambda f: f is not None and f.kept)(rs(GONE, GONE, at(4, 2, 59),
hy)[0]))
check("... and at 03:00 next day it is not (midnight + 3 h; 9 h 55 min old)",
rs(GONE, GONE, at(4, 3, 0), hy)[0] is None)
hm = solarfc.Hold()
rs(st(0.9, at(3, 9, 5)), st(12.0, at(3, 9, 5)), at(3, 9, 10), hm)
check("F1: a 09:05 reading is kept at 20:00 (10 h 55 min) but not at 22:00",
rs(GONE, GONE, at(3, 20, 0), hm)[0] is not None
and rs(GONE, GONE, at(3, 22, 0), hm)[0] is None)
# F2: the 25 h autumn day (2026-10-25) and the 23 h spring day (2027-03-28).
BRU = solarfc.BILLING_TZ # Europe/Brussels, or the encoded EU rule
def bru(y, mo, d, hh, mm=0):
return datetime(y, mo, d, hh, mm, tzinfo=BRU).astimezone(timezone.utc)
for (y, mo, d), label in (((2026, 10, 25), "25 h autumn day 2026-10-25"),
((2027, 3, 28), "23 h spring day 2027-03-28")):
hd = solarfc.Hold()
t17 = bru(y, mo, d, 17, 5)
rs(st(0.4, t17), st(12.0, t17), t17 + timedelta(minutes=5), hd)
# real hours from 20:00 local to the next local midnight, each hour
start = bru(y, mo, d, 20, 0)
end = bru(y, mo, d + 1, 0)
t, miss = start, []
while t < end:
if rs(GONE, GONE, t, hd)[0] is None:
miss.append(t.astimezone(BRU).strftime("%H:%M"))
t += timedelta(minutes=30)
print(f" {label}: not kept at {miss}")
check(f"F2: {label}: the 17:05 figure is kept every half hour "
"from 20:00 to local midnight", not miss)
check(f"... and just before the next local midnight",
rs(GONE, GONE, bru(y, mo, d + 1, 0) - timedelta(minutes=1),
hd)[0] is not None)
# F5: old days are dropped from the stored hold, so it does not grow.
hp5 = solarfc.Hold()
sizes = []
for dd in range(3, 13):
t0_ = at(dd, 17, 5)
rs(st(0.4, t0_), st(12.0, t0_), t0_ + timedelta(minutes=5), hp5)
sizes.append(len(hp5.dump()["r"]))
day0s = {row[2] for row in hp5.dump()["r"]}
print(f" F5: stored entries per day over 10 days: {sizes}")
check("F5: the stored hold does not grow day after day (at most today's and "
"tomorrow's entry, with yesterday's tomorrow)", max(sizes) <= 3
and len(day0s) <= 2)
check("... and a day before today is gone",
all(row[2] >= at(12, 0, 0).isoformat() for row in hp5.dump()["r"]))
# --------------------------------------------------------------------------- #
print("\n2. The real Controller: day/night 0.34 / 0.26, cycle cost 0.07")
A, B = "inverter-a", "inverter-b"
PROFILE = ('{"periods": ['
'{"name": "night", "days": "all", "from": "22:00", "to": "07:00",'
' "eur_per_kwh": 0.26},'
'{"name": "day", "days": "all", "from": "07:00", "to": "22:00",'
' "eur_per_kwh": 0.34}]}')
CHOICE = {"a": (True, "leftover_solar", False),
"b": (True, "cheap_hours", False)}
DAY3 = 3
def opts(cost=0.07):
o = dict(rp.options(True, False, 2))
o.update(cycle_cost_threshold_eur=cost, static_profile=PROFILE,
pv_forecast_entity=T_ENT, pv_forecast_tomorrow_entity=M_ENT)
return o
def stops_at(last, *, down=()):
"""Sensors that update until `last` and then show that reading for
good. `down`: (from, to) windows in which they are unavailable."""
def f(now, ent):
for a, z in down:
if a <= now < z:
return GONE
when = min(now, last)
return st(0.4 if ent == T_ENT else 12.0, when)
return f
def never(now, ent):
return st(0.0 if ent == T_ENT else 12.0, now - timedelta(hours=9))
def drive27(o, sensors, *, pre, start, minutes=20, restart=None,
keep_store=True, snap=False):
"""-> (controller, rows, pre_rows). `pre`: the times the forecast is read
before the first cycle (the evening). `restart`: a time in `pre` at which
the add-on restarts: a new Controller on the saved store."""
store = rp._Store()
for u, (use, mode, sell) in CHOICE.items():
for k, val in (("pack_use", use), ("pack_mode", mode),
("pack_sell", sell)):
store.data[unit_key(f"inverter-{u}", k)] = val
clock = [pre[0] if pre else start]
rows, pre_rows = [], []
cur = [store]
class H(rp.Hass):
async def state(self, e):
return sensors(clock[0], e)
def build(store):
hass = H(("a", "b"), lambda t: 300.0, {"a": 40.0, "b": 40.0})
tmp = tempfile.mkdtemp(prefix="o27-")
c = Controller(dict(o), hass, store, rp._Pub(), stop=asyncio.Event(),
peak_db_path=os.path.join(tmp, "tariff.db"),
price_history_path=os.path.join(tmp, "ph.json"))
for i in c.inverters:
i.dev_min, i.dev_max = -5000.0, 5000.0
c.auto = True
# A house profile a few days old (300 W round the clock), so the day
# plan is made as it is on a site that has run for a week.
c.house_profile.load({d: {str(hh): 300.0 for hh in range(24)}
for d in ("2026-10-31", "2026-11-01",
"2026-11-02")})
return c, hass
async def go():
c, hass = build(store)
real_dt, real_hv = m.datetime, homeview.datetime
class _DT(real_dt):
@classmethod
def now(cls, tz=None):
return clock[0] if tz is None else clock[0].astimezone(tz)
m.datetime = homeview.datetime = _DT
task = None
try:
for t_pre in pre:
clock[0] = t_pre
if restart is not None and t_pre == restart:
# The saved file, through JSON; `keep_store=False` loses
# the hold only (the control for the test).
data = json.loads(json.dumps(cur[0].data))
if not keep_store:
data.pop(m.FC_HOLD_KEY, None)
new = rp._Store()
new.data.update(data)
cur[0] = new
c, hass = build(new)
await c._read_forecast()
pre_rows.append((t_pre, c.solar_fc is not None,
bool(c.solar_fc and c.solar_fc.kept)))
clock[0] = start
task = asyncio.create_task(c.run_tariff_refresh())
t = 0.0
while t < minutes * 60:
now = start + timedelta(seconds=t)
hass.t = t
clock[0] = now
if t == 0.0:
for _ in range(500):
if c.tariff_plan is not None and (
c.tariff_at is not None and c.tariff_at >= now):
break
await asyncio.sleep(0)
imp, exp = split_signed(hass.grid())
c.p1.submit(make_sample("ha_signed", imp, exp, phases=1,
ingest_ts=now,
ingest_mono=time.monotonic()))
await c.read_inputs(now_mono=t)
await c.cycle()
hass.advance(5.0, 10.0)
rp_ = c.mode_runs
rows.append(dict(
t=t, fc_none=c.solar_fc is None,
kept=bool(c.solar_fc and c.solar_fc.kept),
runs_on=None if rp_ is None else rp_.on,
reason=None if rp_ is None else rp_.reason,
roles=dict(c.mode_tick.roles) if c.mode_tick else {},
charging=hass.plants["b"].p < -200.0,
soc_b=hass.plants["b"].soc))
t += 5.0
keep = c.o
c.o = dict(keep, arbitrage_enabled=True)
try:
c.mirror, c.mirror_why = homeview.trade_verdict(
c, c.fresh_tariff_plan(clock[0]), clock[0])
finally:
c.o = keep
c.cards = homeview.battery_rows(c, False, True)
c.block = homeview.solar_block(c, clock[0])
c.hold_saved = json.loads(json.dumps(cur[0].data.get(
m.FC_HOLD_KEY)))
finally:
if task is not None:
task.cancel()
try:
await task
except asyncio.CancelledError:
pass
m.datetime, homeview.datetime = real_dt, real_hv
return c
return asyncio.run(go()), rows, pre_rows
def why_of(c, name):
card = first((b for b in c.cards if b["name"] == name), f"card {name}")
return card["why_key"]
EVENING = [at(DAY3, 16, 0) + timedelta(minutes=30 * i) for i in range(12)]
LAST = at(DAY3, 17, 5)
NIGHT_START = at(DAY3, 22, 0)
# a. a forecast that stops updating at sunset
ca, ra, pa = drive27(opts(), stops_at(LAST), pre=EVENING + [at(DAY3, 21, 55)],
start=NIGHT_START)
la = nth(ra, -1, "last cycle")
print(f" a. 22:00 with the 17:05 forecast: kept={la['kept']} on="
f"{la['runs_on']} reason={la['reason']}; B charging "
f"{sum(r['charging'] for r in ra)} of {len(ra)}; Why A {why_of(ca, A)} "
f"B {why_of(ca, B)}; Home mirror {ca.mirror}")
check("the forecast was read fresh before 17:05 and kept after it",
any(not k for _t, ok, k in pa if _t <= LAST + timedelta(minutes=30))
and all(ok for _t, ok, _k in pa)
and all(k for _t, _ok, k in pa if _t >= at(DAY3, 20, 15)))
check("22:00 night run: planned with the cycle-cost gap (needs 0.07 < 0.08)",
present(la) and not la["fc_none"] and la["kept"]
and la["runs_on"] is True and la["reason"] != "not_worth")
check("... and B (Cheap hours) charges from the grid in the night period",
sum(r["charging"] for r in ra) > len(ra) // 2
and la["roles"].get(B) == modetrade.CHARGE
and why_of(ca, A) != "no_spread" and why_of(ca, B) != "no_spread")
check("the day plan is made from the kept forecast (levels for Leftover "
"solar), as it is from a fresh one",
ca.day_plan is not None and ca.plan_targets() is not None
and ca.block["state"] == "plan")
check("Home's mirror of the gate allows it, as control does",
ca.mirror is not None and ca.mirror_why == "allowed")
# b. no forecast all day
cb, rb, pb = drive27(opts(), never, pre=EVENING + [at(DAY3, 21, 55)],
start=NIGHT_START)
lb = nth(rb, -1, "last cycle")
print(f" b. no forecast all day: on={lb['runs_on']} reason={lb['reason']}; "
f"B charging {sum(r['charging'] for r in rb)}; Why A {why_of(cb, A)} "
f"B {why_of(cb, B)}")
check("with no forecast all day the night run still needs 1.5x (0.105 > 0.08)"
": not worth, no charge, 'Prices differ too little'",
present(lb) and lb["fc_none"] and lb["runs_on"] is False
and lb["reason"] == "not_worth"
and not any(r["charging"] for r in rb)
and why_of(cb, A) == "no_spread" and why_of(cb, B) == "no_spread"
and not any(ok for _t, ok, _k in pb))
check("... and there is no day plan (Leftover solar takes solar only)",
cb.day_plan is None and cb.plan_targets() is None
and cb.block["state"] == "no_forecast")
check("... Home's mirror refuses it too, for the gap ('not_worth')",
cb.mirror is None and cb.mirror_why == "not_worth")
check("... and nothing is saved for the night (no fresh reading was made)",
cb.hold_saved is None or cb.hold_saved.get("r") == [])
# c. an evening restart
cc, rc, pc = drive27(opts(), stops_at(LAST),
pre=EVENING + [at(DAY3, 21, 55)], start=NIGHT_START,
restart=at(DAY3, 21, 55))
lc = nth(rc, -1, "last cycle")
check("an evening restart (new Controller on the saved store) keeps the "
"forecast: the night run is planned and charges",
present(lc) and lc["kept"] and lc["runs_on"] is True
and sum(r["charging"] for r in rc) > len(rc) // 2
and present(nth(pc, -1, "last read")) and pc[-1][1] and pc[-1][2])
cc0, rc0, pc0 = drive27(opts(), stops_at(LAST),
pre=EVENING + [at(DAY3, 21, 55)], start=NIGHT_START,
restart=at(DAY3, 21, 55), keep_store=False)
lc0 = nth(rc0, -1, "last cycle")
check("(control: the same restart with the store lost is blind, 1.5x)",
present(lc0) and lc0["fc_none"] and lc0["runs_on"] is False
and present(nth(pc0, -1, "last read")) and not pc0[-1][1])
# d. a one-hour outage at 18:47
OUT = (at(DAY3, 18, 47), at(DAY3, 19, 47))
quarter = [at(DAY3, 16, 0) + timedelta(minutes=15 * i) for i in range(24)]
cd, rd, pd = drive27(opts(), stops_at(at(DAY3, 18, 5), down=[OUT]),
pre=quarter + [at(DAY3, 21, 55)], start=NIGHT_START)
ld = nth(rd, -1, "last cycle")
inside = [(t_, ok) for t_, ok, _k in pd if OUT[0] <= t_ < OUT[1]]
print(f" d. reads inside the outage: {len(inside)}, with a forecast: "
f"{sum(ok for _t, ok in inside)}")
check("a one-hour outage at 18:47 does not drop the forecast "
"(every read from 16:00 to 22:00 has one)",
present(pd) and len(inside) >= 4 and all(ok for _t, ok, _k in pd))
check("... and the 22:00 night run is planned and charges",
present(ld) and ld["runs_on"] is True
and sum(r["charging"] for r in rd) > len(rd) // 2)
# e. the bound: midnight + 3 h
mid = [at(DAY3, 16, 0) + timedelta(minutes=30 * i) for i in range(12)]
ce, re_, pe = drive27(opts(), stops_at(LAST), pre=mid + [at(4, 2, 30)],
start=at(4, 2, 30))
le = nth(re_, -1, "last cycle")
check("02:30 (before midnight + 3 h): the figure from 17:05 is still used, "
"as today's; the night run is planned",
present(le) and le["kept"] and le["runs_on"] is True)
cf, rf, pf = drive27(opts(), stops_at(LAST), pre=mid + [at(4, 3, 30)],
start=at(4, 3, 30))
lf = nth(rf, -1, "last cycle")
check("03:30 (after midnight + 3 h): a figure from before the night is not "
"used: no forecast, 1.5x, not worth, no charge",
present(lf) and lf["fc_none"] and lf["runs_on"] is False
and lf["reason"] == "not_worth" and not any(r["charging"] for r in rf))
# g. OPT-28's shortfall sizing on a kept figure: Leftover solar (A) is sized
# from the kept forecast exactly as from a fresh one, not "no forecast"
def fresh_all(now, ent):
return st(0.4 if ent == T_ENT else 12.0, now)
cg, rg, pg = drive27(opts(), fresh_all, pre=EVENING + [at(DAY3, 21, 55)],
start=NIGHT_START)
kp, fp_ = ca.day_plan, cg.day_plan
print(f" g. kept : short={kp and kp.short_kwh} uncovered="
f"{kp and kp.uncovered_kwh} targets={kp and kp.targets} "
f"need={kp and kp.need_kwh} sun={kp and kp.sun_kwh}")
print(f" fresh : short={fp_ and fp_.short_kwh} uncovered="
f"{fp_ and fp_.uncovered_kwh} targets={fp_ and fp_.targets} "
f"need={fp_ and fp_.need_kwh} sun={fp_ and fp_.sun_kwh}")
check("a kept figure sizes the night by OPT-28's shortfall, with a level for "
"Leftover solar (A): not 'no forecast', which has no plan at all",
kp is not None and A in kp.targets and kp.short_kwh > 0.0
and cb.day_plan is None and not cg.solar_fc.kept)
check("... and the level, the grid need, the shortfall and the sun counted "
"are those a fresh figure gives (counted once, from the right time)",
fp_ is not None and kp.targets == fp_.targets
and kp.need_kwh == fp_.need_kwh and kp.share_kwh == fp_.share_kwh
and abs(kp.short_kwh - fp_.short_kwh) < 1e-9
and abs(kp.uncovered_kwh - fp_.uncovered_kwh) < 1e-9
and abs(kp.sun_kwh - fp_.sun_kwh) < 1e-9)
# --------------------------------------------------------------------------- #
print("\n3. The screens")
blk = ca.block
print(f" forecast line: {blk['line']!r} (state {blk['state']})")
check("the forecast line says when it was read and that it is kept for "
"tonight", blk["line"] == "Solar forecast read 17:05, kept for tonight.")
check("the same line comes from the Forecast tab (dayview) on today",
homeview.line_solar(0.4, 0.3, "17:05", kept=True) == blk["line"]
and homeview.line_solar(0.4, 0.3, "17:05").startswith(
"Solar still to come today"))
en = json.load(open(os.path.join(HERE, "app", "ui_text", "en.json"),
encoding="utf-8"))
nl = json.load(open(os.path.join(HERE, "app", "ui_text", "nl.json"),
encoding="utf-8"))
k_en = en["home_words"]["solar"]["kept"]
k_nl = nl["home_words"]["solar"]["kept"]
print(f" EN {k_en!r}\n NL {k_nl!r}")
check("EN and NL both say it, with the time, and the NL is not the English",
"{at}" in k_en and "{at}" in k_nl and k_en != k_nl
and "tonight" in k_en and "vannacht" in k_nl)
check("a daytime reading keeps the old line (daytime unchanged)",
"Updated 09:05" in homeview.line_solar(1.0, 0.7, "09:05")
and "Bijgewerkt" in json.dumps(nl["home_words"]["solar"]["line"]))
check("no forecast: the old 'No solar forecast now' line, no 'kept'",
cb.block["state"] == "no_forecast"
and "tonight" not in cb.block["line"])
check("control and Batteries screen agree on the 22:00 run: not 'no_spread'",
why_of(ca, A) != "no_spread" and why_of(ca, B) != "no_spread"
and la["runs_on"] is True)
print()
if fails:
print(f"{len(fails)} FAILED:")
for f in fails:
print(" -", f)
sys.exit(1)
print("all checks passed")
+473
View File
@@ -0,0 +1,473 @@
"""Runnable check for SAFETY-126. `python3 test_safety126.py`
During the night grid charge, with both batteries under the peak reserve,
the add-on logged ERROR PEAK_GUARD_FLOOR_LIMITED and raised "Battery stopped
limiting the peak" at quarter ends, and the Batteries "Why" blinked "The peak
limit needs power from the batteries.", while every quarter landed on the aim.
The cause: the charge lands the quarter on the aim, a 1 W landing error in
the quarter's last second becomes tens of kW through `excess_w()`'s
900 / remaining, and the alarm fired on any `wanted > 0` when no battery can
give. DECISIONS/SAFETY-126-landing-error-is-not-an-overrun.md.
Drives the REAL Controller (read_inputs + cycle, the tariff task on a
day/night profile, a P1 meter on a simulated clock, OPT-15's lag), as
test_safety125.py does, with the cycle a fraction of a second off the
quarter's edge as on the rig.
1. The night grid charge under the reserve: no ERROR, no notification, no
red peak row, no "Why" peak, every quarter on the aim; the guard did want
power at the quarter's end (not vacuous); with the fix reverted the same
run alarms. The commands are the same with and without the fix.
2. A real overrun under the reserve still alarms, a step load at the same
cycle as before, and names both batteries. After it, the latch is reset:
the night's later landings are quiet again.
3. `misses_aim()` on its own.
4. The page and the "Why" on a decision within the aim.
Mutations (MUTATION.md, purged copies) are listed in the pull request.
"""
import asyncio
import dataclasses
import io
import logging
import os
import sys
import tempfile
import types
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"))
import opt11p3_replay as rp # noqa: E402
from named_absence import first, nth, present # noqa: E402
import app.main as m # noqa: E402
import app.p1 as p1mod # noqa: E402
from app import homeview, messages, peakguard, tariff # noqa: E402
from app import price as _price, static_tariff as _st # noqa: E402
from app.main import Controller # noqa: E402
from app.p1 import make_sample, split_signed # noqa: E402
from app.store import unit_key # noqa: E402
from app.tariff import QUARTER_S, Projection # noqa: E402
logging.disable(logging.NOTSET)
logging.getLogger().setLevel(logging.CRITICAL)
_ERRORS = io.StringIO()
_h = logging.StreamHandler(_ERRORS)
_h.setLevel(logging.ERROR)
logging.getLogger("goodwe").addHandler(_h)
logging.getLogger("goodwe").setLevel(logging.ERROR)
logging.getLogger("goodwe").propagate = False
fails = []
def check(name, cond):
print(f" {'ok ' if cond else 'FAIL'} {name}")
if not cond:
fails.append(name)
PEAK = 5413.17 # October's banked peak on the live site
AIM_CAP = 3500.0 - 50.0
AIM_PEAK = PEAK - 50.0
TOL = 10.0 # W over the aim a quarter may land (SAFETY-81)
L = timezone(timedelta(hours=1))
NIGHT = datetime(2026, 10, 30, 22, 0, tzinfo=L).astimezone(timezone.utc)
PROFILE = ('{"periods": ['
'{"name": "night", "days": "all", "from": "22:00", "to": "07:00",'
' "eur_per_kwh": 0.2333},'
'{"name": "day", "days": "all", "from": "07:00", "to": "22:00",'
' "eur_per_kwh": 0.2801}]}')
OWNER = {"a": "leftover_solar", "b": "cheap_hours"}
def opts(*, modes=True, arb=False):
o = dict(rp.options(modes, arb, 2))
o.update(cycle_cost_threshold_eur=0.03, static_profile=PROFILE,
peak_cap_w=3500, peak_margin_w=50)
o["inverters"][0].update(max_w=5000, capacity_kwh=15.36)
o["inverters"][1].update(max_w=4600, capacity_kwh=20.48)
return o
class Harder(rp.Plant):
"""Charges harder than asked, as inverter-a did live (test_safety125.py:
asked -4271 W, measured about -4750 W; asked -5000 W, about -5210 W), up
to `cap` W. So a charge sized to the aim lands a few W over it, as on the
rig (worst +7 W)."""
cap = -5216.0
def step(self, dt, floor):
c = self.cmd
if c < 0:
self.cmd = max(c * 1.11, self.cap)
try:
super().step(dt, floor)
finally:
self.cmd = c
async def _no_fetch(*a, **k):
raise AssertionError("the tariff task reached for a price feed")
def without_fix(fn):
"""release/1.0 cc9bbeb: no decision is ever within the aim."""
real = getattr(Controller, "_peak_landing", None)
if real is None:
return fn()
Controller._peak_landing = lambda self, decision: decision
try:
return fn()
finally:
Controller._peak_landing = real
def drive(o, *, modes=OWNER, socs=(10.0, 10.0), house=600.0, minutes=46,
start=NIGHT, off=0.7, peak=None):
"""The real Controller from `start` + `off` s for `minutes` at a 1 s
cycle with OPT-15's lag, so a cycle lands `1 - off` s before each
quarter's end. `house` is W or a function of t. Returns a dict: the
quarters' true grid averages, the commands, the reserve notifications,
the ERROR lines, and per cycle where the guard engaged: (t, floor
limited, within the aim, battery "Why" peak, red reserve row, red
saturated row, alarm latched)."""
store = rp._Store()
for u, mode in (modes or {}).items():
store.data[unit_key(f"inverter-{u}", "pack_use")] = True
store.data[unit_key(f"inverter-{u}", "pack_mode")] = mode
store.data[unit_key(f"inverter-{u}", "pack_sell")] = False
load = house if callable(house) else (lambda t: float(house))
t0 = start + timedelta(seconds=off)
out = {"cmds": [], "notes": [], "seen": [], "sums": {}}
_ERRORS.seek(0)
_ERRORS.truncate()
async def go():
lag = rp.Lag()
hass = rp.Hass(("a", "b"), load, {"a": socs[0], "b": socs[1]},
limits={"a": (-5000.0, 5000.0),
"b": (-4600.0, 4600.0)}, lag=lag)
for u, cap in (("a", -5216.0), ("b", -4800.0)):
hass.plants[u].__class__ = Harder
hass.plants[u].cap = cap
real_set = hass.set_number
async def set_number(e, v):
out["cmds"].append((round(hass.t, 3), e, float(v)))
return await real_set(e, v)
async def call(*a):
if a[:2] == ("persistent_notification", "create") \
and a[2]["notification_id"] == messages.ID_PEAK_RESERVE:
out["notes"].append((hass.t, a[2]["message"]))
return True
hass.set_number, hass.call = set_number, call
tmp = tempfile.mkdtemp(prefix="s126-")
db = os.path.join(tmp, "tariff.db")
if peak is not None:
tariff.PeakStore(db)._db.execute(
"INSERT INTO monthly_peak VALUES (?, ?, ?, NULL)",
(tariff.month_key(t0), peak,
"2026-10-21T20:15:00+00:00")).connection.commit()
c = Controller(dict(o), hass, store, rp._Pub(), stop=asyncio.Event(),
peak_db_path=db,
price_history_path=os.path.join(tmp, "ph.json"))
c.inverters[0].dev_min, c.inverters[0].dev_max = -5000.0, 5000.0
c.inverters[1].dev_min, c.inverters[1].dev_max = -4600.0, 4600.0
c.auto = True
real = m.datetime, homeview.datetime, p1mod.time
clock, simt = [t0], [0.0]
class _DT(real[0]):
@classmethod
def now(cls, tz=None):
return clock[0] if tz is None else clock[0].astimezone(tz)
m.datetime = homeview.datetime = _DT
p1mod.time = types.SimpleNamespace(monotonic=lambda: 1e6 + simt[0])
real_load = _price.load_prices, _st.load_prices
_price.load_prices = _st.load_prices = _no_fetch
task = (asyncio.create_task(c.run_tariff_refresh())
if o.get("battery_modes_enabled") or o.get("arbitrage_enabled")
else None)
try:
hass.advance(0.0, 10.0)
t = 0.0
while t < minutes * 60:
now = t0 + timedelta(seconds=t)
hass.t, simt[0], clock[0] = t, t, now
if t == 0.0 and task is not None:
for _ in range(500):
if c.tariff_plan is not None:
break
await asyncio.sleep(0)
q = int(now.timestamp() // QUARTER_S)
acc = out["sums"].setdefault(q, [0.0, 0])
if t >= lag.p1_s:
imp, exp = split_signed(hass._past(lag.meter_s
+ lag.p1_s)[0])
c.p1.submit(make_sample(
"ha_signed", imp, exp, phases=1,
ingest_ts=now - timedelta(seconds=lag.p1_s),
ingest_mono=1e6 + t))
await c.read_inputs(now_mono=t)
await c.cycle()
hass.advance(lag.tick, 10.0)
acc[0] += max(0.0, hass.grid())
acc[1] += 1
g = c.peak_guard
if g is not None and g.engaged and g.blind is None:
rows = {r["id"]: r for r in c.checks(now)}
row = rows.get("peak_guard", {})
red = row.get("ok") is False and not row.get("warn")
out["seen"].append((
t, g.floor_limited, getattr(g, "within_aim", False),
homeview.peak_engaged(c),
red and "banner" in row,
red and "banner" not in row,
c._peak_floor_limited))
t += lag.tick
finally:
if task is not None:
task.cancel()
try:
await task
except asyncio.CancelledError:
pass
m.datetime, homeview.datetime, p1mod.time = real
_price.load_prices, _st.load_prices = real_load
return c
out["c"] = asyncio.run(go())
out["quarters"] = {
datetime.fromtimestamp(q * QUARTER_S, L).strftime("%H:%M"): s / k
for q, (s, k) in sorted(out["sums"].items()) if k >= QUARTER_S - 1}
out["errors"] = [x for x in _ERRORS.getvalue().splitlines()
if "PEAK_GUARD_FLOOR_LIMITED" in x]
return out
def show(tag, r):
print(f" {tag}: quarters "
f"{ {k: round(v, 1) for k, v in r['quarters'].items()} }, "
f"guard cycles {len(r['seen'])}, floor-limited "
f"{sum(1 for s in r['seen'] if s[1])}, within-aim "
f"{sum(1 for s in r['seen'] if s[2])}, Why peak "
f"{sum(1 for s in r['seen'] if s[3])}, notes {len(r['notes'])}, "
f"errors {len(r['errors'])}")
def quiet(r):
"""No reserve alarm in any form: ERROR, notification, red row, latch,
"Why" peak, and no red saturated row."""
return (not r["errors"] and not r["notes"]
and not any(s[3] or s[4] or s[5] or s[6] for s in r["seen"]))
def on_aim(r, aim):
q = r["quarters"]
return len(q) >= 3 and all(v <= aim + TOL for v in q.values())
# ============================================================================ #
print("1. the night grid charge under the peak reserve")
NIGHTS = {}
# The owner's shape at the cap's aim. At the banked peak's aim it is
# test_safety125.py's live shape (A Cheap hours, B Solar first), because
# B's 4 600 W alone stays under that aim.
LIVE = {"a": "cheap_hours", "b": "solar_first"}
for tag, kw, aim in (
("aim 3450, cycle 0.3 s before the edge", dict(off=0.7), AIM_CAP),
("aim 3450, cycle 0.01 s before the edge", dict(off=0.99), AIM_CAP),
("month peak 5413, cycle 0.01 s before the edge",
dict(off=0.99, peak=PEAK, modes=LIVE, socs=(11.0, 10.0)),
AIM_PEAK)):
head = drive(opts(), **kw)
base = without_fix(lambda: drive(opts(), **kw))
NIGHTS[tag] = (head, base)
show("head " + tag, head)
show("fix reverted", base)
check(f"⭐ {tag}: no ERROR, no notification, no red peak row and no "
f"'Why' peak", quiet(head))
check(f"{tag}: every quarter at or under the aim ({aim:.0f} W + "
f"{TOL:.0f} W)", on_aim(head, aim))
check(f"{tag}: not vacuous, the guard wanted power under the reserve "
f"at a quarter's end (floor-limited, within the aim)",
any(s[1] and s[2] for s in head["seen"]))
check(f"{tag}: with the fix reverted the same run alarms",
not quiet(base))
check(f"{tag}: the commands are the same with and without the fix",
head["cmds"] == base["cmds"] and len(head["cmds"]) > 100)
_h, _b = NIGHTS["aim 3450, cycle 0.01 s before the edge"]
check("fix reverted, at 0.01 s: the ERROR, the notification and the "
"'Why' peak all come back (the rig's three symptoms)",
_b["errors"] and _b["notes"] and any(s[3] for s in _b["seen"]))
# ============================================================================ #
print("\n2. a real overrun under the reserve still alarms")
def evening(t):
"""600 W, then an oven and a car charger from 20 minutes on."""
return 600.0 if t < 1200 else 6500.0
for tag, onset, kw in (
("night grid charge, then 6500 W", 1200.0, dict(house=evening)),
("modes off, 5500 W all along", 0.0,
dict(house=5500.0, modes=None, socs=(15.0, 12.0)))):
o = opts(modes=kw.get("modes", OWNER) is not None)
head = drive(o, minutes=35, **kw)
base = without_fix(lambda: drive(o, minutes=35, **kw))
show("head " + tag, head)
show("fix reverted", base)
# From the load's onset: before it, the fix reverted shows the night's
# false alarm (part 1).
first_h = next((s[0] for s in head["seen"] if s[6] and s[0] >= onset),
None)
first_b = next((s[0] for s in base["seen"] if s[6] and s[0] >= onset),
None)
check(f"⭐ {tag}: the ERROR and the notification, naming both "
f"batteries",
head["errors"] and head["notes"]
and all("inverter-a and inverter-b" in n for _t, n in head["notes"]))
_th = [t for t, _n in head["notes"]]
_tb = [t for t, _n in base["notes"] if t >= onset]
print(f" notifications at {_th} s; with the fix reverted {_tb} s")
check(f"{tag}: the first at the same cycle as before the fix "
f"({first_h} s, {first_b} s), and none that was not raised before",
first_h is not None and first_h == first_b
and present(nth(_th, 0)) and nth(_th, 0) == nth(_tb, 0)
and set(_th) <= set(_tb))
check(f"{tag}: the red peak row and its banner while it stands",
any(s[4] for s in head["seen"]))
def burst(t):
"""600 W, a 6 500 W load from 20 to 25 minutes, then 600 W again while
the night charge goes on."""
return 6500.0 if 1200 <= t < 1500 else 600.0
# F3 (review): `_peak_landing()` resets its latch when the guard lets go, so
# a real overrun does not keep the later landing errors alarming.
_bh = drive(opts(), house=burst, off=0.99, minutes=61)
_bb = without_fix(lambda: drive(opts(), house=burst, off=0.99,
minutes=61))
show("head real overrun, then the night again", _bh)
show("fix reverted", _bb)
# The burst banked a new month peak, which the later quarters stay under.
_new_peak = _bh["c"].peak_guard.ceiling_w
def _late(r):
"""From the next quarter on: the overrun's own quarter ends over."""
return [x for x in r["seen"] if x[0] >= 1800]
print(f" notifications at {[t for t, _n in _bh['notes']]} s; with the fix "
f"reverted {[t for t, _n in _bb['notes']]} s; limit after the burst "
f"{_new_peak:.0f} W")
check("⭐ after a real overrun ends, the latch is reset: the alarm came "
"during the overrun, and the later quarters' landings are quiet "
"(no notification, no 'Why' peak, no red row) and under the new peak",
any(1200 <= t < 1500 for t, _n in _bh["notes"])
and not [t for t, _n in _bh["notes"] if t >= 1800]
and not any(x[3] or x[4] or x[5] for x in _late(_bh))
and present(nth([v for k, v in _bh["quarters"].items()
if k >= "22:30"], 0))
and all(v <= _new_peak for k, v in _bh["quarters"].items()
if k >= "22:30"))
check("...not vacuous: the guard wanted power at those landings, and with "
"the fix reverted they show the 'Why' peak or an alarm",
any(x[2] for x in _late(_bh))
and (any(x[3] or x[4] or x[5] for x in _late(_bb))
or bool([t for t, _n in _bb["notes"] if t >= 1800])))
# ============================================================================ #
print("\n3. misses_aim() on its own")
START = datetime(2026, 10, 30, 21, 15, tzinfo=timezone.utc)
def proj(projected, elapsed, held):
return Projection(projected, projected, elapsed, 1.0, True, START, held)
check("the rig's case: 3452 W against 3450 W, 0.3 s left, charging "
"1250 W from the grid: not a miss",
not peakguard.misses_aim(proj(3452.0, 899.7, 3452.0), batt_w=-1250.0,
target_w=AIM_CAP))
check("...although decide() wants power there (its gain, 900 / 0.3 s)",
(peakguard.decide(proj(3452.0, 899.7, 3452.0), batt_w=-1250.0,
cap_w=3500, margin_w=50, reading=None,
achievable_w=0.0, soc=10.0, soc_floor=20.0)
.wanted_w or 0.0) > 1000.0)
check("a quarter heading 11 W over the aim with the batteries at 0 W is a "
"miss", peakguard.misses_aim(proj(3461.0, 899.7, 3461.0), batt_w=0.0,
target_w=AIM_CAP))
check("...and 10 W is not (ALARM_MISS_W, as for 'Load too high')",
not peakguard.misses_aim(proj(3460.0, 899.7, 3460.0), batt_w=0.0,
target_w=AIM_CAP))
check("the add-on's own grid charge counts as stoppable: 4000 W on the "
"meter mid-quarter, 3000 W of it charging, is not a miss",
not peakguard.misses_aim(proj(2500.0, 450.0, 4000.0), batt_w=-3000.0,
target_w=AIM_CAP))
check("...and a house of 6000 W the batteries are covering is one",
peakguard.misses_aim(proj(1000.0, 450.0, 0.0), batt_w=6000.0,
target_w=AIM_CAP))
check("no battery reading, or no seconds left: not a miss (decide() files "
"nothing there)",
not peakguard.misses_aim(proj(9000.0, 450.0, 9000.0), batt_w=None,
target_w=AIM_CAP)
and not peakguard.misses_aim(proj(9000.0, 900.0, 9000.0), batt_w=0.0,
target_w=AIM_CAP))
# ============================================================================ #
print("\n4. the page and the 'Why' on a decision within the aim")
_c = NIGHTS["aim 3450, cycle 0.01 s before the edge"][0]["c"]
_dec = peakguard.PeakDecision(
engaged=True, floor_w=5000.0, ceiling_w=3500.0,
ceiling_from=peakguard.CEILING_CAP, projected_w=3451.0, excess_w=9e4,
wanted_w=9e4, target_w=AIM_CAP, saturated=True, soc=30.0, soc_floor=20.0)
def page(d):
"""(the peak row is red, the "Why" says peak) for decision `d`."""
_c.peak_guard, _c._peak_floor_limited = d, False
row = first((r for r in _c.checks(datetime.now(timezone.utc))
if r["id"] == "peak_guard"), "peak row")
if not present(row):
return {"red": None}, None
return dict(row, red=row["ok"] is False and not row.get("warn")), \
homeview.peak_engaged(_c)
_r1, _w1 = page(_dec)
_r2, _w2 = page(dataclasses.replace(_dec, within_aim=True))
check("a saturated decision: red 'Load too high' row and the 'Why' peak, "
"as before", _r1["red"] and _w1)
check("...within the aim: neither", _r2["red"] is False and _w2 is False)
_lim = dataclasses.replace(_dec, floor_w=0.0, saturated=False,
floor_limited=True, soc=10.0,
soc_floor_reason=peakguard.FLOOR_AT_SOC_FLOOR)
_r3, _w3 = page(_lim)
_r4, _w4 = page(dataclasses.replace(_lim, within_aim=True))
check("floor-limited at 0 W: the red reserve row with its banner, as "
"before", _r3["red"] and "banner" in _r3 and not _w3)
check("...within the aim: no red row",
_r4["red"] is False and _w4 is False)
# ============================================================================ #
print()
if fails:
print(f"{len(fails)} SAFETY-126 check(s) FAILED:")
for f in fails:
print(f" - {f}")
sys.exit(1)
print("all SAFETY-126 checks passed")
+13 -1
View File
@@ -48,7 +48,7 @@ logging.disable(logging.WARNING)
import opt11p3_replay as rp # noqa: E402
from named_absence import first, nth, present # noqa: E402
import app.main as m # noqa: E402
from app import arbitrage, homeview, modetrade # noqa: E402
from app import arbitrage, homeview, modetrade, solarfc # noqa: E402
from app.main import Controller # noqa: E402
from app.p1 import make_sample, split_signed # noqa: E402
from app.price import (PricePoint, PriceProduct, PriceRead, # noqa: E402
@@ -59,6 +59,13 @@ from app.store import unit_key # noqa: E402
fails = []
class _NoHold(solarfc.Hold):
"""OPT-27: a hold that never stands in for a lost sensor."""
def stand_in(self, *a, **k):
return None
def check(name, cond):
print(f" {'ok ' if cond else 'FAIL'} {name}")
if not cond:
@@ -163,6 +170,11 @@ def drive(o, *, fc="fresh", points=None, minutes=20, tick=5.0,
for i in c.inverters:
i.dev_min, i.dev_max = -5000.0, 5000.0
c.auto = True
# OPT-27: a sensor that is missing is covered for up to 3 h by its last
# fresh reading. This file is about what the plan does when the
# forecast IS lost, so it runs with a hold that never covers
# (test_opt27.py tests the cover).
c.fc_hold = _NoHold()
real_dt, real_hv = m.datetime, homeview.datetime
clock = [t0]