Compare commits

...
10 Commits
Author SHA1 Message Date
glenn schrooyenandClaude Opus 5.5 6b2d0091d6 PRICE-02 review: keep an open buy over midnight in the real-year replay; say which variant the kWh figure is
checks / unit (pull_request) Successful in 47m40s
infra-pin / fresh (pull_request) Successful in 1s
The production-shaped plan sees tomorrow from 13:00, so most buys are sold
the next day or later. The ledger forgot the open buy at each midnight and
booked 46 of 51 sales as holdings: 5 trips, +1.21 EUR. Carried over, the
same decisions give 51 trips, +13.81 EUR, all clearing the threshold.
Decisions are unchanged and identical on the old and new verdict. No
assertion changed.

DECISIONS: 27.5 -> 42.6 kWh is the refresh-:00, tomorrow-known variant; give
the range of the other three.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Bdqr7PiaHcBnzPituu2yTC
2026-10-22 16:48:43 +02:00
glenn schrooyenandClaude 3dd4dd89e0 PRICE-02 round 3: say that hourly and static plans change too, test plans built off the hour, exact period end and the Home page path, production-shaped real-year replay
checks / unit (pull_request) Successful in 47m47s
infra-pin / fresh (pull_request) Successful in 1s
Docs and tests only; no change under goodwe_controller/app.

Co-Authored-By: Claude <noreply@anthropic.com>
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Bdqr7PiaHcBnzPituu2yTC
2026-10-22 14:59:13 +02:00
glenn schrooyenandClaude 891ed7bdd7 PRICE-02 round 2: require now, undo the test loosening, analyse ties
checks / unit (pull_request) Successful in 47m49s
infra-pin / fresh (pull_request) Successful in 2s
- TariffPlan.arbitrage(now=...) is required (keyword-only, no default); so are
  homeview.trade_verdict/refusal_key's now. Existing test calls pass one.
- test_arbitrage.py: no module-level decide wrapper. Fixtures that probed a
  one-day plan after its extremes now hold two days (as a site does); the
  flap day gets a unique cheapest and dearest hour; the replay breaks its
  ties without direction. The flap bound is <= 4 and the replay check is
  len(_trades) < 365 again. 228 checks pass on the old and the new code.
- test_price02.py part 6: ties (equal quarters, a real 15-minute day,
  alternating hourly prices) and the required now.
- Identity pins: the masked digest is only for the older-tree comparison;
  this tree's own digests are pinned with nothing hidden.
- DECISIONS: the tie analysis with numbers.

Co-Authored-By: Claude <noreply@anthropic.com>
2026-10-22 10:37:43 +02:00
glenn schrooyenandClaude 2ff9be710c PRICE-02: correct the DECISIONS note (which sentences changed, how identity is pinned)
checks / unit (pull_request) Successful in 47m48s
infra-pin / fresh (pull_request) Successful in 1s
Co-Authored-By: Claude <noreply@anthropic.com>
2026-10-22 08:37:39 +02:00
glenn schrooyenandClaude ecd9822861 PRICE-02: pin the modes-off identity with a digest that hides only the changed words
checks / unit (pull_request) Successful in 48m44s
infra-pin / fresh (pull_request) Successful in 2s
test_price_run_fixes.py part 5 and test_modes_p3.py part 1 compare the modes-off
replay with digests printed on older trees. The trade sentence now says
"price period", and a sale quotes another "above the cheapest" figure once the
first cheap period is over. ident.trade_digest() hides exactly those two
things; re-measured on e3e4ba6, 7a00f12, 90bc901 and this tree, the digests
are the same, so every command is unchanged.

Co-Authored-By: Claude <noreply@anthropic.com>
2026-10-22 08:20:50 +02:00
glenn schrooyenandClaude 4297f7f3f1 PRICE-02: the trading verdict skips price periods that have ended
checks / unit (pull_request) Failing after 48m38s
infra-pin / fresh (pull_request) Successful in 1s
TariffPlan.arbitrage(now=...) drops the points whose period has ended
(end <= now), so the cheapest and dearest point can still be traded.
decide() (modes off), the modes-on allowed() gate and the Home verdict
all pass their own now. Hourly feeds and the static tariff are unchanged.
The off-peak sentence and the charge/discharge sentences say "price
period". Tests: test_price02.py; test_arbitrage.py rolls its whole-day
plans to start at now; test_modes_p3.py part 1 digests leave the trade
sentence out (commands unchanged on 7a00f12, 90bc901 and this tree).

Co-Authored-By: Claude <noreply@anthropic.com>
2026-10-22 07:19:28 +02:00
admin 90bc9017b2 Merge pull request 'UX-46: Dutch status rows, banners, notes and wizard lines on NL pages; peak forecast label without (W)' (#205) from UX-46 into release/1.0
infra-pin / fresh (push) Successful in 2s
2026-10-22 06:10:36 +02:00
glenn schrooyenandClaude Opus 5.5 ae0ef4296b UX-46 review: Dutch wording fixes on the NL rows and notes
checks / unit (pull_request) Successful in 45m44s
infra-pin / fresh (pull_request) Successful in 1s
Text only. "stroom te opnemen" -> "stroom op te nemen"; Developer Tools is
"Ontwikkelhulpmiddelen" in HA and ui_text; the green stuck-meter row no longer
starts "Vastgelopen meter:"; maintenance banner names the phase in Dutch;
double "dan" in the device note; plainer peak reserve, peak guard, refused
commands and missing-entity wording. Two stale comments now say the banner
and Home bodies follow the page's language.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-22 04:56:30 +02:00
glenn schrooyenandClaude c1809b4ca9 UX-46: test_ux46 points at MUTATION.md and reads rows without raising
checks / unit (pull_request) Successful in 44m51s
infra-pin / fresh (pull_request) Successful in 2s
Co-Authored-By: Claude <noreply@anthropic.com>
2026-10-22 03:41:18 +02:00
glenn schrooyenandClaude 7a392a5367 UX-46: Dutch status rows, banners, notes and wizard lines on the NL pages
Rows and banners are built once with their Dutch beside them (messages.Bi)
and status_payload() picks one by the page's language. Settings notes and the
wizard's checks carry a Dutch twin (Note.nl) that /config/validate picks by
the posted language. The peak forecast label drops "(W)" in EN and NL.

Co-Authored-By: Claude <noreply@anthropic.com>
2026-10-22 03:21:41 +02:00
25 changed files with 2170 additions and 312 deletions
@@ -0,0 +1,89 @@
# PRICE-02 - the trading verdict skips price periods that have ended
**Status: for review.** Tests: `test_price02.py`. Found by AUDIT-01 (F1).
The refresh floors the plan to the top of the hour, so on the 15-minute feed
it holds the periods of this hour that are already over. The verdict took the
cheapest and dearest over every point. With modes off it could aim at a period
that had passed and then do nothing, for up to 45 min plus the refresh lag.
With modes on the `allowed()` gate could be open only because of such a period.
Decision: `TariffPlan.arbitrage(now=...)` drops the points whose period has
ended (`end <= now`, the span `price._cover_table()` gives, as
`priceruns.firm_slots()` does). Its three callers pass their own `now`:
`arbitrage.decide()`, `Controller._mode_roles()` and `homeview.trade_verdict()`.
- Taken in the verdict, not when the plan is built: a plan built at 03:00 is
still used at 03:20, and the Home chart keeps the day's earlier bars.
- An hourly point that is still current stays in. But an hourly or static plan
is rebuilt every 30 min counted from add-on start, so it is usually off the
hour and, until the next refresh, holds the hour that has just ended. That
hour is dropped too (round 3, owner-accepted 2026-10-22). On a day/night
tariff all day hours tie and the old verdict took the first, the ended one,
so it paused the trade every hour until the refresh: 91 on/off changes over
2 days with the refresh at :10, 4 now (4 before as well, with the refresh on
the hour). With the refresh on the hour nothing changes.
- `now` is required: keyword-only, no default, like the other three arguments. A
caller that could leave it out would get the elapsed periods back and the
defect with them. A test that wants the whole series passes an instant before it.
- The sentences of this verdict in `arbitrage.py` (status row and action log,
English only) say "price period" where they said "hour": the idle one now
reads "now is not the cheapest or the dearest price period". The UI texts
that say "cheapest hour" (AUDIT-01 T1-T5) are not touched here.
- Not changed here: F2 (static profile sampled at HH:00) and F3 (carried value).
Measured: with price off, every command over the modes-off replay is
byte-identical. On a static tariff rebuilt on the hour it is too; rebuilt 20
min off the hour the commands differ (digest 79f06d6381ff3358 on release/1.0,
18de19d9fe919fe5 here), for the reason above. On the dynamic replay every command is
the same too; only a sale's sentence quotes a smaller "above the cheapest"
figure once the first cheap period is over. The identity pins in
`test_modes_p3.py` and `test_price_run_fixes.py` hide just that figure and the
word change (`opt11p3_identity.trade_digest()`).
## Ties (round 2)
The verdict takes `min`/`max` over the points still ahead and the first one wins
a tie, so with equal prices it is the first period still ahead, which is the
current one. Before, it was the first of the whole plan, often one that had
ended. Two effects follow, and both are the card's intent, not a side effect:
- Equal quarters (hour-constant data, or a feed that prints the same price in
several quarters) now act in every quarter of the cheap (or dear) block,
not in the first only. Test: `test_price02.py` part 6.
- After the cheapest period has passed, the next cheapest still ahead is the
target, if its spread still clears the cost. A hourly plan did this at every
refresh already (it started at the current hour); a 15-minute plan did it
only from the next hour on, because it kept the elapsed quarters of this one.
Measured with `scratch/PRICE-02/dev/tiesim.py` (refresh every 30 min from the
floored hour, verdict at 7 min into each period, threshold 0.05, old = the
verdict over the whole plan, new = this card):
| data | days | acting periods old -> new | action changes old -> new |
|---|---|---|---|
| real 15-minute day, energy-charts 2026-09-09 | 1 | 4 -> 5 | 6 -> 6 |
| real 15-minute days 2026-09-09 and -10 | 2 | 14 -> 26 | 24 -> 24 |
| EPEX 2025-26 hourly prices as four equal quarters | 12 | 18 -> 72 | 36 -> 20 |
| the same with random +-0.002 noise per quarter (no ties) | 12 | 18 -> 33 | 36 -> 48 |
So a real 15-minute day does not switch more often (24 -> 24), and a day of
equal quarters switches less. A synthetic day of noisy, tie-free quarters does
(3.0 -> 4.0 changes a day): each period that passes hands the target to the
next cheapest one ahead. Every change is still at a period edge, never within
one. The replay in `test_arbitrage.py` (synthetic year, hourly) gives the same
ledger on the old and the new code once its ties are broken (280 round trips,
85 held, both): it printed one maximum twice per day, and the old fixture
tested the first of them only because its plan began at midnight.
## Battery throughput (round 3)
On real 15-minute days the trade runs longer: about 1.5 to 2.5 times the
minutes spent charging or discharging, with fewer on/off switches (for example
78 -> 68, 92 -> 72), over 16 days (2026-09-02 to 09-17, a decision every
minute, refresh at :00 or :10, tomorrow's prices known always or from 13:00;
reviewer's sim). Through a modelled 10 kWh battery that is 27.5 -> 42.6 kWh
(refresh at :00, tomorrow always known); 19.6-31.9 -> 43.4-48.7 kWh in the
other three variants. Charge to
discharge reversals are unchanged (10 -> 10, 13 -> 13). More cycling is the
price of not pausing.
+1 -1
View File
@@ -211,7 +211,7 @@ The label and description each setting shows on the Configuration tab, in tab or
| Limits | Battery cut-out level (%) | `bms_soc_floor` | The level where the battery's own management system stops giving power. Read it off your battery rather than copying a figure. |
| Limits | Peak limit (W) | `peak_cap_w` | Your own number: the 15-minute average power the add-on keeps the grid under. Take it from your grid contract. |
| Limits | Margin below the peak limit (W) | `peak_margin_w` | The add-on aims this much below the peak limit, so a 15-minute period that ends a little high still stays under it. |
| Limits | 15-minute demand forecast entity (W) | `peak_forecast_entity` | Optional. A forecast of this 15-minute period's average power, in W or kW. With no power unit it is not used. Leave it empty and there is no peak limit. |
| Limits | 15-minute demand forecast entity | `peak_forecast_entity` | Optional. A forecast of this 15-minute period's average power, in W or kW. With no power unit it is not used. Leave it empty and there is no peak limit. |
| Limits | Peak reserve (%) | `peak_soc_floor` | Peak limiting will not draw the battery below this level, but ordinary use can. Derived value, leave it unchanged. |
| Tariff | Dynamic price source | `price_provider` | Where prices come from. Choosing a source starts the price fetch, even with trading off. |
| Tariff | Supplier markup (EUR/kWh) | `price_markup_eur_kwh` | Your supplier's margin. It is added only when the price source does not already include it. |
+10 -10
View File
@@ -172,7 +172,7 @@ IDLE_NOT_COVERED = (
IDLE_NO_SOC = (
"state of charge is not being read, and a battery whose fill nobody knows "
"cannot be told to empty or to fill")
IDLE_OFF_PEAK_HOUR = "this hour is neither the buy hour nor the sell hour"
IDLE_OFF_PEAK_HOUR = "now is not the cheapest or the dearest price period"
# OPT-09. A TRADE NOTHING CAN TAKE IS NOT A TRADE, AND THE ALTERNATIVE IS WORSE
# THAN A MISSED HOUR. `charge_w`/`discharge_w` used to be the site's whole rating
# and are now SIZED to the packs whose state of charge permits that direction, so
@@ -196,7 +196,7 @@ IDLE_HEADROOM_UNMEASURED = (
"creates the billed quantity (§9.1) and nothing is measuring the quarter "
"this one would land in")
NEGATIVE_PRICE = (
"§17: the sell hour is priced BELOW ZERO, so discharging into it would give "
"§17: the sell period is priced BELOW ZERO, so discharging into it would give "
"energy away and pay for the privilege. Arbitrage must not export at a "
"negative price")
@@ -347,7 +347,7 @@ def decide(plan, now: datetime, *, peak_guard_active, charge_headroom_w, soc,
verdict = plan.arbitrage(peak_guard_active=peak_guard_active,
min_spread_eur_kwh=min_spread_eur_kwh,
uncertainty=uncertainty)
uncertainty=uncertainty, now=now)
return verdict.match(
refused=lambda v: _idle(v.reason),
allowed=lambda v: _act(plan, now, v, soc, charge_w, discharge_w,
@@ -362,7 +362,7 @@ def _act(plan, now, v, soc, charge_w, discharge_w, floor_pct, ceiling_pct,
if here is None:
return _idle(
f"the series carries a tradeable spread but no price for this "
f"instant ({now.isoformat()}), which is §11.4 for this hour rather "
f"instant ({now.isoformat()}), which is §11.4 for this period rather "
f"than a reason to reach for a neighbouring one",
spread=v.spread_eur_kwh, required=v.required_spread_eur_kwh)
@@ -381,7 +381,7 @@ def _act(plan, now, v, soc, charge_w, discharge_w, floor_pct, ceiling_pct,
# irreversible action may rest on it.
if here.provisional:
return _idle(
f"§8.3: the price for this hour ({here.start.isoformat()}) is "
f"§8.3: the price for this period ({here.start.isoformat()}) is "
f"CARRIED FORWARD, not published, and no irreversible action may be "
f"taken on a provisional price",
spread=v.spread_eur_kwh, required=v.required_spread_eur_kwh,
@@ -399,18 +399,18 @@ def _act(plan, now, v, soc, charge_w, discharge_w, floor_pct, ceiling_pct,
# negative price reports the reason that actually forbids the trade
# rather than one that happens to fire first on some other site.
if price < 0:
return _idle(f"{NEGATIVE_PRICE} - this hour is "
return _idle(f"{NEGATIVE_PRICE} - this period is "
f"{price:+.4f} EUR/kWh", **common)
if soc <= floor_pct:
return _idle(
f"the sell hour is here at {price:.4f} EUR/kWh but the pack is "
f"the sell period is here at {price:.4f} EUR/kWh but the pack is "
f"at {soc:g} %, at or under the {floor_pct:g} % reserve floor "
f"arbitrage may not spend (§10.1 priority 4)", **common)
if not abs(float(discharge_w)) > 0.0:
return _idle(IDLE_NO_SIZE, **common)
return ArbitrageDecision(
action=DISCHARGE, power_w=abs(float(discharge_w)),
reason=(f"{DISCHARGE}: this is the dearest firm hour at "
reason=(f"{DISCHARGE}: this is the dearest firm price period at "
f"{price:.4f} EUR/kWh, {v.spread_eur_kwh:.4f} above the "
f"cheapest and clear of the {v.required_spread_eur_kwh:.4f} "
f"this cycle had to cover; discharging "
@@ -421,7 +421,7 @@ def _act(plan, now, v, soc, charge_w, discharge_w, floor_pct, ceiling_pct,
if soc >= ceiling_pct:
return _idle(
f"the buy hour is here at {price:.4f} EUR/kWh but the pack is at "
f"the buy period is here at {price:.4f} EUR/kWh but the pack is at "
f"{soc:g} %, at or over the {ceiling_pct:g} % ceiling - there is "
f"nowhere to put the cheap energy", **common)
if not abs(float(charge_w)) > 0.0:
@@ -456,7 +456,7 @@ def _act(plan, now, v, soc, charge_w, discharge_w, floor_pct, ceiling_pct,
capped = size_w < asked_w
return ArbitrageDecision(
action=CHARGE, power_w=-size_w,
reason=(f"{CHARGE}: this is the cheapest firm hour at "
reason=(f"{CHARGE}: this is the cheapest firm price period at "
f"{price:.4f} EUR/kWh, {v.spread_eur_kwh:.4f} below the dearest "
f"and clear of the {v.required_spread_eur_kwh:.4f} this cycle "
f"had to cover; charging {size_w:.0f} W from {soc:g} %"
+15 -10
View File
@@ -262,11 +262,12 @@ def _run(i, kind, start, end, day0, day1, what, why, amount, *, kwh=None,
"eur_kwh": None if price is None else round(price, 4)}
def trade_verdict(c, plan):
def trade_verdict(c, plan, now):
"""The arbitrage verdict the control cycle would reach on this plan, or a
reason there is none. The SAME call and inputs as `Controller.cycle()`:
`plan.arbitrage()` with the peak guard's state, the cycle cost and the
no-weather uncertainty. Returns `(verdict_or_None, why_key)`."""
no-weather uncertainty, and (PRICE-02) the same `now`. Returns
`(verdict_or_None, why_key)`."""
if not arbitrage.enabled(c.o):
return None, "off"
guard = arbitrage.guard_state(c.peak_guard)
@@ -276,20 +277,24 @@ def trade_verdict(c, plan):
return None, "no_prices"
verdict = plan.arbitrage(peak_guard_active=guard,
min_spread_eur_kwh=arbitrage.threshold_eur_kwh(c.o),
uncertainty=frozenset({Uncertainty.NO_WEATHER}))
uncertainty=frozenset({Uncertainty.NO_WEATHER}),
now=now)
return verdict.match(
allowed=lambda v: (v, "allowed"),
refused=lambda _v: (None, refusal_key(plan, guard)))
refused=lambda _v: (None, refusal_key(plan, guard, now)))
def refusal_key(plan, guard):
def refusal_key(plan, guard, now):
"""Which of `TariffPlan.arbitrage()`'s refusals fired, in its own order:
the peak guard, no prices, under two published points, then the gap."""
if guard:
return "peak"
if not plan.has_prices:
return "no_prices"
if sum(1 for p in _points(plan) if not p.provisional) < 2:
pts = _points(plan)
ahead = {id(p) for _s, end, p in _cover_table(pts) if end > now}
pts = [p for p in pts if id(p) in ahead]
if sum(1 for p in pts if not p.provisional) < 2:
return "not_firm"
return "not_worth"
@@ -602,7 +607,7 @@ def build_plan(c, now, shadow, running, held=False, settled=True):
limit_w = c.peak.cap_w # the limit the controller parsed at start
runs, lines = [], []
# The control cycle trades on the fresh plan only, so this does too.
verdict, trade_why = trade_verdict(c, fresh)
verdict, trade_why = trade_verdict(c, fresh, now)
bars = price_bars(seen, day0, day1, price_windows(c, fresh, verdict),
now, days.kinds)
modes_on = bool(getattr(c, "modes_on", False))
@@ -1075,9 +1080,9 @@ def _kwh(v) -> str:
return _dec(f"{float(v):.1f}") + " kWh"
# Row id -> the title on Home. The row's own sentence is the body, and it stays
# in English (app/messages.py). Five titles are the notification's own title,
# which test_ui_text.py checks.
# Row id -> the title on Home. The row's own sentence is the body, in the
# page's language (UX-46, `messages.pick`). Five titles are the notification's
# own title, which test_ui_text.py checks.
_EN = uitext.words("en")["home_words"]
ATTN_TITLES = {(None if k == "none" else k): v for k, v in _EN["attn"].items()}
ATTN_FROM_MESSAGES = {
+15 -6
View File
@@ -2369,7 +2369,8 @@ class Controller:
return plan.arbitrage(
peak_guard_active=active,
min_spread_eur_kwh=arbitrage.threshold_eur_kwh(self.o),
uncertainty=frozenset({Uncertainty.NO_WEATHER})).match(
uncertainty=frozenset({Uncertainty.NO_WEATHER}),
now=now).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
@@ -7464,7 +7465,8 @@ class Controller:
level, banner = "warn", messages.banner_stopped()
elif runs:
level, running_alarm = "bad", True
banner = runs[0].get("banner") or messages.headline(runs[0]["text"])
banner = runs[0].get("banner") or messages.headline_bi(
runs[0]["text"])
elif any(i.delivery_wd.tripped for i in self.inverters):
# ⚠️ SAFETY-42, AND IT IS A `warn` BANNER RATHER THAN A `bad` ONE,
# WHICH IS THE WHOLE "reports, does not act" DECISION IN ONE LINE.
@@ -7487,11 +7489,11 @@ class Controller:
# character; at two, "Maintenance: drain" without a name is a
# banner an operator cannot act on, because the remedy is at one
# device and the site has two.
level, banner = "warn", f"Maintenance: {self.maint.phase}"
if len(self.inverters) > 1:
banner += f" ({self.maint_unit.name})"
level, banner = "warn", messages.banner_maintenance(
self.maint.phase,
self.maint_unit.name if len(self.inverters) > 1 else "")
else:
level, banner = "ok", "Running — holding grid at zero"
level, banner = "ok", messages.banner_running()
def w(v):
return "—" if v is None else f"{v:,.0f} W".replace(",", " ")
@@ -7595,6 +7597,13 @@ class Controller:
now=datetime.now(timezone.utc), why=self.why_words)),
]
held = level == "bad" and not running_alarm
# UX-46. The rows were built in English with their Dutch beside them
# (`messages.Bi`); the page's language is chosen once, here, for the
# Home sentences, the checks list, the rows that carry one and the
# banner. A plain str (no Dutch twin) stays as it is.
checks = [dict(c, text=messages.pick(c["text"], lang)) for c in checks]
rows = [(k, messages.pick(v, lang)) for k, v in rows]
banner = messages.pick(banner, lang)
out = {
"banner": banner, "level": level, "rows": rows, "checks": checks,
"hint": self.hint,
File diff suppressed because it is too large Load Diff
+4 -2
View File
@@ -54,10 +54,12 @@ class SetupOnly:
def status_payload(self, lang: str = "en") -> dict:
return {"setup_only": True, "mode": "setup_only",
"banner": messages.banner_setup_only(), "level": "warn",
"banner": messages.pick(messages.banner_setup_only(),
lang), "level": "warn",
"rows": [["Control", "not started, in setup"]],
"checks": [{"ok": False, "warn": True,
"text": messages.row_setup_only()}],
"text": messages.pick(messages.row_setup_only(),
lang)}],
"hint": self.hint, "events": list(self.events)}
async def handle_action(self, what):
+29 -3
View File
@@ -207,7 +207,7 @@ from datetime import datetime, timedelta, timezone
from . import messages, price_chain
from .price import (
MAX_PLAUSIBLE_EUR_KWH, Absence, PricePoint, PriceProduct, PriceRead,
Uncertainty, _Outcome, brussels, load_prices, price_option,
Uncertainty, _Outcome, _cover_table, brussels, load_prices, price_option,
provider_configured, required_spread, uncertainty_of, widen_factor,
)
from .price_chain import uses_chain
@@ -986,12 +986,25 @@ class TariffPlan:
return here.eur_per_kwh < (sum(p.eur_per_kwh for p in firm) / len(firm))
def arbitrage(self, *, peak_guard_active: bool, min_spread_eur_kwh,
uncertainty) -> Arbitrage:
uncertainty, now: datetime) -> Arbitrage:
"""May OPT-04 trade, and on what numbers. See the module docstring.
⚠️ ALL THREE ARGUMENTS ARE REQUIRED AND KEYWORD-ONLY, for one reason
PRICE-02. `now` (tz-aware) drops the points whose price period has
ended (`end <= now`, the span `price._cover_table()` gives it), so the
cheapest and dearest point are always ones that can still be traded.
It is taken here, per call, and not when the plan is built: a plan
built at 03:00 is still used at 03:20. An hourly point that is still
current stays in.
⚠️ ALL FOUR ARGUMENTS ARE REQUIRED AND KEYWORD-ONLY, for one reason
each, and none of them has a default:
`now` PRICE-02. A caller that could leave it out
would get the whole series back, elapsed
periods included, and the verdict could aim
at a period that is over. A test that wants
the whole series passes an instant before it.
`peak_guard_active` §9.1/§10.1 put peak shaving above arbitrage,
and this is how that ordering survives a
strategy that has not been written yet: the
@@ -1029,6 +1042,12 @@ class TariffPlan:
f"{sorted(map(repr, caller_sources))} - a string widens by "
f"nothing while reporting that it widened.")
if not isinstance(now, datetime) or now.tzinfo is None \
or now.utcoffset() is None:
raise TypeError(f"now must be a tz-aware datetime, not {now!r}; "
f"a naive instant is ambiguous for one hour "
f"every October")
if peak_guard_active:
return Arbitrage.refuse(
"the capacity-tariff peak guard is holding, and §10.1 puts peak "
@@ -1056,6 +1075,8 @@ class TariffPlan:
# day with no fetch at all) has no firm pair left and is refused, which
# is the previous behaviour arrived at by a truer route.
pts = self.read.points
ahead = {id(p) for _s, end, p in _cover_table(pts) if end > now}
pts = tuple(p for p in pts if id(p) in ahead)
firm = tuple(p for p in pts if not p.provisional)
if len(firm) < 2:
# ⚠️ POINTS, AND THE HOUR FIGURE BESIDE IT RATHER THAN INSTEAD OF
@@ -1334,6 +1355,11 @@ async def resolve_tariff(opts: dict, since: datetime, until: datetime,
that, so the spread is over hours that are still ahead. Both directions of
the difference are real: the window gains the unpublished hours it was
missing and loses the elapsed ones it should never have had.
⚠️ PRICE-02: THAT HOLDS FOR AN HOURLY FEED ONLY. The window is floored to
the hour, so a 15-minute feed keeps the elapsed periods of the current
hour, and the plan ages between refreshes. `TariffPlan.arbitrage(now=...)`
drops the periods that have ended, at every call.
"""
stamp = now or datetime.now(timezone.utc)
if not provider_configured(opts):
+2 -2
View File
@@ -17,7 +17,7 @@ from . import uilayout as L
from . import uitext
from . import uimodel as U
from .configui import help_of, label_of, problems_by_field, split_group
from .validate import power_request_entities
from .validate import note_text, power_request_entities
# A choice the page cannot make, only keep. `price_service` needs a token, and
# the token is a secret this page never draws (uilayout.HIDDEN), so it is set on
@@ -314,7 +314,7 @@ def anchors(problems, notes, model, keys, lang="en") -> list:
add(None, None, "bad", text)
for note in notes:
tier = note.tier if note.tier in _TIERS else "info"
text = messages.NOTE_NL.get(note.text, note.text) if lang == "nl" else note.text
text = note_text(note, lang)
for key in note.keys or (None,):
for field, card in (targets(key, note.text, model) if key
else [(None, None)]):
+129 -39
View File
@@ -386,12 +386,23 @@ class Note(NamedTuple):
# also "ok" and "bad" (setup_checks)
keys: tuple
text: str
nl: str = "" # UX-46: the same sentence in Dutch
def _advise(notes, tier: str, keys, text: str) -> None:
"""Add a Note when the caller collects them (the settings page)."""
def _advise(notes, tier: str, keys, text: str, nl: str = "") -> None:
"""Add a Note when the caller collects them (the settings page). `nl` is
the Dutch twin of `text`; the page picks one (`note_text()`)."""
if notes is not None:
notes.append(Note(tier, tuple(keys), text))
notes.append(Note(tier, tuple(keys), text, nl))
def note_text(note, lang: str = "en") -> str:
"""The note's sentence in the page's language, English when it has no
Dutch twin."""
if lang == "nl":
return getattr(note, "nl", "") or messages.NOTE_NL.get(
note.text, note.text)
return note.text
# ⚠️ A MIRROR OF A FIRMWARE CONSTANT, AND IT CANNOT BE IMPORTED. The firmware
@@ -628,6 +639,11 @@ def S(key: str, where: str = "") -> str:
return messages.setting(key, where.strip().strip("()"))
def SN(key: str, where: str = "") -> str:
"""`S()` with the Dutch label, for the Dutch twin of a note."""
return messages.setting(key, where.strip().strip("()"), "nl")
def _labels(keys, joiner: str = "and") -> str:
"""Several settings, each with its key: the config panel files a refusal
under every key it names (`configui.problems_by_field()`)."""
@@ -1216,7 +1232,9 @@ def validate_startup_config(opts: dict, dev_min=None, dev_max=None,
integ_w, max_w)
_advise(notes, "warn", ("integrator_max_w",),
f"{S('integrator_max_w')} is above {S('max_w')}, which only "
"adds overshoot, so set it to 0.")
"adds overshoot, so set it to 0.",
nl=f"{SN('integrator_max_w')} is hoger dan {SN('max_w')}, wat "
"alleen doorschieten geeft, dus zet het op 0.")
# ⚠️ THE SIBLING OF RULE 6, AND NOT THE SAME CASE - see the docstring. The
# drain phase OWNS the actuator (`Claim.set` at P_MAINTENANCE), and a SET is
@@ -1244,7 +1262,10 @@ def validate_startup_config(opts: dict, dev_min=None, dev_max=None,
_advise(notes, "warn", ("maintenance_discharge_w",),
f"{S('maintenance_discharge_w')} is above "
f"{S('max_w', maint_where)}, so the battery empties at "
f"{messages.watts(maint_cap)} and the cycle takes longer.")
f"{messages.watts(maint_cap)} and the cycle takes longer.",
nl=f"{SN('maintenance_discharge_w')} is hoger dan "
f"{SN('max_w', maint_where)}, dus de batterij loopt leeg met "
f"{messages.watts(maint_cap)} en de cyclus duurt langer.")
_LOG.debug(
"Not refused: the drain phase sets the setpoint, the arbiter "
"clamps it to %.0f W and names the clamp, and the drain phase has "
@@ -1473,7 +1494,9 @@ def _price_problems(opts: dict, notes: list | None = None) -> list[str]:
TAXES_PROFILE_OPTION, negative)
_advise(notes, "warn", (TAXES_PROFILE_OPTION,),
f"{S(TAXES_PROFILE_OPTION)} has negative figures, which "
"lower the price, and that is usually a sign error.")
"lower the price, and that is usually a sign error.",
nl=f"{SN(TAXES_PROFILE_OPTION)} heeft negatieve bedragen, "
"die de prijs verlagen, en dat is meestal een tekenfout.")
# ⚠️ TARIFF-15: THE FORMULA, GRADED AGAINST THE OPERATOR'S OWN CARD, AT
# STARTUP RATHER THAN AT 03:00.
@@ -1532,7 +1555,10 @@ def _price_problems(opts: dict, notes: list | None = None) -> list[str]:
_advise(notes, "warn", ("price_formula", "price_provider"),
f"{S('price_formula')} is not used, because "
f"{S('price_provider')} is not 'custom_formula', so clear "
"it or choose that source.")
"it or choose that source.",
nl=f"{SN('price_formula')} wordt niet gebruikt, omdat "
f"{SN('price_provider')} niet 'custom_formula' is, dus wis "
"ze of kies die bron.")
if spec is not None and pid == "custom_formula":
taxes = _number(opts, "price_taxes_eur_kwh",
PRICE_OPTION_DEFAULTS["price_taxes_eur_kwh"])
@@ -1547,7 +1573,8 @@ def _price_problems(opts: dict, notes: list | None = None) -> list[str]:
# cannot start.
_LOG.warning("%s", aged)
_advise(notes, "warn", ("price_formula",),
_card_note("price_formula", spec))
_card_note("price_formula", spec),
nl=_card_note("price_formula", spec, "nl"))
# ⚠️ TARIFF-16: THE INJECTION LEG, REFUSED AT STARTUP ON THE SAME RULE.
#
@@ -1630,7 +1657,10 @@ def _price_problems(opts: dict, notes: list | None = None) -> list[str]:
xspec.offset_eur_kwh)
_advise(notes, "info", (EXPORT_FORMULA_OPTION,),
f"{S(EXPORT_FORMULA_OPTION)} has no card block, so "
"only the shape of its numbers is checked.")
"only the shape of its numbers is checked.",
nl=f"{SN(EXPORT_FORMULA_OPTION)} heeft geen kaartblok, "
"dus alleen de vorm van de getallen wordt "
"gecontroleerd.")
_LOG.debug(
"Accepted because some suppliers (Eneco) print the "
"injection formula with no price beside it. If your card "
@@ -1643,7 +1673,8 @@ def _price_problems(opts: dict, notes: list | None = None) -> list[str]:
if xaged:
_LOG.warning("%s", xaged)
_advise(notes, "warn", (EXPORT_FORMULA_OPTION,),
_card_note(EXPORT_FORMULA_OPTION, xspec))
_card_note(EXPORT_FORMULA_OPTION, xspec),
nl=_card_note(EXPORT_FORMULA_OPTION, xspec, "nl"))
# ⚠️ A WARNING, NOT A REFUSAL, AND THE ASYMMETRY IS DELIBERATE. §11.4 says a
# site with no price data falls back to the static profile; a dynamic site
@@ -1657,13 +1688,16 @@ def _price_problems(opts: dict, notes: list | None = None) -> list[str]:
"fallback.", pid)
_advise(notes, "warn", ("static_profile",),
f"{S('static_profile')} is empty, so the site has no tariff "
"at all while the price feed is down.")
"at all while the price feed is down.",
nl=f"{SN('static_profile')} is leeg, dus de installatie heeft "
"helemaal geen tarief zolang de prijsfeed uitvalt.")
# PRICE-01: not a fault, so an info note and no log line here (the add-on
# logs it once at start). With `price_service` the fixed tariff is only the
# plan's last fallback; the cheap window and the gate do not read it.
if pid == "price_service" and str(opts.get("static_profile", "") or "").strip():
_advise(notes, "info", ("static_profile",), messages.FIXED_TARIFF_NOTE)
_advise(notes, "info", ("static_profile",), messages.FIXED_TARIFF_NOTE,
nl=messages.FIXED_TARIFF_NOTE_NL)
# ⚠️ OPT-04. THE THREE WAYS `arbitrage_enabled: true` CAN BE A LIE, refused
# rather than logged, because a strategy that is armed in the UI and cannot
@@ -1735,7 +1769,11 @@ def _price_problems(opts: dict, notes: list | None = None) -> list[str]:
return problems
def _card_note(option: str, spec) -> str:
def _card_note(option: str, spec, lang: str = "en") -> str:
if lang == "nl":
return (f"{SN(option)} gebruikt een tariefkaart van "
f"{spec.card_month}, dus neem de nieuwe cijfers over als uw "
"contract veranderde.")
return (f"{S(option)} uses a tariff card dated {spec.card_month}, so copy "
"the new figures if your contract changed.")
@@ -1810,7 +1848,10 @@ def config_notes(opts: dict, *, entities: dict | None = None,
if month_peak_w is not None and cap is not None and cap < month_peak_w:
_advise(notes, "warn", ("peak_cap_w",),
f"This month's peak is already {messages.watts(month_peak_w)}, "
f"so a lower {S('peak_cap_w')} saves nothing until next month.")
f"so a lower {S('peak_cap_w')} saves nothing until next month.",
nl=f"De piek van deze maand is al "
f"{messages.watts(month_peak_w)}, dus een lagere "
f"{SN('peak_cap_w')} bespaart niets tot volgende maand.")
return notes
@@ -1823,31 +1864,43 @@ def _entity_notes(notes, opts: dict, entities: dict, bad=()) -> None:
if row is None:
_advise(notes, "bad" if key in bad else "warn", (key,),
f"Home Assistant has no entity {entity}, which "
f"{S(key, where)} names.")
f"{S(key, where)} names.",
nl=f"Home Assistant kent {entity} niet, maar "
f"{SN(key, where)} verwijst ernaar.")
continue
unit = str(row.get("unit") or "")
if key in WATT_ENTITY_KEYS and unit in pv.WATTS_PER and unit != "W":
_advise(notes, "warn", (key,),
f"{entity} reads in {unit}, but {S(key, where)} is "
"taken as watts, so pick a sensor in W.")
"taken as watts, so pick a sensor in W.",
nl=f"{entity} meldt in {unit}, maar {SN(key, where)} wordt "
"als watt gelezen, dus kies een sensor in W.")
if key == "peak_forecast_entity" and unit not in pv.WATTS_PER:
_advise(notes, "warn", (key,),
messages.peak_forecast_unit(unit) + " Pick a sensor in "
"W or kW.")
"W or kW.",
nl=messages.peak_forecast_unit(unit, "nl")
+ " Kies een sensor in W of kW.")
if key in FORECAST_KEYS and unit not in solarfc.KWH_PER:
_advise(notes, "warn", (key,),
f"{entity} does not read in kWh, so pick a solar "
"forecast sensor in kWh.")
"forecast sensor in kWh.",
nl=f"{entity} meet niet in kWh, dus kies een sensor voor "
"de zonneprognose in kWh.")
today = solarfc.entities(opts)
tomorrow = solarfc.entities(opts, tomorrow=True)
if (today or tomorrow) and \
opts.get("battery_modes_enabled") is not True:
_advise(notes, "info", ("pv_forecast_entity",),
"The solar forecast is used only with battery modes on.")
"The solar forecast is used only with battery modes on.",
nl="De zonneprognose wordt alleen gebruikt met batterijmodi "
"aan.")
if tomorrow and not today:
_advise(notes, "warn", ("pv_forecast_tomorrow_entity",),
"Tomorrow's forecast is used only with today's, so set "
f"{S('pv_forecast_entity')} too.")
f"{S('pv_forecast_entity')} too.",
nl="De prognose voor morgen wordt alleen met die van vandaag "
f"gebruikt, dus stel ook {SN('pv_forecast_entity')} in.")
# --------------------------------------------------------------------------- #
@@ -1926,7 +1979,9 @@ def setup_checks(opts: dict, *, entities: dict | None, running=None,
if entities is None:
# Red: only the list can tell a missing power request entity (SAFETY-77).
_advise(notes, "bad", ("setpoint_entity",), "Home Assistant is not "
"answering, so press Check again once it is back.")
"answering, so press Check again once it is back.",
nl="Home Assistant antwoordt niet, dus druk op Opnieuw "
"nakijken zodra het terug is.")
else:
# A missing meter entity is red too (UX-15): with the kettle test
# skipped, nothing else on the way to the save would catch it.
@@ -1948,13 +2003,16 @@ def setup_checks(opts: dict, *, entities: dict | None, running=None,
solar.append(None)
_advise(notes, "warn", ("pv_invert_entities",),
f"{entity} reads below zero now, so turn on its "
"“Reads negative” switch in the Solar step.")
"“Reads negative” switch in the Solar step.",
nl=f"{entity} meet nu onder nul, dus zet de schakelaar "
"“Meet negatief” aan bij de stap Zonnepanelen.")
else:
w = -w if entity in flipped else w
solar.append(w)
_advise(notes, "ok", ("pv_entity",),
f"{entity} reads {_signed(w)} of solar production "
"now.")
"now.",
nl=f"{entity} meet nu {_signed(w)} zonneproductie.")
# A battery sensor that reads the wrong way while giving power makes the
# house use less than nothing. Only with every term known: a missing
# solar reading could be the difference.
@@ -1964,7 +2022,10 @@ def setup_checks(opts: dict, *, entities: dict | None, running=None,
_advise(notes, "warn", ("batt_invert",),
f"The readings add up to {_signed(sum(known))} of house "
"use, so check each "
"battery's “Battery power sign is reversed” switch in step 2.")
"battery's “Battery power sign is reversed” switch in step 2.",
nl=f"De metingen tellen op tot {_signed(sum(known))} "
"huisverbruik, dus controleer bij elke batterij de schakelaar "
"“Teken batterijvermogen omkeren” in stap 2.")
_price_check(notes, opts, running, prices_known)
return notes
@@ -1988,7 +2049,9 @@ def _meter_check(notes, opts, entities, running, meter_now_w):
return float(meter_now_w)
_advise(notes, "info", ("meter_source",), "The add-on reads "
"this meter itself, so Home shows its reading after "
"the restart.")
"the restart.",
nl="De add-on leest deze meter zelf, dus Start toont de "
"meting na de herstart.")
return None
if entities is None:
return None
@@ -2001,7 +2064,9 @@ def _meter_check(notes, opts, entities, running, meter_now_w):
if w is None:
_advise(notes, "warn", ("meter_source",),
f"{entity} reads {text} now, so check it in Home "
"Assistant before you start.")
"Assistant before you start.",
nl=f"{entity} meldt nu {text}, dus controleer ze in Home "
"Assistant voor u start.")
return None
total += sign * w
if source == "off" and opts.get("meter_invert"):
@@ -2012,12 +2077,14 @@ def _meter_check(notes, opts, entities, running, meter_now_w):
def _meter_line(notes, w) -> None:
if abs(w) < 1.0:
text = "The meter reads 0 W now."
text, nl = "The meter reads 0 W now.", "De meter meldt nu 0 W."
elif w > 0:
text = f"The meter reads {messages.watts(w)} taken from the grid now."
nl = f"De meter meldt nu {messages.watts(w)} afname van het net."
else:
text = f"The meter reads {messages.watts(w)} fed into the grid now."
_advise(notes, "ok", ("meter_source",), text)
nl = f"De meter meldt nu {messages.watts(w)} injectie op het net."
_advise(notes, "ok", ("meter_source",), text, nl=nl)
def _card_checks(notes, who, entry, entities):
@@ -2025,40 +2092,55 @@ def _card_checks(notes, who, entry, entities):
positive while giving power, or None when it is not known."""
if entities is None:
return None
wie = who.replace(" for ", " voor ", 1)
setpoint = str(entry.get("setpoint_entity") or "").strip()
if setpoint and setpoint in entities:
_advise(notes, "ok", ("setpoint_entity",),
f"Home Assistant has {setpoint}, the power request "
f"entity{who}.")
f"entity{who}.",
nl=f"Home Assistant heeft {setpoint}, de entiteit voor de "
f"vermogensvraag{wie}.")
soc = str(entry.get("soc_entity") or "").strip()
if not soc:
_advise(notes, "warn", ("soc_entity",), f"There is no battery "
f"level entity{who}, so pick one in step 2 or the add-on "
"holds 0 W.")
"holds 0 W.",
nl=f"Er is geen entiteit voor het laadniveau{wie}, dus kies er "
"een in stap 2 of de add-on houdt 0 W aan.")
elif soc in entities:
level, _unit = _reading(entities, soc)
if level is None or not 0.0 <= level <= 100.0:
_advise(notes, "warn", ("soc_entity",), f"The battery "
f"level{who} reads nothing now, so check {soc}.")
f"level{who} reads nothing now, so check {soc}.",
nl=f"Het laadniveau{wie} meldt niets, dus controleer "
f"{soc}.")
else:
_advise(notes, "ok", ("soc_entity",),
f"The battery level{who} reads {level:.0f} %.")
f"The battery level{who} reads {level:.0f} %.",
nl=f"Het laadniveau{wie} meldt {level:.0f} %.")
batt = str(entry.get("batt_entity") or "").strip()
if batt and batt in entities:
w, _unit = _reading(entities, batt)
if w is None:
_advise(notes, "warn", ("batt_entity",), f"The battery "
f"power{who} reads nothing now, so check {batt}.")
f"power{who} reads nothing now, so check {batt}.",
nl=f"Het batterijvermogen{wie} meldt niets, dus controleer "
f"{batt}.")
return None
w = -w if entry.get("batt_invert") else w
if abs(w) < IDLE_W:
_advise(notes, "info", ("batt_entity",), f"The battery{who} is "
"idle now, so check its direction on Home later.")
"idle now, so check its direction on Home later.",
nl=f"De batterij{wie} staat nu stil, dus controleer later "
"op Start de richting.")
else:
doing = "giving power" if w > 0 else "charging"
doet = "levert" if w > 0 else "laadt"
_advise(notes, "info", ("batt_invert",),
f"The battery{who} reads {messages.watts(w)} {doing} "
"now, so check the inverter shows the same.")
"now, so check the inverter shows the same.",
nl=f"De batterij{wie} {doet} nu {messages.watts(w)}, dus "
"controleer of de omvormer hetzelfde toont.")
return w
return None
@@ -2070,20 +2152,28 @@ def _price_check(notes, opts, running, prices_known) -> None:
if profile:
_advise(notes, "ok", ("static_profile",),
"Your prices come from your contract, so they are "
"always known.")
"always known.",
nl="Uw prijzen komen uit uw contract, dus ze zijn altijd "
"bekend.")
else:
_advise(notes, "info", ("price_provider",),
"No prices are set, so the batteries do not charge "
"on price.")
"on price.",
nl="Er zijn geen prijzen ingesteld, dus de batterijen "
"laden niet op prijs.")
return
same = (isinstance(running, dict)
and str(running.get("price_provider") or "") == provider)
if same and prices_known:
_advise(notes, "ok", ("price_provider",),
f"Prices for today are in, for {prices_known} of 24 hours.")
f"Prices for today are in, for {prices_known} of 24 hours.",
nl=f"De prijzen van vandaag zijn binnen, voor {prices_known} "
"van 24 uur.")
else:
_advise(notes, "info", ("price_provider",),
"Prices are fetched after the restart, and Home shows them.")
"Prices are fetched after the restart, and Home shows them.",
nl="De prijzen worden na de herstart opgehaald, en Start toont "
"ze.")
def _margin_problems(opts: dict) -> list[str]:
+14 -6
View File
@@ -51,7 +51,7 @@ from aiohttp import web
from . import messages, price, selfconfig, uipage, uitext
from .haentities import EntityCache
from .homeview import priced_hours, region_of
from .validate import config_notes, setup_checks
from .validate import config_notes, note_text, setup_checks
from .config_page import CONFIG_PAGE
from .configui import (changed_keys, field_model, hide_secrets, keep_secrets,
problems_by_field, secret_keys)
@@ -105,6 +105,9 @@ def page() -> str:
DEVICE_NOTE = (
"%s did not report its power range, so two checks were skipped, here and "
"at start. If they fail later, the add-on will not start then.")
DEVICE_NOTE_NL = (
"%s meldde geen vermogensbereik, dus twee controles zijn overgeslagen, "
"hier en bij het starten. Falen ze later, dan start de add-on niet.")
def build_app(controller, selfcfg=None) -> web.Application:
@@ -301,22 +304,27 @@ def build_app(controller, selfcfg=None) -> web.Application:
# UX-13: warnings and info beside the refusals. Never block a save.
# The entity checks use the picker's cached list and never wait on HA.
notes += config_notes(options, entities=entities.peek())
# UX-46: one language choice for everything below that has words.
lang = uitext.lang_of(_body.get("lang"))
payload = {"problems": problems, "by_key": by_key,
"unanchored": unanchored,
"notes": [n._asdict() for n in notes]}
if unread:
payload["device_note"] = DEVICE_NOTE % ", ".join(sorted(set(unread)))
payload["device_note"] = (DEVICE_NOTE_NL if lang == "nl"
else DEVICE_NOTE) % ", ".join(
sorted(set(unread)))
if model is not None:
payload["messages"] = (
uipage.derive_messages(errors, problems, notes, model,
_keys_of(options),
uitext.lang_of(_body.get("lang")))
_keys_of(options), lang)
if errors
else uipage.anchors(problems, notes, model, _keys_of(options),
uitext.lang_of(_body.get("lang"))))
lang))
payload["steps"] = uipage.step_problems(payload["messages"], model)
if _body.get("setup"):
payload["setup"] = [n._asdict() for n in await _setup_checks(options)]
payload["setup"] = [
dict(n._asdict(), text=note_text(n, lang))
for n in await _setup_checks(options)]
return web.json_response(payload)
async def _setup_checks(options):
+63 -20
View File
@@ -109,6 +109,11 @@ from app.validate import validate_startup_config
UTC = timezone.utc
HERE = os.path.dirname(os.path.abspath(__file__))
# PRICE-02: `TariffPlan.arbitrage(now=...)` is required. This instant is before every
# point these fixtures build, so no period has ended and the whole series counts.
from datetime import datetime as _DT, timezone as _TZ # noqa: E402
LONG_AGO = _DT(2000, 1, 1, tzinfo=_TZ.utc)
fails = []
@@ -194,13 +199,22 @@ NO_WEATHER = frozenset({Uncertainty.NO_WEATHER})
def plan_of(prices, *, product=PriceProduct.DAY_AHEAD, start=D0,
provisional=False, provider="generic_dayahead"):
provisional=False, provider="generic_dayahead", days=2):
"""A TariffPlan over consecutive hourly prices starting at `start`.
`provisional` may be a single bool or a per-hour sequence.
PRICE-02: the verdict skips price periods that have ended, so a plan
probed at 18:30 no longer sees the cheap hour at 03:00 of the same day. A
running site holds 24-48 h, so `days=2` repeats the day (tomorrow's
prices are today's): probes in day one keep their extremes, and the
first cheapest/dearest hour still ahead is today's. Pass `days=1` for a
plan that must be exactly the given hours.
"""
prices = list(prices) * days
flags = ([bool(provisional)] * len(prices)
if isinstance(provisional, bool) else list(provisional))
if isinstance(provisional, bool)
else list(provisional) * days)
pts = [PricePoint(start=brussels(start + timedelta(hours=i)),
eur_per_kwh=round(float(p), 6), provisional=f)
for i, (p, f) in enumerate(zip(prices, flags))]
@@ -344,7 +358,7 @@ check(f"a realistic Belgian day/night profile has a spread of "
f"{_st_spread!r} EUR/kWh - the number the product decision is about",
near(_st_spread, 0.08))
_shipped = PRICE_OPTION_DEFAULTS["cycle_cost_threshold_eur"]
_st_verdict = derived(lambda: _st_plan.arbitrage(
_st_verdict = derived(lambda: _st_plan.arbitrage(now=LONG_AGO,
peak_guard_active=False, min_spread_eur_kwh=_shipped,
uncertainty=NO_WEATHER))
check(f"...and at the SHIPPED threshold of {_shipped} EUR/kWh that site does "
@@ -361,7 +375,7 @@ check(f"...and it is the SPREAD rung that refuses it, not a flat-tariff or a "
check("...while the SAME site trades once the threshold is set below its "
"spread, so the refusal above is about the number and not about static "
"tariffs being unsupported",
derived(lambda: _st_plan.arbitrage(
derived(lambda: _st_plan.arbitrage(now=LONG_AGO,
peak_guard_active=False,
min_spread_eur_kwh=_st_spread / widen_factor(NO_WEATHER) - 1e-4,
uncertainty=NO_WEATHER).kind) == "allowed")
@@ -1644,7 +1658,7 @@ check("a FULLY carried series trades nothing at all, whatever its spread - "
# standing if a future caller ever hands `_act` a plan the verdict did not come
# from.
_probe = _arb._point_at(PLAN, hour(CHEAP_H))
_verdict = PLAN.arbitrage(peak_guard_active=False, min_spread_eur_kwh=0.0,
_verdict = PLAN.arbitrage(now=LONG_AGO, peak_guard_active=False, min_spread_eur_kwh=0.0,
uncertainty=frozenset())
check("the point matched for the buy hour IS the verdict's own cheapest point, "
"by identity - which is why `_act`'s provisional guard is unreachable "
@@ -1665,6 +1679,14 @@ print("4b. §10.2 hysteresis - a price on the boundary cannot flap at 30 s")
# threshold repeatedly, hour to hour, which is the shape that makes a
# per-tick price-vs-threshold comparison flap.
FLAP = [0.20 + (0.101 if i % 2 else 0.0) for i in range(24)]
# PRICE-02: the verdict skips periods that have ended, and this day had ties
# (0.20 at every even hour, 0.301 at every odd one). With ties the first hour
# still ahead is the extreme, so the verdict moves on every hour - once per
# hour, never per tick, which test_price02.py part 6 grades. This check is
# about a verdict that stays put while the prices keep crossing the threshold,
# so the day gets one cheapest and one dearest hour at its end (22: 0.60,
# 23: 0.05), unique and ahead of every probe until they come.
FLAP[22], FLAP[23] = 0.60, 0.05
_flap_plan = plan_of(FLAP)
check("the flapping fixture really does straddle the threshold hour by hour - "
"every neighbouring pair differs by more than it",
@@ -1846,7 +1868,15 @@ def replay(threshold, unc, *, soc0=50.0, floor=20.0, ceiling=95.0):
neg_sells = 0
for n in range(365):
day0 = REPLAY_START + timedelta(days=n)
prices = synth_day(n)
# PRICE-02: no two hours share a price. The synthetic day is
# symmetric, so it prints the same maximum twice; the verdict now takes
# the first one still AHEAD, where it used to take the first of the
# day (a tie, graded on its own in test_price02.py part 6). So the
# ties are broken by a jitter of at most 5e-5 EUR/kWh that has no
# direction: not later-is-dearer, which would send every day to a
# buy-then-sell and hide the selectivity this replay checks.
prices = [round(p + 1e-6 * ((37 * h + 101 * n) % 50), 6)
for h, p in enumerate(synth_day(n))]
plan = plan_of(prices, start=day0)
open_buy = None
for h in range(24):
@@ -2056,18 +2086,26 @@ check(f"BUT NOT ONE OF THE 365 DAYS HAS A NEGATIVE DEAREST HOUR "
def replay_real(days, threshold, unc, *, soc0=50.0, floor=20.0, ceiling=95.0):
"""The shipped `decide()` over the real year. Returns the ledger, nothing else.
One `TariffPlan` per Brussels day, so "the cheapest hour" means the cheapest
hour of that day - the same shape as section 7's replay, and the same shape
a site with a day-ahead feed actually sees.
The plan is built the way production builds it: one per hour, from that
hour (the refresh floors to the hour) to the end of the Brussels day,
plus the next day from 13:00 Brussels, when the day-ahead prices are
published. So a site sees 11 to 35 hours ahead, never a whole day behind
it and never a day on its own. (An earlier version built one plan per
calendar day with no tomorrow: after the cheap hour no cheap hour was left
ahead, and the ledger shrank to a few trades that no real site would see.)
An open buy is kept over midnight: with tomorrow in view most buys are
sold the next day or later (46 of 51 at the shipped threshold).
"""
soc = soc0
trades, holdings, bad, neg_sells = [], [], [], 0
for day in days:
plan = TariffPlan(
PriceRead.series(day, PriceProduct.DAY_AHEAD, "generic_dayahead",
day[0].start), None, "provider")
open_buy = None
for p in day:
open_buy = None
for di, day in enumerate(days):
nxt = days[di + 1] if di + 1 < len(days) else []
for i, p in enumerate(day):
ahead = day[i:] + (nxt if brussels(p.start).hour >= 13 else [])
plan = TariffPlan(
PriceRead.series(ahead, PriceProduct.DAY_AHEAD,
"generic_dayahead", p.start), None, "provider")
d = decide(plan, p.start + timedelta(minutes=30),
peak_guard_active=False, soc=soc,
min_spread_eur_kwh=threshold, uncertainty=unc,
@@ -2444,7 +2482,7 @@ check("...and every one of its points is provisional, so firm_points() is "
"a fresh observation",
isinstance(_t12_fb, TariffPlan) and _t12_fb.provisional is True
and _t12_fb.firm_points() == ()
and _t12_fb.arbitrage(peak_guard_active=False, min_spread_eur_kwh=0.0,
and _t12_fb.arbitrage(now=LONG_AGO, peak_guard_active=False, min_spread_eur_kwh=0.0,
uncertainty=NO_WEATHER).kind == "refused")
_t12_f.tariff_plan, _t12_f.tariff_at = _t12_good, datetime.now(UTC) - \
timedelta(seconds=1800)
@@ -2789,7 +2827,7 @@ check(f"the shipped rule and this one give DIFFERENT points at both probes - "
print()
print("10b. the buy/sell identity test matches the actual extremum interval")
_qv = QPLAN.arbitrage(peak_guard_active=False, min_spread_eur_kwh=_shipped,
_qv = QPLAN.arbitrage(now=LONG_AGO, peak_guard_active=False, min_spread_eur_kwh=_shipped,
uncertainty=NO_WEATHER)
check("the verdict's cheapest and dearest are the two QUARTERS, not the hours "
"containing them - `static_tariff` reduces over every firm point and "
@@ -2921,7 +2959,7 @@ check(f"...and its firm half's extremes are quarters {_mix_cheap_i} and "
f"{_mix_dear_i}, neither on a whole hour, so the identity test below is "
f"about the quarter and not about the hour",
(_mix_cheap_i - MIX_N_H) % 4 != 0 and (_mix_dear_i - MIX_N_H) % 4 != 0)
_mv = MIX_PLAN.arbitrage(peak_guard_active=False, min_spread_eur_kwh=_shipped,
_mv = MIX_PLAN.arbitrage(now=LONG_AGO, peak_guard_active=False, min_spread_eur_kwh=_shipped,
uncertainty=NO_WEATHER)
_mix_buy_t = qprobe(_mix_cheap_i, gaps=MIX_GAPS)
_mix_sell_t = qprobe(_mix_dear_i, gaps=MIX_GAPS)
@@ -2933,14 +2971,19 @@ check("the verdict over a MIXED series still names firm quarters, and "
is MIX_PTS[_mix_cheap_i]
and derived(lambda: _arb._point_at(MIX_PLAN, _mix_sell_t))
is MIX_PTS[_mix_dear_i])
# PRICE-02: the same day twice, as a running site holds 24-48 h (see plan_of).
# The probes stay in day one and keep their extremes: once the cheap quarter is
# over, the verdict at the dear quarter needs a cheap one still ahead.
MIX2_PLAN = qplan_of(MIX_PRICES * 2, provisional=MIX_FLAGS * 2,
gaps=MIX_GAPS + [900] + MIX_GAPS)
check("...so the mixed day CHARGES and DISCHARGES in its firm quarters, at "
"their own prices",
decide(MIX_PLAN, _mix_buy_t, peak_guard_active=False, soc=50.0,
decide(MIX2_PLAN, _mix_buy_t, peak_guard_active=False, soc=50.0,
min_spread_eur_kwh=_shipped, uncertainty=NO_WEATHER,
charge_headroom_w=HEADROOM_CLEAR,
charge_w=MAX_W, discharge_w=MAX_W, soc_floor_pct=20.0,
soc_ceiling_pct=95.0).action == CHARGE
and decide(MIX_PLAN, _mix_sell_t, peak_guard_active=False, soc=50.0,
and decide(MIX2_PLAN, _mix_sell_t, peak_guard_active=False, soc=50.0,
min_spread_eur_kwh=_shipped, uncertainty=NO_WEATHER,
charge_headroom_w=HEADROOM_CLEAR,
charge_w=MAX_W, discharge_w=MAX_W, soc_floor_pct=20.0,
+2 -2
View File
@@ -264,7 +264,7 @@ check("trading on price: a charge block at the cheapest hour (13:00) and a "
check("...placed where the control cycle's own verdict puts them: the same "
"plan.arbitrage() call on the same plan",
c_arb.fresh_tariff_plan(NOW).arbitrage(
peak_guard_active=False, min_spread_eur_kwh=0.05,
now=NOW, peak_guard_active=False, min_spread_eur_kwh=0.05,
uncertainty=frozenset({__import__("app.price", fromlist=["x"])
.Uncertainty.NO_WEATHER}))
.match(allowed=lambda v: (v.cheapest.start, v.dearest.start),
@@ -717,7 +717,7 @@ def _dn(arb):
_c_off, _p_off = _dn(False)
_c_on, _p_on = _dn(True)
_v_on, _why_on = homeview.trade_verdict(_c_on, _c_on.fresh_tariff_plan(NOW))
_v_on, _why_on = homeview.trade_verdict(_c_on, _c_on.fresh_tariff_plan(NOW), NOW)
check("UX-45: day/night tariff with the stored plan and price trading off: "
"24 bars, no cheap or dear mark",
len(_p_off) == 24 and {b["kind"] for b in _p_off} == {"mid"})
+42 -10
View File
@@ -66,22 +66,54 @@ 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 = {
# Printed on release/1.0 7a00f12 (2026-09-29), goodwe-addon-checks:1, 1080
# cycles each; the trade CHARGEs and DISCHARGEs.
#
# PRICE-02 (2026-10-22) changed two things in the trade sentence and nothing
# else: the plan now drops price periods that have ended, so a sale's "N above
# the cheapest" is another number once the first cheap period is over, and the
# sentences say "price period" for "hour". So there are three pins:
# AT_7A00F12_OLD the digests printed on 7a00f12, sentence included. They no
# longer match, and the check below says so: that is what
# makes the mask necessary.
# AT_7A00F12 the same rows through `ident.trade_digest()`, which hides
# exactly those two things (and only for this comparison with
# an older tree). It equals what 7a00f12 gives through it.
# AT_NOW this tree's own digests, sentence included, with nothing
# hidden: any later change to a command, a sentence or a
# figure moves it. The mask cannot hide a future drift.
AT_7A00F12_OLD = {
"n1": "71245b227ca40bada4f71f57ea6154475f4b0bd60239c9ad4f80bffb0094514c",
"n2": "56cf28010b06765c29e5479abb815d7503d570b39240d82dfc818efa90c93722",
"n1_off": "71245b227ca40bada4f71f57ea6154475f4b0bd60239c9ad4f80bffb0094514c",
"n2_off": "56cf28010b06765c29e5479abb815d7503d570b39240d82dfc818efa90c93722",
}
AT_7A00F12 = {
"n1": "33886ef03f364852883f0370230e6e04823fbb46a4704f88ca75bf9082d98c01",
"n2": "666a31f41491c22750720524853d95944d5fba395d1d14040c460d71414b275f",
"n1_off": "33886ef03f364852883f0370230e6e04823fbb46a4704f88ca75bf9082d98c01",
"n2_off": "666a31f41491c22750720524853d95944d5fba395d1d14040c460d71414b275f",
}
AT_NOW = {
"n1": "939b3d47bbd64fa50e3d70538f092fb98e5c4bd5ead85a7fc696a1dbb514039e",
"n2": "c51008213dd739e1e5142c236d9a950aa585992393e4af120086f1ba52ff3e0d",
"n1_off": "939b3d47bbd64fa50e3d70538f092fb98e5c4bd5ead85a7fc696a1dbb514039e",
"n2_off": "c51008213dd739e1e5142c236d9a950aa585992393e4af120086f1ba52ff3e0d",
}
_cmd_digest = ident.trade_digest
_rows = {k: ident.run(**ident.SCENARIOS[k])[0] for k in AT_7A00F12}
_got = {k: ident.digest(v) for k, v in _rows.items()}
_got = {k: _cmd_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]}")
_full = {k: ident.digest(v) for k, v in _rows.items()}
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)
"integrator, allocation and trade action equals 7a00f12's, at one "
"inverter and at two (the sentence masked as above)",
_got == AT_7A00F12)
check("...the mask is needed and nothing more: unmasked, n1 and n2 differ "
"from 7a00f12's printed digests, and unmasked they equal this tree's "
"own pins, so a later change to any sentence or figure fails here",
all(_full[k] != AT_7A00F12_OLD[k] for k in AT_7A00F12_OLD)
and _full == AT_NOW)
_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)
@@ -96,7 +128,7 @@ def _moved(scn, patch):
real(c, t)
ident._feed = feed
try:
return ident.digest(ident.run(**ident.SCENARIOS[scn])[0]) \
return _cmd_digest(ident.run(**ident.SCENARIOS[scn])[0]) \
!= AT_7A00F12[scn]
finally:
ident._feed = real
@@ -112,7 +144,7 @@ def _moved_headroom(scn):
real = peakguard.charge_headroom_w
peakguard.charge_headroom_w = lambda g: None
try:
return ident.digest(ident.run(**ident.SCENARIOS[scn])[0]) \
return _cmd_digest(ident.run(**ident.SCENARIOS[scn])[0]) \
!= AT_7A00F12[scn]
finally:
peakguard.charge_headroom_w = real
+8 -3
View File
@@ -99,6 +99,11 @@ from app.static_tariff import (
TariffPlan, parse_taxes_profile, static_status)
from app.validate import validate_startup_config
# PRICE-02: `TariffPlan.arbitrage(now=...)` is required. This instant is before every
# point these fixtures build, so no period has ended and the whole series counts.
from datetime import datetime as _DT, timezone as _TZ # noqa: E402
LONG_AGO = _DT(2000, 1, 1, tzinfo=_TZ.utc)
fails = []
@@ -4518,8 +4523,8 @@ _T11_FLAT = TariffPlan(
[PricePoint(_T11_NOW + timedelta(minutes=15 * i), 0.10, False)
for i in range(96)],
PriceProduct.DAY_AHEAD, "generic_dayahead", _T11_NOW), None, "provider")
_r83 = derived(lambda: _T11_ONEFIRM.arbitrage(**_T11_A9).reason)
_rflat = derived(lambda: _T11_FLAT.arbitrage(**_T11_A9).reason)
_r83 = derived(lambda: _T11_ONEFIRM.arbitrage(now=LONG_AGO, **_T11_A9).reason)
_rflat = derived(lambda: _T11_FLAT.arbitrage(now=LONG_AGO, **_T11_A9).reason)
print(f" §8.3 refusal : {_r83[:118]}")
print(f" flat refusal : {_rflat[:118]}")
# 96 quarter-hourly points span 24 hours; ONE firm point spans one of them.
@@ -4828,7 +4833,7 @@ def _t17_verdict(points):
read = PriceRead.series(points, PriceProduct.DAY_AHEAD, "t17",
datetime.now(UTC))
plan = TariffPlan(read, None, "provider")
return plan.arbitrage(peak_guard_active=False,
return plan.arbitrage(now=LONG_AGO, peak_guard_active=False,
min_spread_eur_kwh=PRICE_OPTION_DEFAULTS[
"cycle_cost_threshold_eur"],
uncertainty=frozenset())
+538
View File
@@ -0,0 +1,538 @@
"""Runnable check for PRICE-02. `python3 test_price02.py`
The trading verdict is taken over the price periods that have not ended.
`TariffPlan.arbitrage(now=...)` drops the points with `end <= now` (the span
`price._cover_table()` gives each one) and every caller passes its `now`:
`arbitrage.decide()` (modes off), the `allowed()` gate in
`Controller._mode_roles()` (modes on) and the Home page's `trade_verdict()`.
1. A 15-minute day, probed at :05, :20, :35 and :50 of the cheap hour: the
buy leg and the sell leg move to the current period and never name one that
has ended, and `decide()` trades in it. Cheapest in the past, the current
and a future period are all covered.
2. A plan built at 03:00 and used at 03:20 (the 30-minute refresh lag), and the
shipped refresh (`fill_from_yesterday`, which keeps the whole current hour).
3. An hourly feed and the static day/night tariff: an hourly point that is
still current stays in. A plan built off the hour (the refresh runs every
30 min from add-on start, so :20 and :50 are normal) no longer aims at the
hour that has just ended, so a trade runs through (owner, 2026-10-22), and
a period that ends exactly at `now` is over.
4. Modes on: a spread that exists only because of an ended period no longer
opens the gate (`Controller._mode_roles`), nor the Home page verdict.
5. The reason text says "price period"; `now` is required (no default).
6. Ties. The verdict takes the first extreme still ahead, where it used to
take the first of the whole plan. On hour-constant 15-minute data (four
equal quarters) that is four acting periods in the cheap hour, not one;
the number of times the action changes does not grow. The same on a real
15-minute day, and hourly alternating prices change at most once an hour.
Mutations: MUTATION.md says how to run them (purged copies); the ones of
PRICE-02 are listed in the pull request.
"""
import json
import logging
import os
import sys
from datetime import datetime, timedelta, timezone
HERE = os.path.dirname(os.path.abspath(__file__))
sys.path.insert(0, HERE)
sys.path.insert(0, os.path.join(HERE, "tools"))
logging.disable(logging.WARNING)
from app import arbitrage, homeview # noqa: E402
from app.arbitrage import CHARGE, DISCHARGE, decide # noqa: E402
from app.price import (PricePoint, PriceProduct, PriceRead, # noqa: E402
Uncertainty, brussels)
from app.price_forecast import fill_from_yesterday # noqa: E402
from app.static_tariff import TariffPlan, static_plan # noqa: E402
fails = []
def check(name, cond):
print(f" {'ok ' if cond else 'FAIL'} {name}")
if not cond:
fails.append(name)
UTC = timezone.utc
LONG_AGO = datetime(2000, 1, 1, tzinfo=UTC) # before every point: nothing has ended, the old reach
NO_WEATHER = frozenset({Uncertainty.NO_WEATHER})
D0 = datetime(2026, 10, 20, 0, 0, tzinfo=UTC) # 02:00 Brussels
H3 = D0 + timedelta(hours=1) # 03:00 Brussels
ARGS = dict(peak_guard_active=False, min_spread_eur_kwh=0.10,
uncertainty=NO_WEATHER)
def qplan(prices, start=D0, minutes=15, product=PriceProduct.DAY_AHEAD):
pts = [PricePoint(brussels(start + timedelta(minutes=minutes * i)), p,
False) for i, p in enumerate(prices)]
return TariffPlan(PriceRead.series(pts, product, "generic_dayahead",
start), None, "provider")
ARGS_NO_NOW = dict(ARGS)
ARGS5 = dict(ARGS, min_spread_eur_kwh=0.05, uncertainty=frozenset())
def dec(plan, now, **kw):
base = dict(peak_guard_active=False, charge_headroom_w=5000.0, soc=50.0,
min_spread_eur_kwh=0.10, uncertainty=frozenset(),
charge_w=2500.0, discharge_w=2500.0, soc_floor_pct=10.0,
soc_ceiling_pct=90.0)
base.update(kw)
return decide(plan, now, **base)
def hhmm(t):
return f"{brussels(t):%H:%M}"
def ahead_extreme(prices, now, pick, start=D0, minutes=15):
"""The expected leg, by plain arithmetic: the extreme price over the
periods whose end is after `now`, as the start instant (first on a tie)."""
best = None
for i, p in enumerate(prices):
s = start + timedelta(minutes=minutes * i)
if s + timedelta(minutes=minutes) <= now:
continue
if best is None or pick(p, best[1]):
best = (s, p)
return best[0]
# --------------------------------------------------------------------------- #
print("1. a 15-minute day: the verdict is taken from the current period on")
BUY = [0.30] * 192 # 48 h of quarters from 02:00
for q, p in ((4, 0.05), (5, 0.06), (6, 0.07), (7, 0.08), (68, 0.50)):
BUY[q] = p # 03:00 .05 03:15 .06 03:30 .07 03:45 .08 19:00 .50
BUY_PLAN = qplan(BUY)
check("the fixture: the day's cheapest period is 03:00, then three dearer "
"ones, and the dearest is 19:00 - so every probe has a past, a current "
"and a future period to choose between",
BUY.index(min(BUY)) == 4 and BUY.index(max(BUY)) == 68
and BUY[4] < BUY[5] < BUY[6] < BUY[7] < min(BUY[8:68]))
want_q = {5: 4, 20: 5, 35: 6, 50: 7}
for mins, q in want_q.items():
now = H3 + timedelta(minutes=mins)
v = BUY_PLAN.arbitrage(now=now, **ARGS)
d = dec(BUY_PLAN, now)
got = v.cheapest.start
exp = ahead_extreme(BUY, now, lambda a, b: a < b)
check(f"03:{mins:02d}: the buy leg is {hhmm(exp)} (the first period not "
f"over), the verdict says {hhmm(got)}, and decide() is {d.action}",
got == exp == D0 + timedelta(minutes=15 * q)
and d.action == CHARGE
and abs(d.price_now_eur_kwh - BUY[q]) < 1e-9)
check("before the change the same plan at 03:20 aimed at 03:00 (the "
"verdict without `now`) and decide() did nothing - the defect, shown",
BUY_PLAN.arbitrage(now=LONG_AGO, **ARGS).cheapest.start == H3
and arbitrage._act(BUY_PLAN, H3 + timedelta(minutes=20),
BUY_PLAN.arbitrage(now=LONG_AGO, **ARGS), 50.0, 2500.0, 2500.0,
10.0, 90.0, 5000.0).action is None)
# A cheapest period still AHEAD stays the target from every probe.
FUT = [0.30] * 192
FUT[8], FUT[68] = 0.04, 0.50 # cheapest 04:00, dearest 19:00
FUT_PLAN = qplan(FUT)
check("the cheapest in the FUTURE (04:00) is the buy leg at :05, :20, :35 and "
":50 of the hour before it - nothing ahead is dropped",
all(FUT_PLAN.arbitrage(now=H3 + timedelta(minutes=m), **ARGS)
.cheapest.start == H3 + timedelta(hours=1) for m in (5, 20, 35, 50)))
check("...and at 04:20 (that period is now the current one) it is still the "
"current period at :00-:15 only: 04:20 aims elsewhere",
FUT_PLAN.arbitrage(now=H3 + timedelta(hours=1, minutes=20), **ARGS)
.cheapest.start != H3 + timedelta(hours=1))
print("\n1b. the sell leg")
SELL = [0.30] * 192
for q, p in ((4, 0.70), (5, 0.65), (6, 0.60), (7, 0.55), (68, 0.10)):
SELL[q] = p # dearest 03:00 .70 .. 03:45 .55 ; cheapest 19:00
SELL_PLAN = qplan(SELL)
for mins, q in want_q.items():
now = H3 + timedelta(minutes=mins)
v = SELL_PLAN.arbitrage(now=now, **ARGS)
d = dec(SELL_PLAN, now)
exp = ahead_extreme(SELL, now, lambda a, b: a > b)
check(f"03:{mins:02d}: the sell leg is {hhmm(exp)}, the verdict says "
f"{hhmm(v.dearest.start)}, and decide() is {d.action}",
v.dearest.start == exp == D0 + timedelta(minutes=15 * q)
and d.action == DISCHARGE
and abs(d.price_now_eur_kwh - SELL[q]) < 1e-9)
# --------------------------------------------------------------------------- #
print("\n2. the plan is older than the verdict")
# Built at 03:00 and used at 03:20, 03:35 (the refresh runs every 30 min).
built = H3
feed = qplan(BUY).read
plan_0300 = TariffPlan(fill_from_yesterday(
feed, yesterday=PriceRead.no_data("none"), since=built,
until=built + timedelta(hours=48), stamp=built), None, "provider")
check("the plan built at 03:00 starts at 03:00 (the refresh floors to the hour)",
min(p.start for p in plan_0300.read.points) == H3)
for mins, q in ((20, 5), (35, 6)):
now = H3 + timedelta(minutes=mins)
d = dec(plan_0300, now)
check(f"used at 03:{mins:02d} the verdict is "
f"{hhmm(D0 + timedelta(minutes=15 * q))} and decide() charges there",
plan_0300.arbitrage(now=now, **ARGS).cheapest.start
== D0 + timedelta(minutes=15 * q) and d.action == CHARGE)
# The shipped refresh at 03:20 keeps the whole hour: the verdict still skips it.
now = H3 + timedelta(minutes=20)
plan_refresh = TariffPlan(fill_from_yesterday(
feed, yesterday=PriceRead.no_data("none"), since=now,
until=now + timedelta(hours=48), stamp=now), None, "provider")
check("the shipped refresh at 03:20 still holds the 03:00 period, as before, "
"and the verdict does not aim at it",
min(p.start for p in plan_refresh.read.points) == H3
and plan_refresh.arbitrage(now=now, **ARGS).cheapest.start
== H3 + timedelta(minutes=15))
# --------------------------------------------------------------------------- #
print("\n3. hourly feeds and the static tariff: a current hour stays in")
HR = [0.30] * 48 # 48 hourly points from 02:00
HR[1], HR[17] = 0.05, 0.50 # cheapest 03:00, dearest 19:00
HR_PLAN = qplan(HR, minutes=60)
same = True
for m in (0, 5, 20, 35, 59):
now = H3 + timedelta(minutes=m)
a = HR_PLAN.arbitrage(now=now, **ARGS)
b = HR_PLAN.arbitrage(now=LONG_AGO, **ARGS)
same = same and a.cheapest is b.cheapest and a.dearest is b.dearest \
and a.spread_eur_kwh == b.spread_eur_kwh
check("an hourly point that is still current stays in: at 03:00, :05, :20, "
":35 and :59 the verdict is the one without `now`", same)
check("...and decide() still charges in the current hour at 03:40",
dec(HR_PLAN, H3 + timedelta(minutes=40)).action == CHARGE)
check("...and once the hour has ended its point is gone: at 04:05 the buy leg "
"is no longer 03:00",
HR_PLAN.arbitrage(now=H3 + timedelta(hours=1, minutes=5), **ARGS)
.cheapest.start != H3)
TWO = {"periods": [
{"name": "day", "from": "07:00", "to": "22:00", "days": "mon-sun",
"eur_per_kwh": 0.40},
{"name": "night", "from": "22:00", "to": "07:00", "days": "mon-sun",
"eur_per_kwh": 0.18}]}
OPTS = {"price_provider": "off", "static_profile": json.dumps(TWO)}
ok, n_checked = True, 0
for h in range(0, 24):
for m in (0, 20, 50):
now = D0 + timedelta(hours=h - 2, minutes=m) # from 22:00 UTC the day before
plan = static_plan(OPTS, now, now + timedelta(hours=48), now=now)
a = plan.arbitrage(now=now, **ARGS)
b = plan.arbitrage(now=LONG_AGO, **ARGS)
n_checked += 1
ok = ok and a.kind == b.kind == "allowed" and a.cheapest is b.cheapest \
and a.dearest is b.dearest and a.spread_eur_kwh == b.spread_eur_kwh
check(f"the static day/night plan, built at the instant it is used (so it "
f"holds no ended hour), gives the same verdict with and without `now` "
f"at {n_checked} instants", ok and n_checked == 72)
# 3b. The refresh is NOT on the hour. main.py rebuilds the plan every 30 min
# counted from add-on start (TARIFF_REFRESH_S = 1800), so a plan built at :20
# or :50 starts at the floored hour and, until the next refresh, still holds
# the hour that has just ended. Owner decision 2026-10-22: the verdict skips
# that hour too, so a trade runs through instead of pausing until the refresh.
print("\n3b. hourly and static plans built off the hour: the ended hour is not the target")
HR2 = [0.30] * 48 # hourly from 02:00 Brussels
HR2[1], HR2[2], HR2[17] = 0.05, 0.05, 0.50 # cheap 03:00 and 04:00 (a tie), dear 19:00
built = H3 + timedelta(minutes=50) # the refresh at :50 floors to 03:00
plan_h = TariffPlan(fill_from_yesterday(
qplan(HR2, minutes=60).read, yesterday=PriceRead.no_data("none"),
since=built, until=built + timedelta(hours=48), stamp=built),
None, "provider")
now = H3 + timedelta(hours=1, minutes=5) # 04:05, before the next refresh
check("the hourly plan built at 03:50 starts at 03:00 and is used at 04:05, so "
"it still holds the 03:00 hour, which is over",
min(p.start for p in plan_h.read.points) == H3)
old_v = plan_h.arbitrage(now=LONG_AGO, **ARGS)
new_v = plan_h.arbitrage(now=now, **ARGS)
check("without `now` (the old verdict) the buy leg is the ended 03:00 hour "
"(first of the tie) and decide() does nothing at 04:05 - the pause",
old_v.cheapest.start == H3
and arbitrage._act(plan_h, now, old_v, 50.0, 2500.0, 2500.0, 10.0,
90.0, 5000.0).action is None)
check("with `now` the buy leg is 04:00, the hour in progress, and decide() "
"charges at 04:05: the trade runs through",
new_v.cheapest.start == H3 + timedelta(hours=1)
and dec(plan_h, now).action == CHARGE)
# The day/night tariff, refresh at :20 of the hour: all 15 day hours tie.
b_local = D0 + timedelta(hours=10, minutes=20) # 12:20 Brussels
plan_s = static_plan(OPTS, b_local, b_local + timedelta(hours=48), now=b_local)
now = D0 + timedelta(hours=11, minutes=5) # 13:05 Brussels
old_v = plan_s.arbitrage(now=LONG_AGO, **ARGS)
new_v = plan_s.arbitrage(now=now, **ARGS)
check("the day/night plan built at 12:20 starts at 12:00; used at 13:05 its "
"sell leg without `now` is the ended 12:00 hour (first of the tied day "
"hours) and decide() does nothing - the old hourly pause",
min(p.start for p in plan_s.read.points) == D0 + timedelta(hours=10)
and old_v.dearest.start == D0 + timedelta(hours=10)
and arbitrage._act(plan_s, now, old_v, 50.0, 2500.0, 2500.0, 10.0,
90.0, 5000.0).action is None)
check("with `now` the sell leg is 13:00, the hour in progress, and decide() "
"discharges at 13:05: the trade runs through",
new_v.dearest.start == D0 + timedelta(hours=11)
and dec(plan_s, now).action == DISCHARGE)
ok = True
for m in (0, 5, 20, 35, 50, 59): # every minute class of the hour
t = D0 + timedelta(hours=12, minutes=m) # 14:mm Brussels
v = plan_s.arbitrage(now=t, **ARGS)
ok = ok and v.dearest.start == D0 + timedelta(hours=12) \
and dec(plan_s, t).action == DISCHARGE
check("...and through the whole of 14:00-14:59 the target is the 14:00 hour "
"and the trade stays on (no pause at the top of the hour)", ok)
# The exact boundary: a period ends AT its `end`, so at that instant it is over.
# (`end >= now` would keep it for one more instant.)
print("\n3c. the exact end of a period")
at = H3 + timedelta(minutes=15) # 03:15:00 exactly
v = BUY_PLAN.arbitrage(now=at, **ARGS)
check("at 03:15:00 exactly the 03:00 quarter is over: the buy leg is 03:15 "
"and decide() charges", v.cheapest.start == at and dec(BUY_PLAN, at)
.action == CHARGE)
check("one microsecond earlier the 03:00 quarter is still in view",
BUY_PLAN.arbitrage(now=at - timedelta(microseconds=1), **ARGS)
.cheapest.start == H3)
at_h = H3 + timedelta(hours=1) # 04:00:00 exactly
check("at 04:00:00 exactly the 03:00 hour is over (hourly feed), one "
"microsecond earlier it is not",
plan_h.arbitrage(now=at_h, **ARGS).cheapest.start == at_h
and plan_h.arbitrage(now=at_h - timedelta(microseconds=1), **ARGS)
.cheapest.start == H3)
two = qplan([0.10, 0.20]) # two firm quarters from 02:00
check("the Home page's refusal key at the exact end of the first of two "
"quarters: one is left, so 'not_firm' (with `end >= now` it would say "
"'not_worth'); a microsecond earlier both count",
homeview.refusal_key(two, False, D0 + timedelta(minutes=15)) == "not_firm"
and homeview.refusal_key(two, False, D0 + timedelta(minutes=15)
- timedelta(microseconds=1)) == "not_worth")
# --------------------------------------------------------------------------- #
print("\n4. modes on: an ended period no longer opens the gate")
GATE = [0.30] * 192
GATE[4] = 0.10 # the ONLY spread is the 03:00 period
GATE_PLAN = qplan(GATE)
now = H3 + timedelta(minutes=20)
check("the fixture: with the ended 03:00 period in view the verdict is "
"allowed; from 03:20 on the rest of the series is flat",
GATE_PLAN.arbitrage(now=LONG_AGO, **ARGS).kind == "allowed"
and GATE_PLAN.arbitrage(now=now, **ARGS).kind == "refused"
and "flat" in GATE_PLAN.arbitrage(now=now, **ARGS).reason)
import opt11p3_replay as rp # noqa: E402
WPTS = [(D0 + timedelta(minutes=15 * i), 900, 0.10) for i in range(192)]
def controller_at(at, plan):
"""The real Controller after a few cycles of the modes-on replay at `at`
(two batteries, one on Cheap hours), holding `plan` as its fresh plan."""
_st, c = rp.run("p3_ch_sell", [(D0.timestamp() - 10, 800.0)], WPTS, at,
20.0, soc0={"a": 60.0, "b": 30.0},
extra={"cycle_cost_threshold_eur": 0.05})
c.tariff_plan, c.tariff_at = plan, at
return c
from app import modetrade # noqa: E402
GATE_SEEN = {}
_real_decide = modetrade.decide
def _spy(runs, now_, packs, **kw):
GATE_SEEN.update(charge_ok=kw["charge_ok"], sell_ok=kw["sell_ok"],
trade_why=kw["trade_why"])
return _real_decide(runs, now_, packs, **kw)
modetrade.decide = _spy
c = controller_at(now, GATE_PLAN)
GATE_SEEN.clear()
c._mode_roles(now, None)
late = dict(GATE_SEEN)
c2 = controller_at(H3 + timedelta(minutes=5), GATE_PLAN)
GATE_SEEN.clear()
c2._mode_roles(H3 + timedelta(minutes=5), None)
early = dict(GATE_SEEN)
modetrade.decide = _real_decide
check(f"Controller._mode_roles() (modes on, the `allowed()` gate) at 03:05, "
f"03:00 still current: charge and sell are allowed ({early})",
early.get("charge_ok") is True and early.get("sell_ok") is True)
check(f"...and at 03:20, with only flat periods ahead: neither is allowed, "
f"and the reason is the spread ({late})",
late.get("charge_ok") is False and late.get("sell_ok") is False
and late.get("trade_why") == "not_worth")
c.o["arbitrage_enabled"] = True
v, why = homeview.trade_verdict(c, GATE_PLAN, now)
check(f"the Home page verdict takes the same `now`: no verdict at 03:20 "
f"({why!r}), and a verdict at 03:05",
v is None and why == "not_worth"
and homeview.trade_verdict(c, GATE_PLAN, H3 + timedelta(minutes=5))[0]
is not None)
check("...and the refusal key counts only periods not over: two firm periods "
"left, so 'not_worth', and with one left 'not_firm'",
homeview.refusal_key(GATE_PLAN, False, now) == "not_worth"
and homeview.refusal_key(qplan([0.1, 0.2]), False,
D0 + timedelta(minutes=20)) == "not_firm")
# --------------------------------------------------------------------------- #
print("\n5. the text")
check("the reason for an off-peak period says 'price period', not 'hour'",
arbitrage.IDLE_OFF_PEAK_HOUR
== "now is not the cheapest or the dearest price period")
_b = dec(BUY_PLAN, H3 + timedelta(minutes=20))
_s = dec(SELL_PLAN, H3 + timedelta(minutes=20))
check("...and the charge and the discharge sentences say 'price period' too "
"(the unit is 15 minutes), with no 'firm hour' left",
"cheapest firm price period" in _b.reason
and "dearest firm price period" in _s.reason
and "firm hour" not in _b.reason + _s.reason)
_hi = dec(BUY_PLAN, H3 + timedelta(minutes=20), soc=95.0)
_lo = dec(SELL_PLAN, H3 + timedelta(minutes=20), soc=10.0)
check("...and so do the full-pack and empty-pack sentences ('buy period', "
"'sell period')",
"the buy period is here" in _hi.reason
and "the sell period is here" in _lo.reason)
import inspect # noqa: E402
_par = inspect.signature(TariffPlan.arbitrage).parameters
check("`now` is required: keyword-only with no default, like the three "
"arguments beside it, so a caller that forgets it fails loudly instead "
"of getting the elapsed periods back",
_par["now"].default is inspect.Parameter.empty
and _par["now"].kind is inspect.Parameter.KEYWORD_ONLY
and all(p.default is inspect.Parameter.empty
for n, p in _par.items() if n != "self"))
try:
BUY_PLAN.arbitrage(**ARGS_NO_NOW)
_forgot = False
except TypeError:
_forgot = True
check("...and calling without it raises TypeError", _forgot)
try:
BUY_PLAN.arbitrage(now=datetime(2026, 10, 20, 3, 20), **ARGS)
_naive = False
except TypeError as e:
_naive = "tz-aware" in str(e)
check("a naive `now` is refused: it is ambiguous for one hour every October",
_naive)
# --------------------------------------------------------------------------- #
print("\n6. ties: the first extreme still ahead")
def acts(plan, instants, ago=False, args=ARGS):
out = []
for t in instants:
v = plan.arbitrage(now=LONG_AGO if ago else t, **args)
d = arbitrage._act(plan, t, v, 50.0, 2500.0, 2500.0, 10.0, 90.0,
5000.0) if v.kind == "allowed" else None
out.append(None if d is None else d.action)
return out
# Hour-constant 15-minute data: the cheap hour is four equal quarters.
HC = [round(0.40 - 0.0001 * i, 4) for i in range(192)] # falling, no ties
for q in range(4, 8):
HC[q] = 0.05 # 03:00-04:00 four equal quarters
HC[68:72] = [0.50] * 4 # 19:00-20:00 four equal quarters
HC_PLAN = qplan(HC)
probe = [H3 + timedelta(minutes=m) for m in (5, 20, 35, 50, 65)]
old = acts(HC_PLAN, probe, ago=True)
new = acts(HC_PLAN, probe)
check(f"before: four equal cheap quarters, the verdict is the first of them, "
f"so the trade acts in one quarter ({old})",
old == [CHARGE, None, None, None, None])
check(f"after: the first one still ahead, so it acts in all four and stops "
f"with the hour ({new}) - four acting periods for the same hour, one "
f"block either way",
new == [CHARGE, CHARGE, CHARGE, CHARGE, None])
changes = lambda a: sum(1 for x, y in zip(a, a[1:]) if x != y) # noqa: E731
check("...the action changes the same number of times (one start, one stop)",
changes(old) == changes(new) == 1)
# A real 15-minute day, as production runs it: refresh every 30 min from the
# floored hour, the verdict at 7 min into every period.
REAL = os.path.join(HERE, "scenarios",
"energy-charts-be-price-2026-09-09_2026-09-10-twoday.json")
_d = json.load(open(REAL))
_pts = [PricePoint(brussels(datetime.fromtimestamp(t, UTC)), p / 1000.0, False)
for t, p in zip(_d["unix_seconds"], _d["price"])]
_feed = PriceRead.series(_pts, PriceProduct.DAY_AHEAD, "generic_dayahead",
_pts[0].start)
_day0 = _pts[0].start.astimezone(UTC)
def real_day(ago):
out = []
for q in range(96):
t = _day0 + timedelta(minutes=15 * q + 7)
ref = t.replace(minute=0 if t.minute < 30 else 30, second=0,
microsecond=0)
plan = TariffPlan(fill_from_yesterday(
_feed, yesterday=PriceRead.no_data("none"), since=ref,
until=ref + timedelta(hours=36), stamp=ref), None, "provider")
v = plan.arbitrage(now=LONG_AGO if ago else t, **ARGS5)
out.append(arbitrage._act(plan, t, v, 50.0, 2500.0, 2500.0, 10.0,
90.0, 5000.0).action
if v.kind == "allowed" else None)
return out
r_old, r_new = real_day(True), real_day(False)
n_old = sum(1 for a in r_old if a)
n_new = sum(1 for a in r_new if a)
print(f" real 15-minute day 2026-09-09: acting periods {n_old} -> {n_new}, "
f"action changes {changes(r_old)} -> {changes(r_new)}")
check("a real 15-minute day (energy-charts, 2026-09-09) does not switch more "
"often: the action changes no more often than before",
changes(r_new) <= changes(r_old))
check("...and every period the old verdict acted in is still acted in, with "
"the quarters it had lost to the elapsed-period tie added",
all(n is not None for o, n in zip(r_old, r_new) if o) and n_new >= n_old)
# Alternating hourly prices (the flapping day): ties everywhere, so the verdict
# moves on every hour. At most once an hour, never once a tick.
FL = [0.20 + (0.101 if i % 2 else 0.0) for i in range(24)] * 2
FL_PLAN = qplan(FL, minutes=60)
ticks = [acts(FL_PLAN, [D0 + timedelta(seconds=30 * k)], args=ARGS5)[0]
for k in range(24 * 120)]
check(f"alternating hourly prices: {changes(ticks)} action changes over "
f"2880 ticks - at most one an hour, not one a tick",
changes(ticks) <= 24 and CHARGE in ticks and DISCHARGE in ticks)
# The Home page path: `homeview.build_plan()` hands its own `now` to
# `trade_verdict`. With a `now` floored to the hour it would still aim at the
# 03:00 period at 03:20 and draw a trade that the control cycle does not take.
print("\n4b. the Home page plan (build_plan) uses the instant it is rendered at")
c.o["arbitrage_enabled"] = True
c.modes_on = False
def home_at(t):
c.tariff_plan, c.tariff_at = GATE_PLAN, t
return homeview.build_plan(c, t, False, True)
h_early = home_at(H3 + timedelta(minutes=5))
h_late = home_at(H3 + timedelta(minutes=20))
trade_kinds = ("grid", "give")
check("build_plan at 03:05 (03:00 still current): the spread is real, so the "
"page draws a trade (verdict_key 'trade')",
h_early["verdict_key"] == "trade"
and any(r["kind"] in trade_kinds for r in h_early["runs"]))
check("build_plan at 03:20 (only flat periods ahead): no trade on the page, "
"the same refusal as trade_verdict gives (verdict_key 'house')",
h_late["verdict_key"] == "house"
and not any(r["kind"] in trade_kinds for r in h_late["runs"])
and h_late["lines"] != h_early["lines"])
print()
if fails:
print(f"{len(fails)} FAILED: " + "; ".join(fails))
sys.exit(1)
print("all checks passed")
+14 -9
View File
@@ -87,6 +87,11 @@ from app.static_tariff import Arbitrage, TariffPlan
UTC = timezone.utc
HERE = os.path.dirname(os.path.abspath(__file__))
# PRICE-02: `TariffPlan.arbitrage(now=...)` is required. This instant is before every
# point these fixtures build, so no period has ended and the whole series counts.
from datetime import datetime as _DT, timezone as _TZ # noqa: E402
LONG_AGO = _DT(2000, 1, 1, tzinfo=_TZ.utc)
fails = []
@@ -905,7 +910,7 @@ def min_accepted_spread(product, sources, base, provider="generic_dayahead"):
lo, hi = 0.0, 4.0
for _ in range(60):
mid = (lo + hi) / 2
v = plan_with_spread(mid, product, provider).arbitrage(
v = plan_with_spread(mid, product, provider).arbitrage(now=LONG_AGO,
peak_guard_active=False, min_spread_eur_kwh=base,
uncertainty=sources)
if v.kind == "allowed":
@@ -1013,7 +1018,7 @@ check(f"the fixture is non-degenerate: the whole-series spread "
f"{_all_spread:.6f} and the firm-only spread {_firm_spread:.6f} are "
f"different numbers, so a verdict can be traced to one of them",
not near(_all_spread, _firm_spread, 1e-6))
_v = derived(lambda: _mixed.arbitrage(peak_guard_active=False,
_v = derived(lambda: _mixed.arbitrage(now=LONG_AGO, peak_guard_active=False,
min_spread_eur_kwh=0.0,
uncertainty=frozenset()))
check("a PART-published series still allows arbitrage - §8.3 forbids acting on "
@@ -1032,7 +1037,7 @@ check("...and it reports the threshold it cleared and why, rather than leaving "
_all_prov = TariffPlan(PriceRead.series(_cheap_stale, PriceProduct.DAY_AHEAD,
"generic_dayahead", D0), None,
"fallback")
_vp = derived(lambda: _all_prov.arbitrage(peak_guard_active=False,
_vp = derived(lambda: _all_prov.arbitrage(now=LONG_AGO, peak_guard_active=False,
min_spread_eur_kwh=0.0,
uncertainty=frozenset()))
check("a WHOLLY provisional series - the §11.4 fallback, or a day before any "
@@ -1046,23 +1051,23 @@ _one_firm = TariffPlan(
"generic_dayahead", D0), None, "provider")
check("one firm hour is not enough either: a buy hour and a sell hour must "
"BOTH be published before a cycle can be committed",
_one_firm.arbitrage(peak_guard_active=False, min_spread_eur_kwh=0.0,
_one_firm.arbitrage(now=LONG_AGO, peak_guard_active=False, min_spread_eur_kwh=0.0,
uncertainty=frozenset()).kind == "refused")
check("§10.1 still holds above all of it: the peak guard refuses the same "
"series that is otherwise allowed",
_mixed.arbitrage(peak_guard_active=True, min_spread_eur_kwh=0.0,
_mixed.arbitrage(now=LONG_AGO, peak_guard_active=True, min_spread_eur_kwh=0.0,
uncertainty=frozenset()).kind == "refused")
check("the three questions cannot be skipped: all of peak_guard_active, "
"min_spread_eur_kwh and uncertainty are required and keyword-only",
raises(lambda: _mixed.arbitrage(peak_guard_active=False), TypeError)
and raises(lambda: _mixed.arbitrage(peak_guard_active=False,
raises(lambda: _mixed.arbitrage(now=LONG_AGO, peak_guard_active=False), TypeError)
and raises(lambda: _mixed.arbitrage(now=LONG_AGO, peak_guard_active=False,
min_spread_eur_kwh=0.0), TypeError)
and raises(lambda: _mixed.arbitrage(False, 0.0, frozenset()), TypeError))
check("...nor answered with a truthy 'unknown' or a negative threshold",
raises(lambda: _mixed.arbitrage(peak_guard_active=None,
raises(lambda: _mixed.arbitrage(now=LONG_AGO, peak_guard_active=None,
min_spread_eur_kwh=0.0,
uncertainty=frozenset()), TypeError)
and raises(lambda: _mixed.arbitrage(peak_guard_active=False,
and raises(lambda: _mixed.arbitrage(now=LONG_AGO, peak_guard_active=False,
min_spread_eur_kwh=-0.1,
uncertainty=frozenset()), TypeError))
+10 -5
View File
@@ -118,6 +118,11 @@ from app.static_tariff import resolve_tariff
UTC = timezone.utc
HERE = os.path.dirname(os.path.abspath(__file__))
# PRICE-02: `TariffPlan.arbitrage(now=...)` is required. This instant is before every
# point these fixtures build, so no period has ended and the whole series counts.
from datetime import datetime as _DT, timezone as _TZ # noqa: E402
LONG_AGO = _DT(2000, 1, 1, tzinfo=_TZ.utc)
fails = []
_tmpdirs = []
@@ -699,8 +704,8 @@ check(f"the short answer stays a SERIES of what exists - {_pre.read.kind}, "
# --- the trade set, which is where the whole thing lands ------------------- #
_A9 = dict(peak_guard_active=False, min_spread_eur_kwh=0.0,
uncertainty=frozenset())
_v2 = _pub.arbitrage(**_A9)
_v1 = _pre.arbitrage(**_A9)
_v2 = _pub.arbitrage(now=LONG_AGO, **_A9)
_v1 = _pre.arbitrage(now=LONG_AGO, **_A9)
check("both plans still ALLOW a trade, so the two verdicts below are compared "
"against each other and not against a refusal",
_v2.kind == "allowed" and _v1.kind == "allowed")
@@ -744,8 +749,8 @@ print("2. AC2 - the spread stopped being taken over hours that have gone")
_A = dict(peak_guard_active=False, min_spread_eur_kwh=0.0,
uncertainty=frozenset())
_v_before = _before.arbitrage(**_A)
_v_after = _after.arbitrage(**_A)
_v_before = _before.arbitrage(now=LONG_AGO, **_A)
_v_after = _after.arbitrage(now=LONG_AGO, **_A)
check("both plans still ALLOW a trade, so this section compares two verdicts "
"rather than a verdict and a refusal",
_v_before.kind == "allowed" and _v_after.kind == "allowed")
@@ -843,7 +848,7 @@ check("...and no two of them share a whole hour by accident: the fixture puts "
_q_spread_want = round(max(p.eur_per_kwh for p in _q_win)
- min(p.eur_per_kwh for p in _q_win), 6)
_qv = _q_after.arbitrage(**_A)
_qv = _q_after.arbitrage(now=LONG_AGO, **_A)
_q_spread_got = _qv.match(allowed=lambda v: v.spread_eur_kwh,
refused=lambda _v: None)
check(f"...and the spread is the fixture's own max - min over that window "
+39 -10
View File
@@ -596,30 +596,59 @@ check("selling is off: no sale on any cycle",
# ============================================================================ #
print("\n5. modes off: byte-identical to release/1.0 e3e4ba6")
# Printed by tools/opt11p3_identity.py on release/1.0 e3e4ba6 (2026-10-13),
# goodwe-addon-checks:1, with this branch's harness (same API). n1 and n2 are
# also 5a4cdc2's; the lag ones moved with SAFETY-90/91 (n2_lag now equals n2:
# the next-quarter grading removed the edge tick's blindness at N=2).
AT_BASE = {
# Printed on release/1.0 e3e4ba6 (2026-10-13), goodwe-addon-checks:1, with this
# branch's harness (same API). n1 and n2 are also 5a4cdc2's; the lag ones moved
# with SAFETY-90/91 (n2_lag now equals n2: the next-quarter grading removed the
# edge tick's blindness at N=2).
#
# PRICE-02 (2026-10-22) changed two things in the trade sentence and nothing
# else (see test_modes_p3.py part 1): the "N above the cheapest" figure of a
# sale and "hour" -> "price period". Three pins, as there:
# AT_BASE_OLD the digests printed on e3e4ba6, sentence included (no longer
# matched: the mask is needed);
# AT_BASE the same rows through `ident.trade_digest()`, equal to what
# e3e4ba6 gives through it;
# AT_NOW this tree's own digests with nothing hidden.
AT_BASE_OLD = {
"n1": "71245b227ca40bada4f71f57ea6154475f4b0bd60239c9ad4f80bffb0094514c",
"n2": "56cf28010b06765c29e5479abb815d7503d570b39240d82dfc818efa90c93722",
"n1_lag": "86ef0f5ad8da1498ff7dcf59262a177bef847a0b201ce804e947c8eab5f07190",
"n2_lag": "56cf28010b06765c29e5479abb815d7503d570b39240d82dfc818efa90c93722",
}
_got = {k: ident.digest(ident.run(**ident.SCENARIOS[k])[0])
for k in AT_BASE}
AT_BASE = {
"n1": "33886ef03f364852883f0370230e6e04823fbb46a4704f88ca75bf9082d98c01",
"n2": "666a31f41491c22750720524853d95944d5fba395d1d14040c460d71414b275f",
"n1_lag": "f6c67b351bd732fb69e62cafbfdb2bcff6032f5d8845836659d096b8c677d3c0",
"n2_lag": "666a31f41491c22750720524853d95944d5fba395d1d14040c460d71414b275f",
}
AT_NOW = {
"n1": "939b3d47bbd64fa50e3d70538f092fb98e5c4bd5ead85a7fc696a1dbb514039e",
"n2": "c51008213dd739e1e5142c236d9a950aa585992393e4af120086f1ba52ff3e0d",
"n1_lag": "78aac892975be81a7ceac0c000ed441a731bd4f42ff9f729996cbee53d6b7f02",
"n2_lag": "c51008213dd739e1e5142c236d9a950aa585992393e4af120086f1ba52ff3e0d",
}
_cmd_digest = ident.trade_digest
_rows5 = {k: ident.run(**ident.SCENARIOS[k])[0] for k in AT_BASE}
_got = {k: _cmd_digest(v) for k, v in _rows5.items()}
_full5 = {k: ident.digest(v) for k, v in _rows5.items()}
for _k in AT_BASE:
print(f" {_k} {'same' if _got[_k] == AT_BASE[_k] else 'DIFFERS'} "
f"{_got[_k][:16]}")
check("with battery modes off every command, integrator, allocation and "
"trade sentence equals e3e4ba6's, also when every quarter boundary's "
"cycle still sees the old P1 block", _got == AT_BASE)
"trade action equals e3e4ba6's (the sentence masked as above), also when "
"every quarter boundary's cycle still sees the old P1 block",
_got == AT_BASE)
check("...the mask is needed and nothing more: unmasked, the digests differ "
"from e3e4ba6's printed ones, and equal this tree's own pins, so a "
"later change to any sentence or figure fails here",
all(_full5[k] != AT_BASE_OLD[k] for k in AT_BASE_OLD)
and _full5 == AT_NOW)
_real_project = tariff.PeakTracker.project
tariff.PeakTracker.project = staticmethod(
lambda a, now, roll=False: _real_project(a, now, roll=True))
try:
_moved = ident.digest(ident.run(**ident.SCENARIOS["n1_lag"])[0]) \
_moved = _cmd_digest(ident.run(**ident.SCENARIOS["n1_lag"])[0]) \
!= AT_BASE["n1_lag"]
finally:
tariff.PeakTracker.project = _real_project
+1 -1
View File
@@ -1178,7 +1178,7 @@ async def _modes_plan():
c = W.ctl
plan = c.fresh_tariff_plan(t)
verdict = plan.arbitrage(
peak_guard_active=False, min_spread_eur_kwh=0.05,
now=t, peak_guard_active=False, min_spread_eur_kwh=0.05,
uncertainty=frozenset())
return verdict.match(allowed=lambda v: ("allowed", v.spread_eur_kwh),
refused=lambda v: ("refused", v.reason))
+29 -24
View File
@@ -82,6 +82,11 @@ from app.validate import validate_startup_config
HERE = Path(__file__).resolve().parent
UTC = timezone.utc
# PRICE-02: `TariffPlan.arbitrage(now=...)` is required. This instant is before every
# point these fixtures build, so no period has ended and the whole series counts.
from datetime import datetime as _DT, timezone as _TZ # noqa: E402
LONG_AGO = _DT(2000, 1, 1, tzinfo=_TZ.utc)
fails = []
@@ -333,8 +338,8 @@ check(f"...while the two-period spread is the day/night difference "
f"{_spread_two}, re-derived here as {DAY_P - NIGHT_P:.4f}",
near(_spread_two, DAY_P - NIGHT_P, 1e-9))
_v_fixed = p_fixed.arbitrage(peak_guard_active=False, min_spread_eur_kwh=0.0, uncertainty=frozenset())
_v_two = p_two.arbitrage(peak_guard_active=False, min_spread_eur_kwh=0.0, uncertainty=frozenset())
_v_fixed = p_fixed.arbitrage(now=LONG_AGO, peak_guard_active=False, min_spread_eur_kwh=0.0, uncertainty=frozenset())
_v_two = p_two.arbitrage(now=LONG_AGO, peak_guard_active=False, min_spread_eur_kwh=0.0, uncertainty=frozenset())
check("a fully-fixed tariff REFUSES arbitrage", _v_fixed.kind == "refused")
check("...for the true reason - the series is flat, so there is nothing to buy "
"low and sell high, not because a FIXED enum was matched on",
@@ -354,7 +359,7 @@ _flat_dayahead = TariffPlan(
None, "provider")
check("a DAY_AHEAD series that happens to be flat is refused too, so the rule "
"is arithmetic and not a product label",
_flat_dayahead.arbitrage(peak_guard_active=False, min_spread_eur_kwh=0.0, uncertainty=frozenset()).kind == "refused")
_flat_dayahead.arbitrage(now=LONG_AGO, peak_guard_active=False, min_spread_eur_kwh=0.0, uncertainty=frozenset()).kind == "refused")
# The numbers an optimiser needs exist ONLY on the permitting variant.
check("the allowed verdict carries the spread, the cheapest hour and the "
@@ -379,16 +384,16 @@ check("...and there is no falsy variant, so `if verdict:` cannot look like a "
# §10.1 priority 2 over priority 3, expressed so OPT-04 cannot skip it.
check("peak shaving outranks arbitrage: the SAME allowed plan is refused when "
"the peak guard is holding",
p_two.arbitrage(peak_guard_active=True, min_spread_eur_kwh=0.0, uncertainty=frozenset()).kind == "refused")
p_two.arbitrage(now=LONG_AGO, peak_guard_active=True, min_spread_eur_kwh=0.0, uncertainty=frozenset()).kind == "refused")
check("...and the reason names the ordering rather than the tariff",
"priority 2" in p_two.arbitrage(peak_guard_active=True, min_spread_eur_kwh=0.0, uncertainty=frozenset()).reason)
"priority 2" in p_two.arbitrage(now=LONG_AGO, peak_guard_active=True, min_spread_eur_kwh=0.0, uncertainty=frozenset()).reason)
check("...and the question cannot be skipped: peak_guard_active is required "
"and keyword-only, so a spread is not obtainable without answering it",
raises(lambda: p_two.arbitrage(), TypeError)
and raises(lambda: p_two.arbitrage(False), TypeError))
check("...nor answered with a truthy 'unknown'",
raises(lambda: p_two.arbitrage(peak_guard_active=None), TypeError)
and raises(lambda: p_two.arbitrage(peak_guard_active="no"), TypeError))
raises(lambda: p_two.arbitrage(now=LONG_AGO, peak_guard_active=None), TypeError)
and raises(lambda: p_two.arbitrage(now=LONG_AGO, peak_guard_active="no"), TypeError))
# --------------------------------------------------------------------------- #
@@ -524,7 +529,7 @@ check("...and its underlying read has no .points at all, so `min(read.points)` "
"cannot run against it",
raises(lambda: _none.read.points, Absence))
check("...and arbitrage is refused, per §11.4",
_none.arbitrage(peak_guard_active=False, min_spread_eur_kwh=0.0, uncertainty=frozenset()).kind == "refused"
_none.arbitrage(now=LONG_AGO, peak_guard_active=False, min_spread_eur_kwh=0.0, uncertainty=frozenset()).kind == "refused"
and arbitrage_allowed(_none.read) is False)
check("...and static_status() says so in words rather than reporting a price",
static_status(opts(None), THU) == (False, "No fixed tariff set"))
@@ -776,12 +781,12 @@ check("...marked PROVISIONAL, which is the one flag that differs",
_fb.provisional and plan_is_provisional(_fb.read))
check("...and arbitrage is refused on it, per §11.4, even though a full series "
"exists and has a real spread",
_fb.arbitrage(peak_guard_active=False, min_spread_eur_kwh=0.0, uncertainty=frozenset()).kind == "refused"
_fb.arbitrage(now=LONG_AGO, peak_guard_active=False, min_spread_eur_kwh=0.0, uncertainty=frozenset()).kind == "refused"
and near(_fb.spread_eur_kwh(), DAY_P - NIGHT_P))
check("...while the SAME profile as the configured product allows it - so the "
"refusal is about provisionality and not about the profile",
static_plan(opts(TWO_PERIOD), THU, THU + timedelta(days=1)
).arbitrage(peak_guard_active=False, min_spread_eur_kwh=0.0, uncertainty=frozenset()).kind == "allowed")
).arbitrage(now=LONG_AGO, peak_guard_active=False, min_spread_eur_kwh=0.0, uncertainty=frozenset()).kind == "allowed")
check("...and the provisional plan has no cheap window either: §8.3 forbids an "
"irreversible action on a provisional price, and a grid top-up is one",
not any(_fb.cheap_now(at_local(THU, h)) for h in range(24)))
@@ -821,8 +826,8 @@ check("a live feed is used in preference to the static profile",
and _live_plan.product is PriceProduct.DAY_AHEAD)
check("...and arbitrage is allowed on it, with the spread derived from the "
"feed rather than from the profile",
_live_plan.arbitrage(peak_guard_active=False, min_spread_eur_kwh=0.0, uncertainty=frozenset()).kind == "allowed"
and near(_live_plan.arbitrage(peak_guard_active=False, min_spread_eur_kwh=0.0, uncertainty=frozenset()).spread_eur_kwh,
_live_plan.arbitrage(now=LONG_AGO, peak_guard_active=False, min_spread_eur_kwh=0.0, uncertainty=frozenset()).kind == "allowed"
and near(_live_plan.arbitrage(now=LONG_AGO, peak_guard_active=False, min_spread_eur_kwh=0.0, uncertainty=frozenset()).spread_eur_kwh,
0.023, 1e-9))
@@ -1036,7 +1041,7 @@ check(f"each tiny-spread probe really carries the spread it is named for, and "
all(near(p.spread_eur_kwh(), d, 1e-12)
and len([q for q in p.read.points if not q.provisional]) == 2
for d, p in _tiny9))
_tinyv9 = [(d, rung9(p.arbitrage(**_A9))) for d, p in _tiny9]
_tinyv9 = [(d, rung9(p.arbitrage(now=LONG_AGO, **_A9))) for d, p in _tiny9]
check(f"TEST-16 AC1: a spread that is SMALL BUT REAL is ALLOWED at "
f"min_spread_eur_kwh=0.0 - down to 0.000001 EUR/kWh, so the flat rung is "
f"`spread == 0.0` and not `spread <= epsilon` for any epsilon "
@@ -1045,7 +1050,7 @@ check(f"TEST-16 AC1: a spread that is SMALL BUT REAL is ALLOWED at "
check("...and the exactly flat series on the OTHER side of that boundary is "
"still refused, and refused BY THE FLAT RUNG - without this the sweep "
"above would also pass for a rung that had been deleted outright",
rung9(plan9([0.20000, 0.20000]).arbitrage(**_A9)) == "flat")
rung9(plan9([0.20000, 0.20000]).arbitrage(now=LONG_AGO, **_A9)) == "flat")
_q9 = plan9([0.20000, 0.2000005])
check(f"...and a difference below the rung's own 6-decimal rounding is flat BY "
f"CONSTRUCTION and says so: the raw difference is "
@@ -1056,10 +1061,10 @@ check(f"...and a difference below the rung's own 6-decimal rounding is flat BY "
f"to bisect further",
not near(_q9.spread_eur_kwh(), 0.0, 1e-9)
and near(round(_q9.spread_eur_kwh(), 6), 0.0, 1e-12)
and rung9(_q9.arbitrage(**_A9)) == "flat")
and rung9(_q9.arbitrage(now=LONG_AGO, **_A9)) == "flat")
# --- and what D's survival actually costs: the sentence, not the verdict -----
_lie9 = plan9([0.20000, 0.20050]).arbitrage(**_A9)
_lie9 = plan9([0.20000, 0.20050]).arbitrage(now=LONG_AGO, **_A9)
check(f"TEST-16 AC4: the 0.0005 verdict carries the real spread rather than a "
f"flatness claim about a tariff that is not flat - the false sentence is "
f"what a loosened rung ships to the operator while the outcome at a "
@@ -1085,19 +1090,19 @@ check(f"the one-firm probe is well formed: exactly ONE firm hour out of three, "
check("TEST-16 AC2: one firm hour is refused BY §8.3's FIRM-PAIR RUNG, named "
"in the reason - not by the flat rung that would also have refused it. "
"This is what takes `len(firm) < 2` -> `< 1` red on its own",
rung9(_one9.arbitrage(**_A9)) == "firm-pair")
rung9(_one9.arbitrage(now=LONG_AGO, **_A9)) == "firm-pair")
check("...and the sentence is §8.3's in full, so the operator is told the "
"series is half-published rather than told it is fixed-price",
"§8.3" in _one9.arbitrage(**_A9).reason
and "only 1 of 3" in _one9.arbitrage(**_A9).reason
and "flat" not in _one9.arbitrage(**_A9).reason)
"§8.3" in _one9.arbitrage(now=LONG_AGO, **_A9).reason
and "only 1 of 3" in _one9.arbitrage(now=LONG_AGO, **_A9).reason
and "flat" not in _one9.arbitrage(now=LONG_AGO, **_A9).reason)
# The rung pinned from ABOVE as well. Without this, `len(firm) < 3` - the same
# +0 B edit in the other direction - would pass everything above.
_two9 = plan9([0.20, 0.40, 0.10], [False, False, True])
check("...and TWO firm hours are ALLOWED on the identical series shape, so the "
"rung is pinned from above too and `< 3` is red as well as `< 1`",
rung9(_two9.arbitrage(**_A9)) == "allowed"
and near(_two9.arbitrage(**_A9).spread_eur_kwh, 0.20))
rung9(_two9.arbitrage(now=LONG_AGO, **_A9)) == "allowed"
and near(_two9.arbitrage(now=LONG_AGO, **_A9).spread_eur_kwh, 0.20))
# ...and from below, where the failure is not a wrong sentence but a crash:
# `max()` over an empty `firm` raises ValueError, so `< 0` does not degrade the
# reason, it takes the verdict away entirely.
@@ -1105,7 +1110,7 @@ _zero9 = plan9([0.20, 0.40, 0.10], [True, True, True])
check("...and a FULLY provisional series is refused by the same rung and does "
"not raise - `max()` over an empty firm set is a ValueError, so this row "
"is what makes `len(firm) < 0` red",
rung9(derived(lambda: _zero9.arbitrage(**_A9))) == "firm-pair")
rung9(derived(lambda: _zero9.arbitrage(now=LONG_AGO, **_A9))) == "firm-pair")
# --- AC3: the two rungs do not mask each other ------------------------------
# ⚠️ THE POINT OF THE TICKET, AND IT IS A PROPERTY OF THE TABLE, NOT OF A ROW.
@@ -1130,7 +1135,7 @@ _TABLE9 = (
("two firm hours of three", _two9, "allowed"),
("no firm hours at all", _zero9, "firm-pair"),
)
_got9 = [(name, want, rung9(derived(lambda p=pl: p.arbitrage(**_A9))))
_got9 = [(name, want, rung9(derived(lambda p=pl: p.arbitrage(now=LONG_AGO, **_A9))))
for name, pl, want in _TABLE9]
check(f"TEST-16 AC3: each stimulus is answered by the rung that OWNS it, and "
f"by no other - the flat rung is asked a question the firm-pair rung "
+583
View File
@@ -0,0 +1,583 @@
"""UX-46: the Dutch pages show Dutch status rows, notes and wizard lines.
`python3 -B test_ux46.py`
Before this, Home's status rows, the settings page's notes and the setup
wizard's checks were built in English and sent as they were, so the NL pages
showed English under Dutch titles (SAFETY-118 I-1, UX-44 I-2, UX-45 I-3).
1. Rows: every row in `messages.status_samples()` carries a Dutch twin
(`messages.Bi`), the Dutch follows STYLE.md, and the Home page's status
payload gives it for lang=nl and the English text for everything else.
2. Notes: every `_advise()` call in app/validate.py gives a Dutch twin, the
notes the validator really builds carry it, and /config/validate sends the
page's language, for the settings notes and for the wizard's check list.
3. The setting labels the Dutch notes quote are the labels in nl.yaml, and the
peak forecast label has no "(W)", since any power unit is read.
4. The rules above catch spoiled input.
Mutating a module this file grades? Read MUTATION.md first (purge __pycache__).
"""
import ast
import asyncio
import copy
import json
import logging
import os
import re
import sys
from datetime import datetime, timezone
from pathlib import Path
from aiohttp import ClientSession, web as aioweb
HERE = Path(__file__).resolve().parent
sys.path.insert(0, str(HERE / "tools"))
logging.disable(logging.WARNING)
import translations_lint as TL # noqa: E402
import ux13_demo as D # noqa: E402
from style_rules import sentences, text_problems, words # noqa: E402
from app import messages, uimodel as U, uitext, web # noqa: E402
from app.main import Controller # noqa: E402
from named_absence import field, nth, present # noqa: E402
from app.validate import (Note, note_text, setup_checks, # noqa: E402
config_notes)
fails = []
def check(name, cond):
print(f" {'ok ' if cond else 'FAIL'} {name}")
if not cond:
fails.append(name)
# STYLE.md for a Dutch row or note: the same length and shape rules, plus the
# Dutch words that test_ui_text.py bans for the same things. A quoted setting
# label may keep its own words ("Setpoint-entiteit" is the label in nl.yaml).
JARGON_NL = re.compile(
r"\b(?:setpoints?|setpunt\w*|SoC|laadtoestand|BMS|PV|failsafe|regelaar|"
r"regellus|controller|latch|ontlad\w*|payload|endpoint|JSON|API|"
r"time-?out|boolean|config|kwartuur\w*|commando\w*|geparkeerd|ESP32|"
r"master)\b", re.I)
# English function words that never belong in a Dutch sentence.
ENGLISH = re.compile(r"\b(?:the|not|and|for|with|from|reads|check|so|no|set|"
r"keeps|stays|has|have|ran|ago|seconds?|minutes?)\b", re.I)
QUOTED = re.compile(r"“[^”]*”")
def dutch_problems(text):
bare = re.sub(r"\b[a-z_]+\.[a-z0-9_]+\b", "X", QUOTED.sub("X", text))
out = []
if JARGON_NL.search(bare):
out.append("jargon: " + JARGON_NL.search(bare).group(0))
names = re.sub(r"\binverter-[a-z0-9]+\b|\b[a-z_]+\.[a-z0-9_]+\b", "X",
bare)
eng = ENGLISH.search(names)
if eng:
out.append("English word: " + eng.group(0))
return out
def row_problems_nl(colour, text):
n_s, n_w = sentences(text), len(words(text))
out = []
if colour == messages.RED:
if n_s > 3 or n_w > 30:
out.append(f"red row over 3 sentences / 30 words ({n_s}/{n_w})")
elif n_s > 1 or n_w > 15:
out.append(f"row over 1 sentence / 15 words ({n_s}/{n_w})")
if ";" in text:
out.append("semicolon")
return out + text_problems(text, keys=False, entities=False) \
+ dutch_problems(text)
# --------------------------------------------------------------------------- #
print("1. every status row has a Dutch twin, and the page picks one")
# --------------------------------------------------------------------------- #
ROWS = [(fn, c, t) for fn, c, t in messages.status_samples() if c is not None]
check(f"there are rows to grade ({len(ROWS)})", len(ROWS) > 80)
_none = [(fn, str(t)) for fn, c, t in ROWS
if not (isinstance(t, messages.Bi) and t.nl)]
check(f"every row has a Dutch text ({_none})", not _none)
_same = [(fn, str(t)) for fn, c, t in ROWS
if isinstance(t, messages.Bi) and t.nl == str(t)]
check(f"no Dutch row is the English one ({_same})", not _same)
_bad = [(fn, p) for fn, c, t in ROWS if isinstance(t, messages.Bi) and t.nl
for p in row_problems_nl(c, t.nl)]
for x in _bad:
print(f" STYLE {x}")
check(f"every Dutch row follows STYLE.md ({len(_bad)} problems)", not _bad)
BANNERS = [(fn, t) for fn, c, t in messages.status_samples() if c is None]
_nob = [(fn, str(t)) for fn, t in BANNERS
if not (isinstance(t, messages.Bi) and t.nl and t.nl != str(t))]
check(f"every banner has a Dutch text ({_nob})", BANNERS and not _nob)
_bb = [(fn, t.nl) for fn, t in BANNERS if isinstance(t, messages.Bi) and t.nl
and (len(words(t.nl)) > 12 or ";" in t.nl or dutch_problems(t.nl)
or text_problems(t.nl, keys=False, entities=False))]
check(f"every Dutch banner is at most 12 words and follows STYLE.md ({_bb})",
not _bb)
_cat = {k for k, v in vars(messages).items()
if callable(v) and k.startswith("row_")}
check("the samples cover every row_ function, so none can skip the check",
_cat <= {fn for fn, _c, _t in ROWS})
_lat = {"row_direction_trip": "herstart heft dit niet op",
"row_site_direction_trip": "regeling starten heft dit op",
"row_circulating_trip": "regeling starten heft dit op",
"row_not_delivering": "stoppen en starten heft dit op"}
_unclear = [fn for fn, _c, t in ROWS if fn in _lat and _lat[fn] not in t.nl]
check(f"a Dutch latch row says how it clears too ({_unclear})", not _unclear)
check("pick() gives the English for en, the Dutch for nl, and leaves other "
"text alone",
messages.pick(messages.row_estop_fitted(), "en") == "Emergency stop: fitted"
and type(messages.pick(messages.row_estop_fitted(), "en")) is str
and messages.pick(messages.row_estop_fitted(), "nl") == "Noodstop: geplaatst"
and messages.pick("Plain", "nl") == "Plain"
and messages.pick(messages.Bi("A"), "nl") == "A")
check("a Bi survives copy and pickle, and JSON",
copy.deepcopy(messages.row_estop_fitted()).nl == "Noodstop: geplaatst"
and json.dumps(messages.row_estop_fitted()) == '"Emergency stop: fitted"')
# --- the page: a real Controller -------------------------------------------- #
class _Store:
def __init__(self):
self.data = {}
def set(self, k, v):
self.data[k] = v
def get(self, k, d=None):
return self.data.get(k, d)
def set_time(self, k, v):
self.data[k] = v.isoformat()
def get_time(self, k):
v = self.data.get(k)
return None if v is None else datetime.fromisoformat(v)
class _Pub:
enabled = False
def publish(self, _v):
pass
def close(self):
pass
class _Hass:
session = None
token = ""
async def limits(self, _e):
return (-5000.0, 5000.0, 1.0)
async def number(self, _e, invert=False):
return None
async def state(self, _e):
return None
async def set_number(self, _e, _v):
return True
async def call(self, *_a):
return True
async def states(self):
return []
ONE = {"setpoint_entity": "number.sp", "meter_entity": "sensor.grid",
"soc_entity": "sensor.soc", "batt_entity": "sensor.batt",
"meter_source": "off", "max_w": 5000, "shadow_mode": "off",
"maintenance_enabled": False, "peak_forecast_entity": "sensor.fc"}
TWO = dict(ONE, inverters=[
{"name": "Battery A", "setpoint_entity": "number.a", "soc_entity":
"sensor.soc_a", "batt_entity": "sensor.batt_a", "max_w": 2500},
{"name": "Battery B", "setpoint_entity": "", "soc_entity":
"sensor.soc_b", "batt_entity": "sensor.batt_b", "max_w": 2500}])
def make(opts, soc=(62.0, 35.0), batt=(-2500.0, -300.0)):
c = Controller(dict(opts), _Hass(), _Store(), _Pub(), stop=asyncio.Event())
c.auto = True
for i, inv in enumerate(c.inverters):
inv.soc, inv.batt = soc[i], batt[i]
inv.dev_min, inv.dev_max = -5000.0, 5000.0
c.grid = 10.0
return c
def attention(payload, rid):
return next((a for a in payload["attention"] if a["id"] == rid), None)
c = make(ONE, soc=(62.0,), batt=(960.0,))
c._peak_fc_fault = "kWh"
s_en, s_nl = c.status_payload(), c.status_payload(lang="nl")
def fc_row(payload):
return nth([r for r in payload["checks"] if r.get("id") == "peak_forecast"])
_en, _nl = fc_row(s_en), fc_row(s_nl)
check("Home, English: the peak forecast row is the English sentence",
present(_en) and field(_en, "text") == messages.peak_forecast_unit("kWh")
and attention(s_en, "peak_forecast")["body"]
== messages.peak_forecast_unit("kWh"))
check("Home, Dutch: the row, the Needs-attention entry and the title are "
"Dutch",
present(_nl) and field(_nl, "text") == messages.peak_forecast_unit("kWh", "nl")
and attention(s_nl, "peak_forecast")["body"]
== messages.peak_forecast_unit("kWh", "nl")
and attention(s_nl, "peak_forecast")["title"]
== uitext.words("nl")["home_words"]["attn"]["peak_forecast"])
c._peak_fc_fault = ""
_nl0 = fc_row(c.status_payload(lang="nl"))
check("...and with no unit at all, the Dutch is the 'no unit' sentence",
present(_nl0) and field(_nl0, "text") == messages.peak_forecast_unit("", "nl")
and messages.peak_forecast_unit("", "nl")
!= messages.peak_forecast_unit("kWh", "nl"))
check("a language that is not Dutch gets the English, and the default call "
"is unchanged",
c.status_payload(lang="fr")["checks"] == c.status_payload()["checks"]
and all(type(r["text"]) is str for r in c.status_payload()["checks"]))
# A site with faults of many kinds: every row it shows is Dutch on the NL page.
SCEN = {}
SCEN["two, one without a setpoint, meter stale"] = make(TWO)
SCEN["one, nothing read"] = make(ONE, soc=(None,), batt=(None,))
SCEN["one, no forecast set"] = make(dict(ONE, peak_forecast_entity=""))
SCEN["one, shadow, estop fitted"] = make(dict(ONE, shadow_mode="on",
estop_fitted=True))
SCEN["one, solar sources"] = make(dict(ONE, pv_entity="sensor.pv"))
SCEN["two, solar list"] = make(dict(TWO, pv_entities=["sensor.a", "sensor.b"]))
SCEN["one, peak forecast unit fault"] = c
SCEN["one, control stopped"] = make(ONE)
SCEN["one, control stopped"].auto = False
_English, _rows_seen = [], 0
for name, ctl in SCEN.items():
raw = ctl.checks()
en = ctl.status_payload()["checks"]
nl = ctl.status_payload(lang="nl")["checks"]
check(f"[{name}] {len(raw)} rows: same rows in both languages",
len(raw) == len(en) == len(nl))
for r, e, n in zip(raw, en, nl):
_rows_seen += 1
if not (isinstance(r["text"], messages.Bi) and r["text"].nl):
_English.append((name, r.get("id"), str(r["text"])))
elif e["text"] != str(r["text"]) or n["text"] != r["text"].nl:
_English.append((name, r.get("id"), "page did not pick"))
_banners = {n: (ctl.status_payload()["banner"],
ctl.status_payload(lang="nl")["banner"])
for n, ctl in SCEN.items()}
print(f" banners: {_banners}")
check("the banner at the top of Home is Dutch on the Dutch page, English on "
"the English one", all(e != n and not dutch_problems(n)
for e, n in _banners.values()))
check(f"{_rows_seen} rows over {len(SCEN)} sites: none shows English on the "
f"Dutch page when it has no Dutch text ({_English})", not _English)
check("the row's details list in /status is picked the same way",
all(isinstance(v, str) and not isinstance(v, messages.Bi)
for ctl in SCEN.values()
for _k, v in ctl.status_payload(lang="nl")["rows"]))
# --------------------------------------------------------------------------- #
print("\n2. every settings note and wizard line has a Dutch twin")
# --------------------------------------------------------------------------- #
tree = ast.parse((HERE / "app" / "validate.py").read_text(encoding="utf-8"))
_calls = [n for n in ast.walk(tree) if isinstance(n, ast.Call)
and ast.unparse(n.func) == "_advise"]
_bare = [n.lineno for n in _calls if not any(k.arg == "nl" for k in n.keywords)]
check(f"every _advise() call in validate.py gives `nl` ({len(_calls)} calls, "
f"without: {_bare})", len(_calls) >= 30 and not _bare)
OPTS = dict(D.defaults(D.CONFIG), setpoint_entity="number.sp",
meter_entity="sensor.grid", soc_entity="sensor.soc",
batt_entity="sensor.batt", pv_entity="sensor.pv",
peak_forecast_entity="sensor.fc")
def ent(state, unit=None):
return {"state": state, "unit": unit}
GOOD = {"number.sp": ent("0"), "sensor.grid": ent("402", "W"),
"sensor.soc": ent("62", "%"), "sensor.batt": ent("-800", "W"),
"sensor.pv": ent("-1200", "W"), "sensor.fc": ent("820", "kWh")}
NOTES = []
NOTES += setup_checks(OPTS, entities=GOOD, running=None)
NOTES += setup_checks(OPTS, entities=None)
NOTES += setup_checks(dict(OPTS, meter_invert=True, pv_invert_entities=["sensor.pv"]),
entities=dict(GOOD, **{"sensor.batt": ent("0", "W")}))
NOTES += setup_checks(dict(OPTS, soc_entity="sensor.x", batt_entity="sensor.y",
setpoint_entity="number.typo"),
entities=GOOD)
NOTES += setup_checks(dict(OPTS, meter_source="homewizard_local"),
entities=GOOD, running=None)
NOTES += setup_checks(dict(OPTS, meter_source="homewizard_local",
price_provider="generic_dayahead"),
entities=GOOD, running=dict(OPTS, meter_source="homewizard_local",
price_provider="generic_dayahead"),
meter_now_w=-350.0, prices_known=18)
NOTES += setup_checks(dict(OPTS, price_provider="generic_dayahead"), entities=GOOD,
running=None)
NOTES += setup_checks(dict(OPTS, price_provider="off", static_profile=""),
entities=GOOD)
NOTES += setup_checks(dict(OPTS, price_provider="off",
static_profile='{"periods":[]}'), entities=GOOD)
NOTES += setup_checks(dict(OPTS, batt_entity="sensor.batt2"),
entities=dict(GOOD, **{"sensor.batt2": ent("unavailable")}))
NOTES += setup_checks(dict(OPTS, soc_entity="sensor.soc2"),
entities=dict(GOOD, **{"sensor.soc2": ent("unavailable")}))
NOTES += setup_checks(dict(OPTS, meter_entity="sensor.m2"),
entities=dict(GOOD, **{"sensor.m2": ent("unavailable")}))
_TWO = dict(OPTS, inverters=[
{"name": "a", "setpoint_entity": "number.sp", "soc_entity": "sensor.soc",
"batt_entity": "sensor.batt", "max_w": 2000, "bms_soc_floor": 10},
{"name": "b", "setpoint_entity": "number.b", "max_w": 2000,
"bms_soc_floor": 10}])
NOTES += setup_checks(_TWO, entities=dict(GOOD, **{"sensor.batt": ent("-0.5", "kW")}))
NOTES += setup_checks(dict(OPTS, meter_source="ha_signed",
p1_net_entity="sensor.net"),
entities=dict(GOOD, **{"sensor.net": ent("-3.1", "kW")}))
NOTES += config_notes(dict(OPTS, integrator_max_w=9000, max_w=2000,
peak_cap_w=2000), entities=GOOD, month_peak_w=3000)
NOTES += config_notes(dict(OPTS, pv_forecast_entity="sensor.f1",
pv_forecast_tomorrow_entity="sensor.f2",
peak_forecast_entity="sensor.fc2"),
entities=dict(GOOD, **{"sensor.f1": ent("5", "W"),
"sensor.f2": ent("6", "kWh"),
"sensor.fc2": ent("5")}))
NOTES += config_notes(dict(OPTS, meter_entity="sensor.gone"), entities=GOOD)
NOTES += config_notes(dict(OPTS, pv_forecast_tomorrow_entity="sensor.f2"),
entities=dict(GOOD, **{"sensor.f2": ent("6", "kWh")}))
NOTES += config_notes(dict(OPTS, batt_entity="sensor.kw"),
entities=dict(GOOD, **{"sensor.kw": ent("1", "kW")}))
print(f" {len(NOTES)} notes built")
_nonl = [n.text for n in NOTES if not n.nl]
check(f"every note the validator builds has a Dutch twin ({_nonl})",
NOTES and not _nonl)
_same = [n.text for n in NOTES if n.nl == n.text]
check(f"...which is not the English ({_same})", not _same)
_bad = []
for n in NOTES:
nw, ns = len(words(n.nl)), sentences(n.nl)
p = text_problems(n.nl, keys=True, entities=True) + dutch_problems(n.nl)
if nw > 25 and not n.text.startswith(("The 15-minute", "Home Assistant")):
p.append(f"over 25 words ({nw})")
if ns > 2:
p.append(f"over 2 sentences ({ns})")
if ";" in n.nl:
p.append("semicolon")
if len(words(n.nl)) > len(words(n.text)) + 6:
p.append("much longer than the English")
_bad += [(n.nl, x) for x in p]
for x in _bad:
print(f" STYLE {x}")
check(f"every Dutch note follows STYLE.md ({len(_bad)} problems)", not _bad)
_l = {t: [n for n in NOTES if n.text.startswith(t)] for t in (
"Prices for today are in", "The 15-minute demand forecast")}
check("the wizard's price line, and the peak forecast unit note, are among "
"them", all(_l.values()))
check("note_text() picks: Dutch for nl, English for en, the old twin table "
"for a note with no `nl`",
note_text(_l["Prices for today are in"][0], "nl")
== "De prijzen van vandaag zijn binnen, voor 18 van 24 uur."
and note_text(_l["Prices for today are in"][0], "en")
== "Prices for today are in, for 18 of 24 hours."
and note_text(Note("info", (), messages.FIXED_TARIFF_NOTE), "nl")
== messages.FIXED_TARIFF_NOTE_NL
and note_text(Note("info", (), "Other"), "nl") == "Other")
# --- over HTTP: the settings page and the wizard, in both languages --------- #
SCHEMA = D.schema_ui(D.CONFIG)
TRANSLATIONS = {l: TL.parse_translation_file(HERE / "translations" / f"{l}.yaml")
for l in ("en", "nl")}
STATES = [{"entity_id": e, "state": v["state"],
"attributes": ({"unit_of_measurement": v["unit"]} if v["unit"]
else {})} for e, v in GOOD.items()]
class Hass:
session = None
token = ""
async def limits(self, _e):
return (-5000.0, 5000.0, 1.0)
async def states(self):
return copy.deepcopy(STATES)
async def state(self, entity):
return next((copy.deepcopy(s) for s in STATES
if s["entity_id"] == entity), None)
class Supervisor:
def __init__(self, options):
self.options = copy.deepcopy(options)
async def info(self):
return {"options": copy.deepcopy(self.options), "schema": SCHEMA,
"translations": TRANSLATIONS}
async def write_options(self, options):
pass
async def restart(self, confirm):
pass
class Ctl:
def __init__(self, opts):
self.o, self.hass, self.inverters = dict(opts), Hass(), []
self.store, self.auto, self.grid = None, True, 402.0
def log_event(self, text, hint=None):
pass
def status_payload(self, lang="en"):
return {"banner": "", "level": "ok", "rows": [], "checks": [],
"plan": {"prices": [{"label": f"{h:02d}:00", "eur_kwh": 0.2}
for h in range(20)]
+ [{"label": f"{h:02d}:00", "eur_kwh": None}
for h in range(20, 24)]}}
async def serve(opts, steps):
runner = aioweb.AppRunner(web.build_app(Ctl(opts), Supervisor(opts)))
await runner.setup()
site = aioweb.TCPSite(runner, "127.0.0.1", 0)
await site.start()
root = f"http://127.0.0.1:{site._server.sockets[0].getsockname()[1]}/"
try:
async with ClientSession() as s:
return await steps(s, root)
finally:
await runner.cleanup()
async def post(s, url, body):
async with s.post(url, json=body) as r:
return await r.json()
RUN = dict(OPTS, meter_source="homewizard_local", p1_host="192.0.2.50",
price_provider="generic_dayahead")
async def pages(s, root):
out = {}
# The settings notes read the entity list the picker has cached, so the
# first call only warms it.
await post(s, root + "config/validate",
{"model": U.options_to_ui(RUN), "setup": True})
for lang in ("en", "nl"):
async with s.get(root + f"config/model?lang={lang}") as r:
m = await r.json()
model = U.options_to_ui(RUN)
out[lang] = await post(s, root + "config/validate",
{"model": model, "setup": True, "lang": lang})
out["none"] = await post(s, root + "config/validate", {
"model": U.options_to_ui(RUN), "setup": True})
return out
R = asyncio.run(serve(RUN, pages))
_texts = {l: [m["text"] for m in R[l].get("messages", [])] for l in ("en", "nl")}
_setup = {l: [m["text"] for m in R[l].get("setup", [])] for l in ("en", "nl")}
FC_EN = messages.peak_forecast_unit("kWh") + " Pick a sensor in W or kW."
FC_NL = messages.peak_forecast_unit("kWh", "nl") + " Kies een sensor in W of kW."
print(f" settings notes en: {_texts['en']}")
print(f" settings notes nl: {_texts['nl']}")
print(f" wizard lines nl: {_setup['nl']}")
check("settings page, English: the peak forecast unit note is English",
FC_EN in _texts["en"])
check("settings page, Dutch: it is the Dutch sentence, and no English "
"sentence of it is left", FC_NL in _texts["nl"]
and FC_EN not in _texts["nl"])
check("wizard, English: the price line and the checks are English",
"Prices for today are in, for 20 of 24 hours." in _setup["en"])
check("wizard, Dutch: the price line is Dutch",
"De prijzen van vandaag zijn binnen, voor 20 van 24 uur."
in _setup["nl"] and "Prices for today are in, for 20 of 24 hours."
not in _setup["nl"])
_left = [t for t in _setup["nl"] + _texts["nl"]
if ENGLISH.search(QUOTED.sub("X", t))
and not t.startswith("Home Assistant has no entity")]
check(f"no wizard line or settings note on the Dutch page has English words "
f"left ({_left})", not _left)
check("a request with no language gets English (an old page keeps working)",
[m["text"] for m in R["none"].get("setup", [])]
== [m["text"] for m in R["en"].get("setup", [])])
check("the raw `notes` list keeps the English `text` and adds `nl`",
R["nl"]["notes"] and all(n["text"] != n["nl"] for n in R["nl"]["notes"]))
# --------------------------------------------------------------------------- #
print("\n3. labels")
# --------------------------------------------------------------------------- #
for lang in ("en", "nl"):
name = TRANSLATIONS[lang]["configuration"]["peak_forecast_entity"]["name"]
check(f"[{lang}] the peak forecast label has no unit: {name!r}",
"(W)" not in name and "(kW)" not in name
and name.split("·")[-1].strip() == (
"15-minute demand forecast entity" if lang == "en"
else "Entiteit kwartierprognose"))
check("DOCS.md shows the same label, so the table cannot drift",
"| Limits | 15-minute demand forecast entity | `peak_forecast_entity` |"
in (HERE / "DOCS.md").read_text(encoding="utf-8"))
def stems(key):
tr = TRANSLATIONS["nl"]["configuration"]
out = []
for c in (tr, tr["inverters"]["fields"]):
if isinstance(c.get(key), dict):
n = c[key]["name"].split("·")[-1].strip()
while re.search(r"\s*\([^()]*\)$", n):
n = re.sub(r"\s*\([^()]*\)$", "", n)
out.append(n)
return out
_drift = sorted(k for k, v in messages.LABELS_NL.items() if v not in stems(k))
check(f"every Dutch label a note quotes is the setting's name in nl.yaml "
f"(drifted: {_drift})", not _drift
and set(messages.LABELS_NL) == set(messages.LABELS))
_unit = sorted(k for k, v in messages.UNIT_LABELS_NL.items() if v not in stems(k))
check(f"...and so are the two the inverter list names differently ({_unit})",
not _unit)
check("setting() in Dutch keeps the key, so the config panel can find it",
messages.setting("max_w", "", "nl") == "“Vermogen van de omvormer” (max_w)"
and messages.setting("max_w", "inverter-b", "nl")
== "“Vermogen van deze omvormer” voor inverter-b (inverters.max_w)"
and messages.setting("max_w") == "“Inverter rating” (max_w)")
# --------------------------------------------------------------------------- #
print("\n4. the rules catch spoiled input")
# --------------------------------------------------------------------------- #
check("a Dutch row left in English is caught",
row_problems_nl(messages.AMBER, "Control: last ran 2 seconds ago"))
check("an over-long Dutch row is caught",
row_problems_nl(messages.AMBER, " ".join(["woord"] * 16) + "."))
check("Dutch jargon is caught", dutch_problems("Het commando is geweigerd"))
check("a row with no Dutch is caught",
not isinstance(messages.pick("x", "nl"), messages.Bi)
and not getattr(messages.Bi("x"), "nl"))
check("the untranslated-call scan sees a bare _advise",
any(isinstance(n, ast.Call) and not any(k.arg == "nl" for k in n.keywords)
for n in ast.walk(ast.parse('_advise(notes, "warn", (), "x")'))))
print()
if fails:
print(f"{len(fails)} FAILED:")
for f in fails:
print(" -", f)
sys.exit(1)
print("all passed")
@@ -17,6 +17,7 @@ test_modes_p3.py part 1 holds this tree to the digests printed on 7a00f12.
import functools
import hashlib
import os
import re
import sys
import tempfile
import time
@@ -117,6 +118,31 @@ def digest(rows) -> str:
return hashlib.sha256(repr(rows).encode()).hexdigest()
def _trade_words(text):
"""PRICE-02: the trade sentence as release/1.0 before it printed it. The
plan now drops price periods that have ended, so a sale's "above the
cheapest" figure is a different number once the first cheap period is
over, and the sentences say "price period" where they said "hour". Both
maps are applied to both sides, so only those two things are hidden.
ONLY FOR COMPARING WITH A TREE OLDER THAN PRICE-02. A pin of this tree
itself uses `digest()`, with nothing hidden."""
text = text.replace("this hour is neither the buy hour nor the sell hour",
"now is not the cheapest or the dearest price period")
text = re.sub(r"[0-9]+\.[0-9]+ above the cheapest",
"N above the cheapest", text)
return text.replace("firm hour at", "firm price period at")
def trade_digest(rows) -> str:
"""`digest()` with the trade sentence passed through `_trade_words()`:
every command, integrator and allocation, and the action, watts and the
rest of the sentence, byte for byte."""
return digest([r[:-1] + (None if r[-1] is None else
(r[-1][0], r[-1][1], _trade_words(r[-1][2])),)
for r in rows])
def digests() -> dict:
return {k: digest(run(**v)[0]) for k, v in SCENARIOS.items()}
+1 -1
View File
@@ -164,7 +164,7 @@ configuration:
name: "Limits · Margin below the peak limit (W)"
description: "The add-on aims this much below the peak limit, so a 15-minute period that ends a little high still stays under it."
peak_forecast_entity:
name: "Limits · 15-minute demand forecast entity (W)"
name: "Limits · 15-minute demand forecast entity"
description: "Optional. A forecast of this 15-minute period's average power, in W or kW. With no power unit it is not used. Leave it empty and there is no peak limit."
peak_soc_floor:
name: "Limits · Peak reserve (%)"
+1 -1
View File
@@ -145,7 +145,7 @@ configuration:
name: "Grenzen · Marge onder de piekgrens (W)"
description: "De add-on mikt zoveel onder de piekgrens, zodat een kwartier dat iets hoger uitvalt er toch onder blijft."
peak_forecast_entity:
name: "Grenzen · Entiteit kwartierprognose (W)"
name: "Grenzen · Entiteit kwartierprognose"
description: "Optioneel. Een prognose van het gemiddelde vermogen in het lopende kwartier, in W of kW. Zonder vermogenseenheid wordt ze niet gebruikt. Laat u dit leeg, dan is er geen piekgrens."
peak_soc_floor:
name: "Grenzen · Piekreserve (%)"