SAFETY-79 review: margin under a banked month peak too; identity harness reaches headroom and maintenance

- quarter_target_w() is now the margin under quarter_ceiling_w(): a peak
  banked between cap - margin and the cap was aimed at exactly, so one
  overshoot crossed the cap; above the cap the same overshoot raised the bill.
- safety79_identity.py hashes charge_headroom_w() and the shaping suspension,
  and adds n1_maint/n2_maint (cheap window, forecast across cap - 50 .. cap).
  Digests re-printed on ed739d4 and pinned; margin 0 still identical.
- test_margin_under_banked_peak; DECISIONS and DOCS updated.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

goodwe_controller/DECISIONS/OPT-11-battery-modes.md

@@ -17,5 +17,16 @@ P1 is its section 6 P1, and the pull request lists what was measured.
   capped grid charge release an Off battery. Both are graded in
   `test_arbiter.py` part 11. The loop bound leaves a side that reaches the
   rail to the rail, so SAFETY-03 still flags a loop winding into it.
+- P2 (HA entities): a select and a switch per battery, only while modes are
+  on; commands go through `handle_action()`, the page's path. A retained
+  command is dropped, not applied: the store already holds the last choice,
+  and a replay after a restart could otherwise undo a newer one. The config
+  topics announced are kept in the store (`mqtt_mode_configs`), so turning
+  modes off removes exactly those, and an install that never turned them on
+  publishes what ed739d4 did (`test_mqtt.py` part 10). A removed topic stays
+  in that list and its empty config is re-sent on each start, so a start with
+  no broker cannot lose it; re-sending is harmless. The HA options are the
+  page's English names, pinned by a test, and are not renamed with the page:
+  automations send them.
 - In P1 only Off closes a direction; Leftover solar differs by charging last.
   Price runs, selling (`pack_sell`, reserved) and the forecast are P3/P4.

goodwe_controller/DECISIONS/SAFETY-79-peak-margin.md

@@ -0,0 +1,41 @@
+# SAFETY-79 — the peak guard keeps a margin below the peak limit
+
+**Status: DECIDED 2026-09-28 (owner decision).**
+
+## Decision
+
+The peak guard aims `peak_margin_w` under the limit it holds: under
+`peak_cap_w`, or under a month peak already banked above it
+(`quarter_target_w()`). `peak_margin_w` is a setting: default 50 W, range
+0-500 W, whole watts. An install without the key gets 50.
+
+**Why (owner):** the guard aimed exactly at the cap, and in the OPT-11 P1 test
+a quarter banked 3501.9 W against a 3500 W limit (OPT-11-P1-test-report.md,
+check 4). A peak even 1 W over still sets the billed peak. It is a setting and
+not a constant because the add-on is sold to several sites.
+
+**Banked peak (review, 2026-09-29):** the PR first aimed at a banked month
+peak as it stood. That can bank a few W over it, which raises the bill the
+same way; and a peak banked between cap - margin and the cap (common in the
+month an install upgrades, since the old guard held quarters at the cap) was
+aimed at exactly, so one overshoot went over the cap. The margin now applies
+under it too. Cost: up to `peak_margin_w` of a peak already paid for goes
+unused (more discharge, less charge headroom); no bill cost. Undo is one line
+in `quarter_target_w()`.
+
+## What uses which number
+
+- **the aim:** `peakguard.decide()` (the claim's floor, and so arbitrage's
+  `charge_headroom_w()` and OPT-11's give-way, which read it), and, as cap -
+  margin, `shaping_suspended()` and `maintenance_charge_floor()` (the forecast
+  checks, via `Controller.peak_aim_w()`).
+- **the real cap (reporting and billing):** `PeakTracker` (month peak,
+  headroom, 80 % warning), the decision's `ceiling_w` and every notice and
+  status row built from it, Home's limit, and `validate.config_notes()`.
+
+## Checks
+
+`peak_margin_w: 0` is byte-identical to ed739d4 (test_peak_margin.py part 1,
+digests from tools/safety79_identity.py, which also runs a maintenance charge
+with a forecast). validate.py rule 12 refuses a value outside 0-500, not
+whole, or not below `peak_cap_w`.

goodwe_controller/DOCS.md

@@ -204,6 +204,7 @@ The label and description each setting shows on the Configuration tab, in tab or
 | Limits | Inverter rating (W) | `max_w` | Your own number: the most power the add-on may ever ask of the inverter. Start low and raise it after commissioning. |
 | 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. 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. |
@@ -274,6 +275,7 @@ When the add-on refuses to start, each reason names the setting by its label and
 - **Inverter rating** (`max_w`): your own number, not a derived one. It is not read when the inverter list is filled in, since each entry carries its own and the site limit is their sum. Several safety thresholds must stay below it, and the add-on refuses to start if one does not.
 - **Battery cut-out level** (`bms_soc_floor`, `inverters.bms_soc_floor`): measured, not chosen. Register 0x0560 reads 10 % on the reference GW5000S-BP. It exists so the add-on can refuse a “Peak reserve” at or below it, which could never take effect. Each inverter entry states its own, because one default for two different batteries protects the shallower one with the deeper one's number.
 - **Peak limit** (`peak_cap_w`): your own number, from your grid contract. In Belgium it is the capacity-tariff peak your distribution charge is billed on.
+- **Margin below the peak limit** (`peak_margin_w`): 50 W by default, 0 to 500 W (SAFETY-79). Peak limiting aims at the peak limit minus this margin, because a 15-minute period aimed exactly at the limit can end a watt or two over it (3501.9 W against 3500 W in a test), and even 1 W over sets the billed peak. Once this month's peak is already above the limit, it aims the same margin below that peak. The bill, this month's peak and the warnings still compare against the peak limit itself. 0 aims at the limit, as before.
 - **Peak reserve** (`peak_soc_floor`): derived, not picked (SAFETY-24). The reference battery holds 13.8 kWh usable and the worst modelled peak day asks for 878 Wh, or 6.4 %. 20 % sits ten points above the cut-out level and leaves about 1.6 worst days of room. It bounds peak limiting only: ordinary use goes on down to the cut-out level. It must stay above “Battery cut-out level”.
 - **Dynamic price source** (`price_provider`): selecting a provider is what starts the fetch, every half hour, whether or not trading is on. It may not be combined with the two older price entities. engie_dynamic, totalenergies_dynamic and eneco_eddy are withdrawn and refuse to start: their endpoints never existed. They stay in the list only so such a site gets that explanation.
 - **Supplier markup** (`price_markup_eur_kwh`): under manual_csv, whose prices are all in, it is ignored and logged as ignored.
@@ -522,6 +524,10 @@ Off by default. Turn it on, restart the add-on, and the Batteries screen (and Se
 
 The peak limit, the safety stops and the monthly cycle always come before a battery's mode.
 
+**In Home Assistant.** With MQTT set up, each battery also gets a *Mode* list and a *Use* switch, for example `select.goodwe_inverter_a_mode` and `switch.goodwe_inverter_a_use`. They do the same as the Batteries screen. Turn battery modes off and restart, and they are removed.
+
+*For installers:* commands arrive on `goodwe_ctl/unit/<battery>/mode/set` (a *Mode* option, such as `Solar first`) and `.../use/set` (`ON` or `OFF`). They go through the same checks as the Batteries screen. A refusal is logged once, and later refusals are counted in the log. A retained command is never applied, so the broker cannot replay an old choice after a restart; the add-on keeps the last choice itself.
+
 *For installers:* a mode is a limit at priority 35, kept apart from the hard limits. It gives way only as far as a limit above it (the peak guard) needs. A mode range always holds 0 W, so a mode can forbid a direction and never force one. With one inverter the mode also bounds the loop. The record is `DECISIONS/OPT-11-battery-modes.md`.
 
 ### Reading the grid-following loop: `Loop integrator`, `Loop integrator lead`, `Loop reason`
@@ -1612,6 +1618,7 @@ help - see "Why the tuning is what it is".
 |---|---|
 | `peak_forecast_entity` | Quarter-hour demand forecast, for capacity-tariff markets. Empty = no cap |
 | `peak_cap_w` | The site's capacity-tariff target |
+| `peak_margin_w` (50) | Watts the peak guard aims under `peak_cap_w` (SAFETY-79). Reports and the bill use `peak_cap_w` itself |
 | `peak_soc_floor` (20) | **The percent below which the capacity-peak guard will not discharge the battery.** ⚠️ It bounds **the peak guard only** — it is **not a pack-wide reserve**, and ordinary self-consumption discharges straight past it (SAFETY-25, see below). Not optional and not the same thing as `maintenance_soc_floor` |
 | `bms_soc_floor` (10) | The pack's **own** cut-out percent, the number the BMS enforces whatever you set above. `peak_soc_floor` must be **strictly above** it or the add-on refuses to start. ⚠️ **At a declared `inverters:` roster this key moves INTO each entry and is REQUIRED there** (SAFETY-32): it describes a battery, and two packs of different models have different cut-outs, so one number for both protects the shallower pack with the deeper pack's BMS. Nothing is inherited — an entry that omits it is refused **by name** at startup, and the flat key below is then not read at all |
 | `price_now_entity`, `price_avg_entity` | Dynamic tariff. Empty = never force a paid grid top-up |

goodwe_controller/app/main.py

@@ -72,8 +72,9 @@ from .maintenance import (
     IDLE, POWER_MOVING, MaintConfig, Maintenance, MaintResult,
     arbitrate as maint_arbitrate,
 )
-from .mqtt import (MqttPublisher, announced_units, optional_keys, unit_slugs,
-                   unit_state_key)
+from .mqtt import (MODE_CONFIGS_KEY, MODE_LABELS, USE_OFF, USE_ON,
+                   MqttPublisher, announced_units, mode_configs,
+                   optional_keys, parse_command, unit_slugs, unit_state_key)
 from . import peakguard
 from . import packshare
 from . import pv as pv_read
@@ -893,6 +894,10 @@ class Controller:
         # every option family this file builds: no numeric fallback in main.py
         # that config.yaml has not shipped.
         self.peak_guard = None
+        # SAFETY-79. The guard aims this far under `peak_cap_w`; the bill and
+        # every report keep the real cap (`self.peak.cap_w`).
+        self.peak_margin_w = float(opts.get(
+            "peak_margin_w", peakguard.PEAK_OPTION_DEFAULTS["peak_margin_w"]))
         self._peak_engaged = False
         self._peak_saturated = False
         self._peak_floor_limited = False
@@ -901,9 +906,12 @@ class Controller:
         # turned them on behaves exactly as before (SAFETY-10 section 1). Each
         # battery's mode and on/off switch live in the store, not in options,
         # so a change applies at once (`pack_use`, `pack_mode` actions).
-        self.modes_on = opts.get("battery_modes_enabled") is True
+        self.modes_on = packmode.enabled(opts)
         # The units whose mode fault is logged already (edge-logged).
         self._mode_faults = set()
+        # OPT-11 P2: (topic, reason) of MQTT commands refused and logged once.
+        self._mqtt_refused = set()
+        self._mqtt_refused_more = 0
         # [(pack name, why)] for packs that are due and waiting their turn.
         # Empty at a roster of one, where nothing can defer anything.
         self._maint_deferred = []
@@ -1823,6 +1831,11 @@ class Controller:
         """
         return float(sum(inv.rating_w() for inv in self.inverters))
 
+    def peak_aim_w(self) -> float:
+        """SAFETY-79. `peak_cap_w` less `peak_margin_w`: what the forecast
+        checks aim at. Reports and the bill use the real cap."""
+        return float(self.o.get("peak_cap_w", 3500)) - self.peak_margin_w
+
     def site_dev_bound(self, which: str):
         """The roster's summed device floor / ceiling, or None if unknown.
 
@@ -2053,6 +2066,65 @@ class Controller:
         pm = self.pack_mode(inv)
         return {"unit": inv.name, "use": pm.use, "mode": pm.mode}
 
+    def mode_values(self) -> dict:
+        """OPT-11 P2: each battery's mode and switch, for their HA state
+        topics. Read from the store, so they are right after a restart. Empty
+        while battery modes are off."""
+        if not self.modes_on:
+            return {}
+        out = {}
+        names = [inv.name for inv in self.inverters]
+        for inv, slug in zip(self.inverters, unit_slugs(names)):
+            pm = self.pack_mode(inv)
+            out[unit_state_key(slug, "mode")] = MODE_LABELS.get(pm.mode,
+                                                               pm.mode)
+            out[unit_state_key(slug, "use")] = USE_ON if pm.use else USE_OFF
+        return out
+
+    async def mqtt_command(self, topic, payload, retained) -> None:
+        """OPT-11 P2: one command from a battery's HA switch or mode list.
+
+        It goes through handle_action(), the page's /action path, so the
+        checks and the store write are the same. A retained command (one the
+        broker replays on subscribe, e.g. after a restart) is never applied:
+        the store already holds the last choice. A refusal is logged once per
+        topic and reason, at most 32, and further refusals are counted in the
+        log at 1, 10, 100 and so on. Never raises: MQTT must not reach control.
+        """
+        try:
+            slug, action, value, why = parse_command(
+                topic, payload, retained, packmode.MODES)
+            if why is None:
+                names = [inv.name for inv in self.inverters]
+                unit = dict(zip(unit_slugs(names), names)).get(slug)
+                result = await self.handle_action(action, unit=unit,
+                                                  value=value)
+                if not (isinstance(result, dict) and "error" in result):
+                    self.mqtt.publish(self.mode_values())
+                    return
+                why = result["error"]
+            else:
+                why = messages.pack_refused(why)
+            # Printable only: a topic is sender-chosen, so no log injection.
+            topic_s = "".join(c if c.isprintable() else "?"
+                              for c in str(topic)[:128])
+            _LOG.debug("Home Assistant command %s %s refused: %s", topic_s,
+                       repr(payload)[:48], why)
+            if (topic_s, why) in self._mqtt_refused \
+                    or len(self._mqtt_refused) >= 32:
+                # Not logged again, but counted: 1, 10, 100... stay visible.
+                self._mqtt_refused_more += 1
+                more = self._mqtt_refused_more
+                if str(more).rstrip("0") == "1":
+                    _LOG.warning("Home Assistant commands still ignored: %d "
+                                 "more so far.", more)
+                return
+            self._mqtt_refused.add((topic_s, why))
+            _LOG.warning("Home Assistant command on %s ignored: %s", topic_s,
+                         why)
+        except Exception:                               # noqa: BLE001
+            _LOG.exception("Home Assistant command could not be handled")
+
     def _maint_stop_off(self, machine, now) -> None:
         """OPT-11: stop this battery's maintenance cycle because it is off."""
         if machine.phase == IDLE:
@@ -3390,6 +3462,7 @@ class Controller:
             # fraction of the discharge already happening.
             batt_w=self.site_batt_w,
             cap_w=float(self.o.get("peak_cap_w", 3500)),
+            margin_w=self.peak_margin_w,
             reading=self.peak.peak(now),
             # ⚠️ SUMMED FROM THE PER-UNIT CONJUNCTION, WHICH IS AC 4. Each
             # unit contributes `min(its max_w, |its own device bounds|)`, so
@@ -3650,8 +3723,7 @@ class Controller:
                 # charge together, this flag would have to become per-pack - and
                 # so would the reasoning above it.
                 if peakguard.shaping_suspended(
-                        self.peak_guard, self.peak_fc,
-                        float(self.o.get("peak_cap_w", 3500))):
+                        self.peak_guard, self.peak_fc, self.peak_aim_w()):
                     # Edge-logged: level-logging here would write one line per
                     # cycle for as long as the event lasts and flush the 40-line
                     # action log of the very context that explains it.
@@ -3678,7 +3750,7 @@ class Controller:
                         # rate the pack in front of us is charging at.
                         floor = maintenance_charge_floor(
                             self.maint.cfg.charge_w,
-                            self.peak_fc, float(self.o.get("peak_cap_w", 3500)))
+                            self.peak_fc, self.peak_aim_w())
                         if floor > 0:
                             hi, reason = -floor, "cheap-window charge"
                     # PER-INVERTER, like the SET above and for the same reason -
@@ -5144,6 +5216,8 @@ class Controller:
             values[unit_state_key(slug, "soc")] = inv.soc
             values[unit_state_key(slug, "sign_latch")] = (
                 "tripped" if inv.sign_wd.tripped else "clear")
+        # OPT-11 P2: each battery's mode and switch; nothing while modes are off.
+        values.update(self.mode_values())
         self.mqtt.publish(values)
 
     async def shutdown(self) -> None:
@@ -6059,6 +6133,21 @@ class ControlTaskDied(RuntimeError):
     """The control task ended while nothing had asked it to stop."""
 
 
+def mode_announcement(opts, store) -> tuple:
+    """OPT-11 P2: `(modes, retract)` for the MQTT publisher at start-up.
+
+    The config topics announced are kept in the store, so turning modes off
+    removes exactly those (`mqtt.mode_configs()`). An install that never
+    turned modes on gets nothing, and its store is not written.
+    """
+    modes = packmode.MODES if packmode.enabled(opts) else ()
+    stored = store.data.get(MODE_CONFIGS_KEY)
+    keep, retract = mode_configs(roster_names(opts), modes, stored)
+    if keep and keep != stored:
+        store.set(MODE_CONFIGS_KEY, keep)
+    return modes, retract
+
+
 def control_task_done(task: asyncio.Task, *, controller, stop: asyncio.Event,
                       died: list) -> None:
     """Done-callback on the control task: log its death now and wake amain()
@@ -6189,6 +6278,9 @@ async def amain() -> None:
         # took the control loop down with it.
         broker = None
         try:
+            # OPT-11 P2: each battery's switch and mode list, only while
+            # battery modes are on (`mode_announcement()`).
+            _modes, _retract = mode_announcement(opts, store)
             broker = await hass.mqtt_service()
             pub = MqttPublisher(
                 broker.get("host") if broker else None,
@@ -6208,6 +6300,8 @@ async def amain() -> None:
                 # is what `build_roster()` itself reads. At a roster of one this
                 # announces nothing new - see `announced_units()`.
                 units=roster_names(opts),
+                modes=_modes,
+                retract=_retract,
             )
         except Exception as err:  # noqa: BLE001
             _LOG.warning("MQTT unavailable (%s) - continuing without status entities", err)
@@ -6221,6 +6315,8 @@ async def amain() -> None:
 
         stop = asyncio.Event()
         controller = Controller(opts, hass, store, pub, stop=stop)
+        if callable(getattr(pub, "bind", None)):
+            pub.bind(asyncio.get_running_loop(), controller.mqtt_command)
         runner = await web.start(controller, port=INGRESS_PORT)
 
         loop = asyncio.get_running_loop()

goodwe_controller/app/messages.py

@@ -932,6 +932,11 @@ PACK_REFUSED = {
     "off": "Battery modes are off. Turn on “Battery modes” first.",
     "unit": "There is no battery with that name.",
     "value": "That choice is not possible for this battery.",
+    # OPT-11 P2: a Home Assistant command dropped before it reaches a battery.
+    "retained": "An old saved command is not used.",
+    "size": "That command is too long.",
+    "encoding": "That command is not plain text.",
+    "topic": "That is not a battery command.",
 }
 
 
@@ -1152,6 +1157,7 @@ LABELS = {
     "integrator_max_w": "Catch-up limit",
     "maintenance_discharge_w": "Draw-down rate",
     "peak_cap_w": "Peak limit",
+    "peak_margin_w": "Margin below the peak limit",
     "peak_forecast_entity": "15-minute demand forecast entity",
     "meter_entity": "Grid power entity",
     "soc_entity": "Battery level entity",

goodwe_controller/app/mqtt.py

@@ -7,8 +7,10 @@ That separation is deliberate; observability must never be able to take down
 control.
 """
 
+import asyncio
 import json
 import logging
+import re
 
 try:
     import paho.mqtt.client as mqtt
@@ -375,7 +377,26 @@ def _payload(name, object_id, unique_id, state_topic, device,
     return cfg
 
 
-def discovery_payloads(units=(), omit=(), include=()) -> list:
+def unit_unique_id(slug: str, key: str) -> str:
+    """One unit's entity `unique_id` (also its config topic's middle part)."""
+    return f"{BASE}_unit_{slug}_{key}"
+
+
+def _unit_places(names):
+    """`(slug, device, prefix)` for each roster name, in order."""
+    slugs = unit_slugs(names)
+    for i, (name, slug) in enumerate(zip(names, slugs)):
+        # ⚠️ `i == 0` IS THE ASYMMETRY, IN ONE LINE. The first roster entry
+        # keeps the device it already has and carries its name in the ENTITY
+        # name instead; every later one gets its own device and does not.
+        # Either way HA composes device-name + entity-name into the same
+        # string, which is why `object_id` can be written once for both.
+        device = DEVICE if i == 0 else unit_device(name, slug)
+        prefix = f"{_title(name)} " if i == 0 else ""
+        yield slug, device, prefix
+
+
+def discovery_payloads(units=(), omit=(), include=(), modes=()) -> list:
     """Every `(topic, payload)` this add-on announces, in announcement order.
 
     Split out of `_announce()` so that the acceptance criterion can be graded as
@@ -389,6 +410,8 @@ def discovery_payloads(units=(), omit=(), include=()) -> list:
     the output here (`_payload()`'s note, and test_mqtt.py part 1 asserts an
     ordered LIST precisely because a re-order is a zero-byte edit that
     re-announces every entity on a live install).
+
+    `modes` (OPT-11 P2) is the battery modes offered, empty when they are off.
     """
     omit = set(omit)
     include = [k for k in include if k not in omit]
@@ -400,19 +423,10 @@ def discovery_payloads(units=(), omit=(), include=()) -> list:
             f"homeassistant/sensor/{BASE}_{key}/config",
             _payload(name, object_id, f"{BASE}_{key}", f"{BASE}/{key}",
                      DEVICE, unit, dev_class, state_class, icon)))
-    names = announced_units(units)
-    slugs = unit_slugs(names)
     site_slug = _slug(DEVICE["name"])
-    for i, (name, slug) in enumerate(zip(names, slugs)):
-        # ⚠️ `i == 0` IS THE ASYMMETRY, IN ONE LINE. The first roster entry
-        # keeps the device it already has and carries its name in the ENTITY
-        # name instead; every later one gets its own device and does not.
-        # Either way HA composes device-name + entity-name into the same
-        # string, which is why `object_id` below can be written once for both.
-        device = DEVICE if i == 0 else unit_device(name, slug)
-        prefix = f"{_title(name)} " if i == 0 else ""
+    for slug, device, prefix in _unit_places(announced_units(units)):
         for key, oid, label, unit, dev_class, state_class, icon in UNIT_SENSORS:
-            unique_id = f"{BASE}_unit_{slug}_{key}"
+            unique_id = unit_unique_id(slug, key)
             out.append((
                 f"homeassistant/sensor/{unique_id}/config",
                 _payload(prefix + label,
@@ -427,12 +441,134 @@ def discovery_payloads(units=(), omit=(), include=()) -> list:
             f"homeassistant/sensor/{BASE}_{key}/config",
             _payload(name, object_id, f"{BASE}_{key}", f"{BASE}/{key}",
                      DEVICE, unit, dev_class, state_class, icon)))
+    # OPT-11 P2: last, and only when battery modes are on.
+    out.extend(mode_payloads(units, modes))
     return out
 
 
+# --- OPT-11 P2: each battery's switch and mode, as HA entities ---------------
+# Announced only when battery modes are on, after every other row, so no
+# existing row moves. A command goes through Controller.handle_action(), the
+# page's /action path. DECISIONS/OPT-11-battery-modes.md has the record.
+#
+# The select's options are these labels, in `packmode.MODES` order. A new mode
+# (P3) is a new entry here and a new option at the end; no id changes.
+MODE_LABELS = {"solar_first": "Solar first", "leftover_solar": "Leftover solar"}
+USE_ON, USE_OFF = "ON", "OFF"
+# A longer command payload is refused before it is decoded.
+MAX_COMMAND_BYTES = 64
+# The store key holding the mode config topics last announced, so that turning
+# modes off can remove exactly those (an empty retained config) and an install
+# that never turned them on publishes nothing new.
+MODE_CONFIGS_KEY = "mqtt_mode_configs"
+_MODE_CONFIG_RE = re.compile(
+    rf"homeassistant/(select|switch)/{BASE}_unit_[a-z0-9_]+_(mode|use)/config")
+_COMMAND_RE = re.compile(rf"{BASE}/unit/([a-z0-9_]+)/(mode|use)/set")
+# (entity key, HA component, entity name, icon)
+MODE_ENTITIES = [
+    ("mode", "select", "Mode", "mdi:battery-sync"),
+    ("use", "switch", "Use", "mdi:power"),
+]
+
+
+def mode_payloads(units, modes) -> list:
+    """The `(topic, payload)` for each battery's select and switch.
+
+    Every roster entry gets them, also at a roster of one. The device rule is
+    TEL-07's: the first battery's rows go on the site device with its name in
+    front, each later battery's on its own device. Empty when `modes` is empty.
+    """
+    if not modes:
+        return []
+    site_slug = _slug(DEVICE["name"])
+    options = [MODE_LABELS[m] for m in modes]
+    out = []
+    for slug, device, prefix in _unit_places([str(u) for u in units]):
+        for key, component, label, icon in MODE_ENTITIES:
+            unique_id = unit_unique_id(slug, key)
+            state = f"{BASE}/{unit_state_key(slug, key)}"
+            cfg = {
+                "name": prefix + label,
+                "object_id": f"{site_slug}_{slug}_{key}",
+                "unique_id": unique_id,
+                "state_topic": state,
+                "command_topic": f"{state}/set",
+            }
+            if component == "select":
+                cfg["options"] = options
+            else:
+                cfg["payload_on"] = USE_ON
+                cfg["payload_off"] = USE_OFF
+            # HA sends commands unretained; a retained one is ignored anyway.
+            cfg["retain"] = False
+            cfg["availability_topic"] = AVAILABILITY
+            cfg["device"] = device
+            cfg["icon"] = icon
+            out.append((f"homeassistant/{component}/{unique_id}/config", cfg))
+    return out
+
+
+def mode_retractions(stored, current) -> list:
+    """Config topics announced before and not now: each gets an empty
+    retained config, which removes the entity from HA. Anything in the store
+    that is not one of our mode config topics is ignored."""
+    if not isinstance(stored, list):
+        return []
+    keep = set(current)
+    return list(dict.fromkeys(t for t in stored if isinstance(t, str)
+                              and _MODE_CONFIG_RE.fullmatch(t)
+                              and t not in keep))
+
+
+def mode_configs(units, modes, stored) -> tuple:
+    """`(keep, retract)` at start-up: the mode config topics to store, and the
+    ones to remove now. A removal stays in `keep`, so a start with no broker
+    does not forget it; resending an empty config is harmless."""
+    current = [t for t, _ in mode_payloads(units, modes)]
+    retract = mode_retractions(stored, current)
+    return current + retract, retract
+
+
+def parse_command(topic, payload, retained, modes):
+    """`(slug, action, value, reject)` for one message on a command topic.
+
+    `action` is `pack_use` or `pack_mode`. `value` is what /action would be
+    sent: a bool for the switch, a mode key for the select, or None for a
+    payload that is neither (the store path then refuses it). `reject` is a
+    reason when the message is dropped here: replayed by the broker (retained),
+    too long, not UTF-8, or not a command topic.
+    """
+    m = _COMMAND_RE.fullmatch(topic) if isinstance(topic, str) else None
+    if m is None:
+        return None, None, None, "topic"
+    slug, key = m.group(1), m.group(2)
+    action = "pack_use" if key == "use" else "pack_mode"
+    if retained:
+        return slug, action, None, "retained"
+    if not isinstance(payload, (bytes, bytearray)) \
+            or len(payload) > MAX_COMMAND_BYTES:
+        return slug, action, None, "size"
+    try:
+        text = bytes(payload).decode("utf-8")
+    except UnicodeDecodeError:
+        return slug, action, None, "encoding"
+    if key == "use":
+        value = {USE_ON: True, USE_OFF: False}.get(text)
+    else:
+        value = next((k for k in modes if MODE_LABELS.get(k) == text), None)
+    return slug, action, value, None
+
+
 class MqttPublisher:
+    # OPT-11 P2. Class defaults, so a publisher built without them (and every
+    # test fixture that builds one with __new__) behaves as before.
+    modes = ()
+    retract = ()
+    _loop = None
+    _handler = None
+
     def __init__(self, host, port, username=None, password=None, omit=(),
-                 units=(), include=()):
+                 units=(), include=(), modes=(), retract=()):
         # `omit` drops sensor keys from discovery entirely. ⚠️ Announcing a
         # sensor that nothing will ever publish to is not harmless here:
         # p1_sample_age_s is a watchdog input, and an entity that exists but is
@@ -448,6 +584,10 @@ class MqttPublisher:
         # second reading of the option. Defaults to empty, so a caller that has
         # not been taught about it announces exactly what it announced before.
         self.include = [str(k) for k in include]
+        # OPT-11 P2: the modes offered (empty when battery modes are off), and
+        # the mode config topics to remove (`mode_retractions()`).
+        self.modes = tuple(modes)
+        self.retract = [str(t) for t in retract]
         self.enabled = mqtt is not None and bool(host)
         self.client = None
         if not self.enabled:
@@ -470,7 +610,8 @@ class MqttPublisher:
         # paho DROPS QoS-0 publishes when disconnected, silently. The result is
         # an add-on that logs "MQTT connected" and creates no entities at all.
         # As a bonus, this also re-announces after every reconnect.
-        self.client.on_connect = lambda *_args, **_kw: self._announce()
+        self.client.on_connect = lambda *_args, **_kw: self._on_connect()
+        self.client.on_message = self._on_message
         try:
             self.client.connect(host, int(port), keepalive=60)
             self.client.loop_start()
@@ -485,10 +626,35 @@ class MqttPublisher:
         # diffing what actually goes on the wire - and a payload built here
         # instead would be one this add-on publishes and no check has ever seen.
         for topic, cfg in discovery_payloads(self.units, self.omit,
-                                             self.include):
+                                             self.include, self.modes):
             self.client.publish(topic, json.dumps(cfg), retain=True)
+        for topic in self.retract:
+            self.client.publish(topic, "", retain=True)
         self.client.publish(AVAILABILITY, "online", retain=True)
 
+    def _on_connect(self) -> None:
+        self._announce()
+        if self.modes:
+            for key in ("mode", "use"):
+                self.client.subscribe(f"{BASE}/unit/+/{key}/set")
+
+    def bind(self, loop, handler) -> None:
+        """OPT-11 P2: hand each command to `handler(topic, payload, retained)`,
+        a coroutine function run on `loop` (the controller's)."""
+        self._loop, self._handler = loop, handler
+
+    def _on_message(self, _client, _userdata, msg) -> None:
+        # paho's network thread: nothing is done here but the handoff.
+        loop, handler = self._loop, self._handler
+        if loop is None or handler is None:
+            _LOG.debug("MQTT command before start-up ignored: %s", msg.topic)
+            return
+        try:
+            asyncio.run_coroutine_threadsafe(
+                handler(msg.topic, bytes(msg.payload), bool(msg.retain)), loop)
+        except RuntimeError as err:  # the loop has closed
+            _LOG.debug("MQTT command dropped: %s", err)
+
     def publish(self, values: dict) -> None:
         if not self.enabled or self.client is None:
             return

goodwe_controller/app/packmode.py

@@ -42,6 +42,12 @@ MODE_KEY = "pack_mode"
 SELL_KEY = "pack_sell"              # reserved for P3
 
 
+def enabled(opts) -> bool:
+    """Battery modes are on only when the option is exactly true. One answer,
+    read by the controller and by the MQTT announcement."""
+    return opts.get("battery_modes_enabled") is True
+
+
 @dataclass(frozen=True)
 class PackMode:
     """One battery's choice: on or off, its mode, and (P3) selling."""

goodwe_controller/app/peakguard.py

@@ -143,6 +143,10 @@ because the money is spent on `max(month's peak so far, this quarter)`:
 
     ceiling_w = max(peak_cap_w, the month's banked peak)
 
+SAFETY-79: the guard AIMS `peak_margin_w` under that ceiling
+(`quarter_target_w()`), so a quarter that lands a little high still stays under
+it. `ceiling_w` stays the real limit for reports and the bill.
+
 Once a month's peak already stands at 5013 W, holding every later quarter to
 3500 W buys nothing - the bill for that month is already 5013 W - while the
 holding itself costs real charging the site wanted. Raising the ceiling to a peak
@@ -453,7 +457,13 @@ PEAK_OPTION_DEFAULTS = {
     # and it is consistent with DoD 90 on a full pack. It exists so validate.py
     # rule 8 can refuse a `peak_soc_floor` at or below it - see that rule.
     "bms_soc_floor": 10,
+    # SAFETY-79. Watts the guard aims under `peak_cap_w`, so a quarter that
+    # lands a little high still stays under the cap. Owner decision 2026-09-28.
+    "peak_margin_w": 50,
 }
+# SAFETY-79. config.yaml's schema range for `peak_margin_w`, mirrored so
+# validate.py can refuse it in words; test_peakguard.py pins it to the file.
+PEAK_MARGIN_RANGE_W = (0, 500)
 
 
 @dataclass(frozen=True)
@@ -474,6 +484,7 @@ class PeakDecision:
     projected_w: float | None = None
     excess_w: float = 0.0
     wanted_w: float | None = None    # the floor before `achievable_w` clamped it
+    target_w: float | None = None    # SAFETY-79: what the guard aims at
     saturated: bool = False
     # SAFETY-24. True when a forced discharge WAS wanted and the EMS-side SoC
     # floor refused to size it. Mutually exclusive with `saturated` by
@@ -524,6 +535,15 @@ def quarter_ceiling_w(cap_w, reading) -> tuple[float, str]:
     return cap, CEILING_CAP
 
 
+def quarter_target_w(cap_w, margin_w, reading) -> float:
+    """SAFETY-79. The average the guard AIMS at: `margin_w` under
+    `quarter_ceiling_w()`, so under a month peak already banked as well as
+    under the cap. Aiming AT a banked peak can bank a few W over it, which
+    raises the bill as surely as going over the cap. Reports and the bill keep
+    the real limit (`ceiling_w`)."""
+    return quarter_ceiling_w(cap_w, reading)[0] - float(margin_w)
+
+
 def excess_w(projection, ceiling_w) -> float:
     """Watts that must come off the meter NOW to land the quarter on the ceiling.
 
@@ -713,8 +733,8 @@ def binding_soc_floor(packs, soc_floor):
     return max(floors)
 
 
-def decide(projection, *, batt_w, cap_w, reading, achievable_w, soc, soc_floor,
-           max_hold_s=None, frozen: bool = False) -> PeakDecision:
+def decide(projection, *, batt_w, cap_w, margin_w, reading, achievable_w, soc,
+           soc_floor, max_hold_s=None, frozen: bool = False) -> PeakDecision:
     """The whole guard. Pure: no clock, no I/O, no state carried between calls.
 
     `batt_w`       MEASURED battery power, positive = discharge. None when the
@@ -732,8 +752,13 @@ def decide(projection, *, batt_w, cap_w, reading, achievable_w, soc, soc_floor,
                    the pack for money. REQUIRED, and deliberately without a
                    default: a default here would be CFG-03's fourth defaults
                    site and TEST-14's quiet fallback in one line.
+    `margin_w`     SAFETY-79. Watts to aim under `cap_w` (`peak_margin_w`).
+                   REQUIRED for the same reason. `ceiling_w` stays the real
+                   limit for every report; only `excess_w` uses the target.
     """
     ceiling, from_where = quarter_ceiling_w(cap_w, reading)
+    target = quarter_target_w(cap_w, margin_w, reading)
+    aim = "" if target == ceiling else f", aiming at {target:.0f} W"
     ceil_top = abs(float(achievable_w))
 
     blind = blindness(projection, max_hold_s, frozen)
@@ -743,7 +768,7 @@ def decide(projection, *, batt_w, cap_w, reading, achievable_w, soc, soc_floor,
         # see", which is one-directional and cannot cost pack energy.
         return PeakDecision(
             engaged=True, floor_w=0.0, ceiling_w=ceiling, ceiling_from=from_where,
-            blind=blind, projected_w=None,
+            target_w=target, blind=blind, projected_w=None,
             reason=(f"PEAK_GUARD_BLIND: {blind}. The battery may not charge "
                     f"while this 15-minute period cannot be measured, so it "
                     f"is asked for no charging, floor 0 W"))
@@ -755,7 +780,7 @@ def decide(projection, *, batt_w, cap_w, reading, achievable_w, soc, soc_floor,
         # measurement that is not there is a made-up number.
         return PeakDecision(
             engaged=False, floor_w=None, ceiling_w=ceiling,
-            ceiling_from=from_where, projected_w=projected,
+            ceiling_from=from_where, target_w=target, projected_w=projected,
             reason="no battery measurement to size a floor from")
     if QUARTER_S - float(projection.elapsed_s) <= 0.0:
         # A block with no seconds left cannot be influenced. This is degenerate
@@ -763,10 +788,10 @@ def decide(projection, *, batt_w, cap_w, reading, achievable_w, soc, soc_floor,
         # the battery's own current output would be a claim that says nothing.
         return PeakDecision(
             engaged=False, floor_w=None, ceiling_w=ceiling,
-            ceiling_from=from_where, projected_w=projected,
+            ceiling_from=from_where, target_w=target, projected_w=projected,
             reason="the quarter has no seconds left to influence")
 
-    ex = excess_w(projection, ceiling)
+    ex = excess_w(projection, target)
     wanted = float(batt_w) + ex
     if wanted <= 0.0:
         # ⚠️ THE TEST IS ON THE FLOOR, NOT ON `excess > 0`, AND THE REPLAY IS WHY.
@@ -787,7 +812,8 @@ def decide(projection, *, batt_w, cap_w, reading, achievable_w, soc, soc_floor,
         # the quantity being thresholded is simply the right one.
         return PeakDecision(
             engaged=False, floor_w=None, ceiling_w=ceiling,
-            ceiling_from=from_where, projected_w=projected, excess_w=ex,
+            ceiling_from=from_where, target_w=target, projected_w=projected,
+            excess_w=ex,
             wanted_w=wanted,
             reason="the load is under the level this quarter can afford")
 
@@ -805,7 +831,8 @@ def decide(projection, *, batt_w, cap_w, reading, achievable_w, soc, soc_floor,
         # not the grid-following loop, which still has no SoC guard.
         return PeakDecision(
             engaged=True, floor_w=0.0, ceiling_w=ceiling,
-            ceiling_from=from_where, projected_w=projected, excess_w=ex,
+            ceiling_from=from_where, target_w=target, projected_w=projected,
+            excess_w=ex,
             wanted_w=wanted, floor_limited=True, soc=soc, soc_floor=soc_floor,
             soc_floor_reason=denied,
             # UX-10: plain sentences, every figure kept (test_messages.py
@@ -813,8 +840,8 @@ def decide(projection, *, batt_w, cap_w, reading, achievable_w, soc, soc_floor,
             reason=(
                 f"PEAK_GUARD_FLOOR_LIMITED: this 15-minute period heads for "
                 f"{projected:.0f} W against a {ceiling:.0f} W limit "
-                f"({from_where}), and {wanted:+.0f} W from the battery would "
-                f"hold it. But {denied}"
+                f"({from_where}){aim}, and {wanted:+.0f} W from the battery "
+                f"would hold it. But {denied}"
                 + ("" if soc is None else f", at {float(soc):.0f} %")
                 + ("" if soc is None or soc_floor is None
                    else f" against {float(soc_floor):.0f} %")
@@ -825,11 +852,11 @@ def decide(projection, *, batt_w, cap_w, reading, achievable_w, soc, soc_floor,
     saturated = floor < wanted
     return PeakDecision(
         engaged=True, floor_w=floor, ceiling_w=ceiling, ceiling_from=from_where,
-        blind=None, projected_w=projected, excess_w=ex, wanted_w=wanted,
-        saturated=saturated, soc=soc, soc_floor=soc_floor,
+        target_w=target, blind=None, projected_w=projected, excess_w=ex,
+        wanted_w=wanted, saturated=saturated, soc=soc, soc_floor=soc_floor,
         reason=(
             f"PEAK_GUARD: this 15-minute period heads for {projected:.0f} W "
-            f"against a {ceiling:.0f} W limit ({from_where}), with "
+            f"against a {ceiling:.0f} W limit ({from_where}){aim}, with "
             f"{QUARTER_S - projection.elapsed_s:.0f} s left, so {ex:.0f} W must "
             f"come off the meter. The battery measured {float(batt_w):+.0f} W "
             f"and is asked for {floor:+.0f} W"

goodwe_controller/app/ui_text/en.json

@@ -13,7 +13,7 @@
     "inverters.modes": {"name": "Battery modes"},
     "tariff.selling": {"name": "Charging and selling on price"},
     "tariff.feed": {"name": "Price feed details"},
-    "safety.peak": {"name": "Peak limit"},
+    "safety.peak": {"name": "Peak limit", "description": "The add-on aims a little below your limit, to be safe."},
     "safety.start": {"name": "Starting and stopping"},
     "safety.stops": {"name": "Safety stops", "description": "Measured on real hardware, so leave them unchanged."},
     "maintenance.support": {"name": "Support"}

goodwe_controller/app/ui_text/nl.json

@@ -13,7 +13,7 @@
     "inverters.modes": {"name": "Batterijmodi"},
     "tariff.selling": {"name": "Laden en verkopen op prijs"},
     "tariff.feed": {"name": "Details van de prijsfeed"},
-    "safety.peak": {"name": "Piekgrens"},
+    "safety.peak": {"name": "Piekgrens", "description": "De add-on mikt iets onder uw grens, voor de zekerheid."},
     "safety.start": {"name": "Starten en stoppen"},
     "safety.stops": {"name": "Veiligheidsstops", "description": "Gemeten op echte hardware, dus laat ze zoals ze zijn."},
     "maintenance.support": {"name": "Support"}

goodwe_controller/app/uilayout.py

@@ -182,6 +182,7 @@ GROUPS = (
         Part("peak", (
             Field("peak_cap_w", SLIDER, "kW", scale=1000, step=100,
                   simple=True, wizard=5, note="month_peak"),
+            Field("peak_margin_w", SLIDER, "W", step=10),   # SAFETY-79
             Field("peak_forecast_entity", ENTITY, kind="power_w", wizard=5),
             Field("peak_soc_floor", LOCKED, "%"),
         )),

goodwe_controller/app/validate.py

@@ -243,6 +243,9 @@ NOT READY banner already names the problem, (b) does not apply.
      note below on `gain`, which is the term that owns that and which SAFETY-39
      measured and did not change.
 
+ 12. `peak_margin_w` outside 0-500 W, not whole, or not below `peak_cap_w`.
+     SAFETY-79; the Supervisor checks the range too, but not for sim/inject.py.
+
 WHAT IS DELIBERATELY *NOT* REFUSED, because it is untidy rather than unsafe:
 
   - `integrator_max_w` above `max_w`. DOCS.md is right that the only thing the
@@ -333,7 +336,7 @@ from .inverter import roster_entries
 from .maintenance import MAINT_OPTION_DEFAULTS
 from .p1 import (
     SOURCE_HA, SOURCE_HA_SIGNED, is_enabled, validate_meter_options)
-from .peakguard import PEAK_OPTION_DEFAULTS
+from .peakguard import PEAK_MARGIN_RANGE_W, PEAK_OPTION_DEFAULTS
 from .price import (
     PRICE_OPTION_DEFAULTS, PROVIDERS, RETIRED_PROVIDERS,
     SUPERSEDED_TAXES_DEFAULTS_EUR_KWH)
@@ -1194,6 +1197,9 @@ def validate_startup_config(opts: dict, dev_min=None, dev_max=None,
                 f"reserve above {bms_floor:.0f} %, or correct the cut-out "
                 "level.")
 
+    # --- 12. SAFETY-79: the margin the peak guard keeps under the limit -----
+    problems += _margin_problems(opts)
+
     # --- warnings: real, but not worth refusing over -------------------------
     integ = opts.get("integrator_max_w")
     integ_w = None if not integ else _number(opts, "integrator_max_w", None)
@@ -2022,6 +2028,23 @@ def _price_check(notes, opts, running, prices_known) -> None:
                 "Prices are fetched after the restart, and Home shows them.")
 
 
+def _margin_problems(opts: dict) -> list[str]:
+    """Rule 12: `peak_margin_w` is a whole number in the schema's range, and
+    below the peak limit, which would leave nothing to aim at."""
+    raw = opts.get("peak_margin_w")
+    margin = _number(opts, "peak_margin_w",
+                     PEAK_OPTION_DEFAULTS["peak_margin_w"])
+    lo, hi = PEAK_MARGIN_RANGE_W
+    if margin is None or margin != int(margin) or not lo <= margin <= hi:
+        return [f"{S('peak_margin_w')} must be a whole number from {lo} to "
+                f"{hi} W. It is {messages.shown(raw)}."]
+    cap = _number(opts, "peak_cap_w", None)
+    if cap is not None and margin >= cap:
+        return [f"{S('peak_margin_w')} is {margin:.0f} W, not below the "
+                f"{S('peak_cap_w')} of {cap:.0f} W. Lower the margin."]
+    return []
+
+
 def _number(opts: dict, key, default):
     raw = opts.get(key, default)
     if raw is None or raw == "":

goodwe_controller/config.yaml

@@ -290,6 +290,8 @@ options:
   # protects the shallower pack with the deeper pack's number.
   bms_soc_floor: 10
   peak_cap_w: 3500
+  # SAFETY-79: the peak guard aims this many watts under peak_cap_w.
+  peak_margin_w: 50
   peak_forecast_entity: ""
 
   # The peak guard can COMMAND a discharge to hold a billed quarter under
@@ -1019,6 +1021,7 @@ schema:
   max_w: int(100,5000)
   bms_soc_floor: int(0,50)
   peak_cap_w: int(500,15000)
+  peak_margin_w: int(0,500)
   peak_forecast_entity: str?
 
   # SAFETY-24. No "off" value: a guard that can command a discharge does not get

goodwe_controller/test_arbitrage.py

@@ -1039,7 +1039,7 @@ def s45_loop(*, sized: bool, cap_w=S45_CAP_W, load_w=S45_LOAD_W,
             now = t0 + timedelta(seconds=t)
             claims = []
             pk = _pg.decide(PeakTracker.project(avg, now), batt_w=batt,
-                            cap_w=cap_w, reading=None, achievable_w=MAX_W,
+                            cap_w=cap_w, margin_w=0.0, reading=None, achievable_w=MAX_W,
                             soc=50.0, soc_floor=20.0, max_hold_s=900.0,
                             frozen=False)
             pc = _pg.claim(pk)
@@ -1380,7 +1380,7 @@ def _s47_headroom(*, meter_w, batt_w, elapsed_s=450.0):
         t += 5.0
     pk = _pg.decide(
         PeakTracker.project(avg, S45_T0 + timedelta(seconds=elapsed_s)),
-        batt_w=batt_w, cap_w=S45_CAP_W, reading=None, achievable_w=MAX_W,
+        batt_w=batt_w, cap_w=S45_CAP_W, margin_w=0.0, reading=None, achievable_w=MAX_W,
         soc=50.0, soc_floor=20.0, max_hold_s=900.0, frozen=False)
     return _pg.charge_headroom_w(pk)
 

goodwe_controller/test_messages.py

@@ -845,7 +845,8 @@ DATA = {
     "main.py:'cancelled' if task.cancelled() else repr(exc) if exc else 'returned'",
     "main.py:EXIT_CONTROL_DIED", "main.py:first.name", "main.py:OPTIONS_PATH",
     "main.py:err", "main.py:', '.join(sorted(stranded))",
-    "main.py:sorted(on_roster)", "main.py:inv.name",
+    "main.py:sorted(on_roster)", "main.py:inv.name", "main.py:topic_s",
+    "main.py:more",
     "main.py:inv.batt_fresh.bound.release_s", "main.py:inv.batt_fresh.bound.move_w",
     "main.py:inv.batt_fresh.bound.unchanged_s", "main.py:inv.batt_fresh.held",
     "main.py:value", "main.py:float(value)",
@@ -964,7 +965,7 @@ def _peak_texts():
     proj = Projection(projected_w=4338.0, elapsed_w=4338.0, elapsed_s=799.0,
                       covered_frac=1.0, complete=False, block_start=_T0,
                       held_w=0.0)
-    kw = dict(cap_w=3500.0, reading=None, soc_floor=30.0, max_hold_s=30.0)
+    kw = dict(cap_w=3500.0, margin_w=0.0, reading=None, soc_floor=30.0, max_hold_s=30.0)
     sat = _pg.decide(proj, batt_w=1500.0, achievable_w=5000.0, soc=80.0, **kw)
     low = _pg.decide(proj, batt_w=1500.0, achievable_w=5000.0, soc=20.0, **kw)
     unread = _pg.decide(proj, batt_w=1500.0, achievable_w=5000.0, soc=None, **kw)
@@ -1058,6 +1059,8 @@ TEXT = {
     "main.py:rep.explain()": _migration_texts,
     "store.py:rep.explain()": _migration_texts,
     "main.py:plan.words": _tariff_texts,
+    # OPT-11 P2: why an MQTT command was ignored.
+    "main.py:why": lambda: list(messages.PACK_REFUSED.values()),
     # Families: every builder of the text, as a template.
     "main.py:refusal": lambda: (_static(("main.py", "return", "write_refusal"))
                                 + _static(("allocator.py", "return", "authorises"))),

goodwe_controller/test_mqtt.py

@@ -42,6 +42,7 @@ part is a diff against it, reported as `N topics, M missing, C changed, A added`
           re-executed from mutated source, each one named with the property it
           breaks. A property nothing breaks is a property nothing is watching.
   Part 9  What these checks CANNOT see, enumerated and sized.
+  Part 10 OPT-11 P2: each battery's select and switch, commands, removal.
 
 ⚠️ THE FIXTURE GIVES A AND B DIFFERENT NAMES, DIFFERENT RATINGS, DIFFERENT SoC,
 DIFFERENT BATTERY POWER AND DIFFERENT LATCH STATES, ON PURPOSE. A fixture whose
@@ -821,8 +822,8 @@ MUTATIONS = [
     ("per-unit entities announced at a roster of ONE",
      [("PER_UNIT_MIN_ROSTER = 2", "PER_UNIT_MIN_ROSTER = 1")]),
     ("the unit is dropped from the unique_id",
-     [('unique_id = f"{BASE}_unit_{slug}_{key}"',
-       'unique_id = f"{BASE}_unit_{key}"')]),
+     [('return f"{BASE}_unit_{slug}_{key}"',
+       'return f"{BASE}_unit_{key}"')]),
     ("the unit is dropped from the state topic",
      [('return f"unit/{slug}/{key}"', 'return f"unit/{key}"')]),
     ("the slug de-collision is removed (rule 9(c) does not cover this)",
@@ -944,6 +945,536 @@ for line in [
 ]:
     print("    " + line)
 
+# ---------------------------------------------------------------------------
+print("\n10. OPT-11 P2: each battery's switch and mode as HA entities")
+# ---------------------------------------------------------------------------
+import logging     # noqa: E402
+import os          # noqa: E402
+import tempfile    # noqa: E402
+
+from app import messages, packmode    # noqa: E402
+from app.store import Store, unit_key  # noqa: E402
+
+
+def announce_p2(units=(), omit=(), include=(), modes=(), retract=(),
+                module=mqtt):
+    """What `_on_connect()` sends: the announcement, then any subscriptions."""
+    pub = module.MqttPublisher.__new__(module.MqttPublisher)
+    pub.omit, pub.units, pub.include = set(omit), [str(u) for u in units], \
+        [str(k) for k in include]
+    pub.modes, pub.retract = tuple(modes), list(retract)
+    pub.enabled, pub.client = True, FakeClient()
+    subs = []
+    pub.client.subscribe = lambda topic, *a: subs.append(topic)
+    pub._on_connect()
+    return pub.client.msgs, subs
+
+
+# --- 10a. modes off: byte-identical to ed739d4 --------------------------------
+# The digest below is GENERATED from release/1.0 at ed739d4 (the base of this
+# PR): the same matrix run through that commit's `_announce()`. Reproduce with
+# `git show ed739d4:goodwe_controller/app/mqtt.py` and this loop.
+ROSTERS = [[], ["inverter-a"], ["inverter-a", "inverter-b"], ["a", "b", "c"],
+           ["inverter-b", "Inverter B", "x", "y"], ["café"]]
+ED739D4_OFF = "501a7d93782a260b"
+_h, _n, _subs_off = hashlib.sha256(), 0, []
+for _r in ROSTERS:
+    for _om in [(), ("p1_age",), ("p1_age", "grid")]:
+        for _inc in [(), ("pv",)]:
+            _m, _s = announce_p2(_r, _om, _inc)
+            _h.update(repr(_m).encode())
+            _subs_off += _s
+            _n += 1
+print(f"    modes off, {_n} configurations: digest {_h.hexdigest()[:16]} "
+      f"(ed739d4: {ED739D4_OFF})")
+check("modes off: every configuration announces byte-for-byte what ed739d4 "
+      "announces, in the same order, and subscribes to nothing",
+      _h.hexdigest()[:16] == ED739D4_OFF and not _subs_off)
+
+
+# --- 10b. modes on: the ordered payload diff ----------------------------------
+MODES = packmode.MODES
+ON1, SUB1 = announce_p2(["inverter-a"], modes=MODES)
+m, c, a = report("roster of one, modes on", ON1)
+_opt = '["Solar first", "Leftover solar"]'
+MODE_N1 = [
+    ("homeassistant/select/goodwe_ctl_unit_inverter_a_mode/config",
+     '{"name": "Inverter A Mode", "object_id": "goodwe_inverter_a_mode", '
+     '"unique_id": "goodwe_ctl_unit_inverter_a_mode", '
+     '"state_topic": "goodwe_ctl/unit/inverter_a/mode", '
+     '"command_topic": "goodwe_ctl/unit/inverter_a/mode/set", '
+     '"options": ' + _opt + ', "retain": false, '
+     '"availability_topic": "goodwe_ctl/availability", "device": ' + _DEV + ', '
+     '"icon": "mdi:battery-sync"}', True),
+    ("homeassistant/switch/goodwe_ctl_unit_inverter_a_use/config",
+     '{"name": "Inverter A Use", "object_id": "goodwe_inverter_a_use", '
+     '"unique_id": "goodwe_ctl_unit_inverter_a_use", '
+     '"state_topic": "goodwe_ctl/unit/inverter_a/use", '
+     '"command_topic": "goodwe_ctl/unit/inverter_a/use/set", '
+     '"payload_on": "ON", "payload_off": "OFF", "retain": false, '
+     '"availability_topic": "goodwe_ctl/availability", "device": ' + _DEV + ', '
+     '"icon": "mdi:power"}', True),
+]
+check("modes on at a roster of one: every existing message byte-identical and "
+      "in place, then exactly the select and the switch, then availability",
+      ON1 == SITE_NOW[:-1] + MODE_N1 + SITE_NOW[-1:])
+check("...and it subscribes to the two command topics, and nothing else",
+      SUB1 == ["goodwe_ctl/unit/+/mode/set", "goodwe_ctl/unit/+/use/set"])
+
+ON2, _ = announce_p2(NAMES2, modes=MODES)
+m, c, a = report("roster of two, modes on", ON2, TWO)
+check("modes on at a roster of two: nothing missing or changed, four added, "
+      "and the modes-off list is an unchanged prefix",
+      (m, c) == ([], []) and len(a) == 4 and ON2[:len(TWO) - 1] == TWO[:-1]
+      and ON2[-1] == TWO[-1])
+_mc = cfgs(ON2)[-4:]
+check("...A's two stay on the site device with its name in front; B's two are "
+      "on B's own device, as TEL-07's rule has it",
+      [c_["name"] for c_ in _mc] == ["Inverter A Mode", "Inverter A Use",
+                                      "Mode", "Use"]
+      and _mc[0]["device"] == mqtt.DEVICE == _mc[1]["device"]
+      and _mc[2]["device"] == _mc[3]["device"]
+      == mqtt.unit_device("inverter-b", "inverter_b"))
+check("...unique_ids are stable per battery and distinct from every other id",
+      [c_["unique_id"] for c_ in _mc]
+      == ["goodwe_ctl_unit_inverter_a_mode", "goodwe_ctl_unit_inverter_a_use",
+          "goodwe_ctl_unit_inverter_b_mode", "goodwe_ctl_unit_inverter_b_use"]
+      and len({c_["unique_id"] for c_ in cfgs(ON2)}) == len(cfgs(ON2)))
+_ids = [compose(c_["device"]["name"], c_["name"]).split(".", 1)[1]
+        for c_ in _mc]
+print(f"    composed ids: {_ids}")
+check("...and HA composes the ids the design names: "
+      "goodwe_inverter_<x>_mode and goodwe_inverter_<x>_use",
+      _ids == ["goodwe_inverter_a_mode", "goodwe_inverter_a_use",
+               "goodwe_inverter_b_mode", "goodwe_inverter_b_use"])
+check("every mode has a plain label, in MODES order, and a later mode only "
+      "appends an option: no existing option or id changes",
+      all(k in mqtt.MODE_LABELS for k in MODES)
+      and len(set(mqtt.MODE_LABELS.values())) == len(mqtt.MODE_LABELS)
+      and json.loads(MODE_N1[0][1])["options"]
+      == [mqtt.MODE_LABELS[k] for k in MODES])
+# Review of #152: the HA options are what automations send, so they are kept
+# apart from the page's words on purpose. They start equal; if the page's
+# English changes, this goes red and the author decides: keep the HA option
+# (and change this pin) or rename it (a breaking change, in DECISIONS).
+_en = json.loads((HERE / "app" / "ui_text" / "en.json").read_text())
+check("each HA mode option is the page's English name for that mode, in "
+      "both places the page names it",
+      all(mqtt.MODE_LABELS[k] == _en["home_words"]["modes"][k]
+          == _en["home"]["batteries"]["choice"][k]["name"] for k in MODES))
+_later = dict(mqtt.MODE_LABELS, cheap_hours="Cheap hours only")
+_orig = mqtt.MODE_LABELS
+mqtt.MODE_LABELS = _later
+try:
+    _p3 = mqtt.mode_payloads(NAMES2, MODES + ("cheap_hours",))
+finally:
+    mqtt.MODE_LABELS = _orig
+_p2 = mqtt.mode_payloads(NAMES2, MODES)
+check("...(simulated P3: adding `cheap_hours` changes only the options list, "
+      "by one entry at its end)",
+      [t for t, _ in _p3] == [t for t, _ in _p2]
+      and all({k: v for k, v in x.items() if k != "options"}
+              == {k: v for k, v in y.items() if k != "options"}
+              for (_, x), (_, y) in zip(_p3, _p2))
+      and _p3[0][1]["options"] == _p2[0][1]["options"] + ["Cheap hours only"])
+
+
+# --- 10c. modes turned off: removal ------------------------------------------
+_cur2 = [t for t, _ in mqtt.mode_payloads(NAMES2, MODES)]
+_ret = mqtt.mode_retractions(_cur2, [])
+OFF2, _subs = announce_p2(NAMES2, retract=_ret)
+check("modes turned off: the modes-off list, then an EMPTY retained config "
+      "for each of the four, then availability, and no subscription",
+      OFF2 == TWO[:-1] + [(t, "", True) for t in _cur2] + TWO[-1:]
+      and not _subs)
+check("...the store keeps what was announced, so a roster that shrinks removes "
+      "only the battery that left",
+      mqtt.mode_retractions(_cur2, [t for t, _ in
+                                    mqtt.mode_payloads(["inverter-a"], MODES)])
+      == _cur2[2:])
+check("...and nothing from the store is removed that is not one of our mode "
+      "configs: a non-list, a number, a site sensor, another add-on's select",
+      mqtt.mode_retractions("homeassistant/select/x/config", []) == []
+      and mqtt.mode_retractions(
+          [3, None, "homeassistant/sensor/goodwe_ctl_grid/config",
+           "homeassistant/select/other_mode/config",
+           "homeassistant/select/goodwe_ctl_unit_a_mode/config/x",
+           "homeassistant/select/goodwe_ctl_unit_a_mode/config"], [])
+      == ["homeassistant/select/goodwe_ctl_unit_a_mode/config"])
+
+
+_on1 = [t for t, _ in mqtt.mode_payloads(["inverter-a"], MODES)]
+_k, _r = mqtt.mode_configs(["inverter-a"], MODES, _cur2)
+check("start-up (`mode_configs`): a battery that left is removed, and stays "
+      "in the store, so a start with no broker does not forget it",
+      _r == _cur2[2:] and _k == _on1 + _cur2[2:])
+_k, _r = mqtt.mode_configs(NAMES2, (), _cur2 + _cur2[:1])
+check("...modes turned off: all four are removed, once each, and kept",
+      _r == _cur2 and _k == _cur2)
+check("...never turned on: nothing to store and nothing to remove, so the "
+      "store is not written",
+      mqtt.mode_configs(NAMES2, (), None) == ([], []))
+_k, _r = mqtt.mode_configs(NAMES2, MODES, _cur2)
+check("...turned on again: nothing removed, and the list does not grow",
+      _r == [] and _k == _cur2)
+
+
+from app.main import mode_announcement  # noqa: E402
+_st = Store(os.path.join(tempfile.mkdtemp(), "ann.json"))
+_OPTS1M = dict(OPTS2, battery_modes_enabled=True,
+               inverters=OPTS2["inverters"][:1])
+_seen = [mode_announcement(dict(OPTS2), _st),
+         dict(_st.data)]
+_seen += [mode_announcement(dict(OPTS2, battery_modes_enabled=True), _st),
+          list(_st.data.get(mqtt.MODE_CONFIGS_KEY) or [])]
+_seen += [mode_announcement(_OPTS1M, _st),
+          list(_st.data.get(mqtt.MODE_CONFIGS_KEY) or [])]
+_seen += [mode_announcement(dict(OPTS2, battery_modes_enabled=False), _st),
+          list(_st.data.get(mqtt.MODE_CONFIGS_KEY) or [])]
+check("main's start-up, through a real Store: never on announces and stores "
+      "nothing; on stores the four; a roster of one removes B's two and keeps "
+      "them; off removes all four and the store keeps them",
+      _seen[0] == ((), []) and mqtt.MODE_CONFIGS_KEY not in _seen[1]
+      and _seen[2] == (MODES, []) and _seen[3] == _cur2
+      and _seen[4] == (MODES, _cur2[2:]) and sorted(_seen[5]) == sorted(_cur2)
+      and _seen[6] == ((), _seen[5])
+      and sorted(_seen[7]) == sorted(_cur2))
+_st = Store(os.path.join(tempfile.mkdtemp(), "ren.json"))
+mode_announcement(dict(OPTS2, battery_modes_enabled=True), _st)
+_ren = dict(OPTS2, battery_modes_enabled=True, inverters=[
+    OPTS2["inverters"][0], dict(OPTS2["inverters"][1], name="inverter-c")])
+_r = mode_announcement(_ren, _st)[1]
+_kept = list(_st.data.get(mqtt.MODE_CONFIGS_KEY) or [])
+check("...B renamed C while on: B's two are removed and stay stored beside "
+      "A's and C's, so a start with no broker does not forget them",
+      _r == _cur2[2:] and _kept[-2:] == _cur2[2:] and len(_kept) == 6
+      and mode_announcement(_ren, _st)[1] == _cur2[2:])
+check("battery modes are on only for exactly `true`: not \"true\", 1 or "
+      "\"false\"",
+      [packmode.enabled({"battery_modes_enabled": v})
+       for v in (True, "true", 1, "false", None)]
+      == [True, False, False, False, False])
+
+
+# --- 10d. state topics, from the store ----------------------------------------
+_tmp = tempfile.mkdtemp()
+OPTS2M = dict(OPTS2, battery_modes_enabled=True)
+
+
+def ctl_for(opts, path):
+    pub = _Pub()
+    pub.sent = []
+    pub.publish = lambda values, _p=pub: (_p.sent.append(dict(values)),
+                                          setattr(_p, "last", dict(values)))
+    return Controller(dict(opts), None, Store(path), pub,
+                      stop=asyncio.Event()), pub
+
+
+_path = os.path.join(_tmp, "state.json")
+ctlm, pubm = ctl_for(OPTS2M, _path)
+ctlm.publish()
+_fed = {f"{mqtt.BASE}/{k}" for k in pubm.last
+        if k.endswith("/mode") or k.endswith("/use")}
+check("every announced mode entity is fed, and nothing else is: the state "
+      "topics published are exactly the ones announced",
+      _fed == {c_["state_topic"] for c_ in _mc})
+check("...defaults with nothing in the store: both On, both Solar first",
+      [pubm.last[k] for k in ("unit/inverter_a/mode", "unit/inverter_a/use",
+                              "unit/inverter_b/mode", "unit/inverter_b/use")]
+      == ["Solar first", "ON", "Solar first", "ON"])
+_c_off, _p_off = ctl_for(dict(OPTS2, battery_modes_enabled=False),
+                         os.path.join(_tmp, "off2.json"))
+_c_off.publish()
+_c_abs, _p_abs = ctl_for(OPTS2, os.path.join(_tmp, "off3.json"))
+_c_abs.publish()
+check("modes off (false or absent): the controller publishes the same keys as "
+      "before, with no mode or switch topic",
+      set(_p_off.last) == set(_p_abs.last) == set(vals2)
+      and not [k for k in _p_off.last if k.endswith(("/mode", "/use"))])
+
+
+# --- 10e. commands: the /action path, validated -------------------------------
+class _Logs(logging.Handler):
+    def __init__(self):
+        super().__init__(logging.DEBUG)
+        self.rows = []
+
+    def emit(self, record):
+        self.rows.append((record.levelno, record.getMessage()))
+
+
+_logs = _Logs()
+logging.getLogger("goodwe").addHandler(_logs)
+logging.getLogger("goodwe").setLevel(logging.DEBUG)
+_calls = []
+_real_handle = ctlm.handle_action
+
+
+async def _spy(what, **kw):
+    _calls.append((what, kw))
+    return await _real_handle(what, **kw)
+
+
+ctlm.handle_action = _spy
+
+
+def cmd(topic, payload, retained=False, ctl=None):
+    ctl = ctl or ctlm
+    asyncio.run(ctl.mqtt_command(topic, payload, retained))
+
+
+def stored(ctl=None):
+    d = (ctl or ctlm).store.data
+    return {k: d.get(k) for k in (unit_key("inverter-a", "pack_use"),
+                                  unit_key("inverter-a", "pack_mode"),
+                                  unit_key("inverter-b", "pack_use"),
+                                  unit_key("inverter-b", "pack_mode"))}
+
+
+T_A = "goodwe_ctl/unit/inverter_a"
+T_B = "goodwe_ctl/unit/inverter_b"
+pubm.sent.clear()
+cmd(f"{T_B}/use/set", b"OFF")
+cmd(f"{T_A}/mode/set", b"Leftover solar")
+check("good commands go through handle_action() with the /action arguments",
+      _calls == [("pack_use", {"unit": "inverter-b", "value": False}),
+                 ("pack_mode", {"unit": "inverter-a",
+                                "value": "leftover_solar"})])
+check("...and reach the store under P1's keys, and nowhere else",
+      stored() == {unit_key("inverter-a", "pack_use"): None,
+                   unit_key("inverter-a", "pack_mode"): "leftover_solar",
+                   unit_key("inverter-b", "pack_use"): False,
+                   unit_key("inverter-b", "pack_mode"): None})
+check("...and the state topics say so at once, from the store",
+      pubm.sent and pubm.sent[-1] == {
+          "unit/inverter_a/mode": "Leftover solar", "unit/inverter_a/use": "ON",
+          "unit/inverter_b/mode": "Solar first", "unit/inverter_b/use": "OFF"})
+check("...and the action log records each change",
+      any("inverter-b switched off" in e for e in ctlm.events)
+      and any("inverter-a mode set to leftover solar" in e
+              for e in ctlm.events))
+
+_before = stored()
+_n_calls = len(_calls)
+BAD = [
+    (f"{T_A}/use/set", b"on"), (f"{T_A}/use/set", b"ON\n"),
+    (f"{T_A}/use/set", b"true"), (f"{T_A}/use/set", b"1"),
+    (f"{T_A}/use/set", b""),
+    (f"{T_A}/mode/set", b"solar_first"),          # a key, not the option
+    (f"{T_A}/mode/set", b"Cheap hours only"),     # not offered before P3
+    (f"{T_A}/mode/set", b"solar first"),
+    (f"{T_A}/mode/set", "Sοlar first".encode()),  # Greek omicron
+    (f"{T_A}/mode/set", b"Solar first\x00"),
+    (f"{T_A}/mode/set", b'{"action": "pack_use", "value": false}'),
+    (f"{T_A}/mode/set", b"Solar first; rm -rf /"),
+    (f"{T_A}/mode/set", b"' OR 1=1 --"),
+    (f"{T_A}/use/set", b"\xff\xfe"),              # not UTF-8
+    (f"{T_A}/mode/set", b"Solar first" * 10),     # too long
+    ("goodwe_ctl/unit/inverter_z/use/set", b"OFF"),   # no such battery
+    ("goodwe_ctl/unit/../use/set", b"OFF"),
+    ("goodwe_ctl/unit/inverter_a/soc/set", b"OFF"),
+    ("goodwe_ctl/unit/inverter_a/use/set/x", b"OFF"),
+    ("goodwe_ctl/unit/Inverter_A/use/set", b"OFF"),
+    (f"{T_A}/use/set", None),
+]
+for _t, _p in BAD:
+    cmd(_t, _p)
+check(f"all {len(BAD)} bad commands are refused: the store is unchanged",
+      stored() == _before)
+_passed = [c_ for c_ in _calls[_n_calls:]]
+check("...those that reach handle_action() carry a value /action also refuses "
+      "(None) or no known battery; nothing is coerced into a valid choice",
+      all(kw["value"] is None or kw["unit"] is None for _, kw in _passed))
+
+# Log once per topic and reason.
+_logs.rows.clear()
+for _ in range(5):
+    cmd(f"{T_B}/use/set", b"maybe")
+_warn = [r for r in _logs.rows if r[0] >= logging.WARNING]
+check("a refused command repeated 5 times is logged once at WARNING, in plain "
+      "words, and at DEBUG the other times",
+      len(_warn) == 1 and messages.pack_refused("value") in _warn[0][1]
+      and len([r for r in _logs.rows if r[0] == logging.DEBUG]) == 5)
+_logs.rows.clear()
+_flood, _ = ctl_for(OPTS2M, os.path.join(_tmp, "flood.json"))
+for _i in range(100):
+    cmd(f"goodwe_ctl/unit/x{_i}/use/set", b"OFF", ctl=_flood)
+_w = [r[1] for r in _logs.rows if r[0] >= logging.WARNING]
+_per = [w for w in _w if "still ignored" not in w]
+_more = [w for w in _w if "still ignored" in w]
+print(f"    flood: {len(_per)} per-topic WARNINGs, then {_more}")
+check("...and a flood of 100 different bad topics logs 32 WARNINGs, then the "
+      "rest only as a count, at 1 and 10 more (review of #152)",
+      len(_per) == 32 and len(_more) == 2
+      and "1 more so far" in _more[0] and "10 more so far" in _more[1])
+_logs.rows.clear()
+for _ in range(100):
+    cmd(f"{T_B}/use/set", b"maybe", ctl=_flood)
+_more = [r[1] for r in _logs.rows
+         if r[0] >= logging.WARNING and "still ignored" in r[1]]
+check("...so a repeated attack on ONE topic still shows it goes on: the count "
+      "reaches 100 in 3 lines, not 100",
+      len(_more) == 1 and "100 more so far" in _more[0])
+_logs.rows.clear()
+_inj, _ = ctl_for(OPTS2M, os.path.join(_tmp, "inj.json"))
+cmd("goodwe_ctl/unit/a\nERROR forged line\x1b[31m/use/set", b"OFF", ctl=_inj)
+_w = [r[1] for r in _logs.rows if r[0] >= logging.WARNING]
+check("...and a topic with a newline or an escape code cannot forge a log "
+      "line: it is logged printable only",
+      len(_w) == 1 and "\n" not in _w[0] and "\x1b" not in _w[0]
+      and "a?ERROR forged line?[31m" in _w[0])
+
+# Modes off: refused on the same path, even if a message arrives.
+cmd(f"{T_A}/use/set", b"OFF", ctl=_c_off)
+check("with modes off a command is refused by the same check /action makes",
+      stored(_c_off)[unit_key("inverter-a", "pack_use")] is None)
+
+
+# --- 10f. a retained command after a restart ----------------------------------
+_logs.rows.clear()
+_before = stored()
+_n_calls = len(_calls)
+cmd(f"{T_A}/use/set", b"OFF", retained=True)
+cmd(f"{T_B}/use/set", b"ON", retained=True)
+check("a RETAINED command (the broker replays it on subscribe, e.g. after a "
+      "restart) is never applied, and never reaches handle_action()",
+      stored() == _before and len(_calls) == _n_calls)
+check("...and it is logged once, in plain words",
+      any(messages.pack_refused("retained") in r[1]
+          for r in _logs.rows if r[0] >= logging.WARNING))
+ctl_r, pub_r = ctl_for(OPTS2M, _path)      # a restart: same state file
+ctl_r.publish()
+check("after a restart the state topics come from the store: A Leftover solar "
+      "and on, B off",
+      [pub_r.last[k] for k in ("unit/inverter_a/mode", "unit/inverter_a/use",
+                               "unit/inverter_b/use")]
+      == ["Leftover solar", "ON", "OFF"])
+logging.getLogger("goodwe").removeHandler(_logs)
+
+
+# --- 10g. the paho side ----------------------------------------------------------
+class _Msg:
+    def __init__(self, topic, payload, retain):
+        self.topic, self.payload, self.retain = topic, payload, retain
+
+
+_got = []
+
+
+async def _handler(*args):
+    _got.append(args)
+
+
+async def _drive():
+    pub = mqtt.MqttPublisher.__new__(mqtt.MqttPublisher)
+    pub._on_message(None, None, _Msg("t", b"x", 0))      # not bound: dropped
+    pub.bind(asyncio.get_running_loop(), _handler)
+    pub._on_message(None, None, _Msg(f"{T_A}/use/set", b"OFF", 1))
+    await asyncio.sleep(0.05)
+
+asyncio.run(_drive())
+check("paho's thread hands (topic, bytes, retained) to the controller's loop, "
+      "and drops a message that arrives before the controller exists",
+      _got == [(f"{T_A}/use/set", b"OFF", True)])
+
+_kws = {k.arg: ast.unparse(k.value) for c_ in _call for k in c_.keywords}
+_main_src = (HERE / "app" / "main.py").read_text()
+check("(source read) main() passes modes and retractions to the publisher, "
+      "and binds its commands to the controller",
+      _kws.get("modes") == "_modes" and _kws.get("retract") == "_retract"
+      and "pub.bind(asyncio.get_running_loop(), controller.mqtt_command)"
+      in _main_src
+      and "_modes, _retract = mode_announcement(opts, store)" in _main_src)
+
+
+# --- 10h. the guards, broken one at a time --------------------------------------
+def p2_properties(mod):
+    out = {}
+    try:
+        off, s_off = announce_p2(NAMES2, module=mod)
+        on1, s_on = announce_p2(["inverter-a"], modes=MODES, module=mod)
+        on2, _ = announce_p2(NAMES2, modes=MODES, module=mod)
+        out["off identical"] = off == TWO and not s_off
+        out["on payloads"] = on1 == ON1 and on2 == ON2
+        out["subscribes when on"] = s_on == SUB1
+        cur = [t for t, _ in mod.mode_payloads(NAMES2, MODES)]
+        out["retract filters store"] = mod.mode_retractions(
+            [3, "homeassistant/sensor/goodwe_ctl_grid/config",
+             "homeassistant/select/other/config"] + cur, cur) == []
+        out["retract keeps current"] = mod.mode_retractions(cur, cur) == []
+        out["retract published"] = announce_p2(
+            NAMES2, retract=cur[:1], module=mod)[0] == \
+            TWO[:-1] + [(cur[0], "", True)] + TWO[-1:]
+        pc = mod.parse_command
+        out["retained dropped"] = pc(f"{T_A}/use/set", b"OFF", True,
+                                     MODES)[3] == "retained"
+        out["size capped"] = pc(f"{T_A}/mode/set", b"Solar first" * 10,
+                                False, MODES)[3] == "size"
+        out["utf-8 only"] = pc(f"{T_A}/use/set", b"\xff", False,
+                               MODES)[3] == "encoding"
+        out["switch exact"] = (
+            pc(f"{T_A}/use/set", b"on", False, MODES)[2] is None
+            and pc(f"{T_A}/use/set", b"OFF", False, MODES)[2] is False)
+        out["select labels only"] = (
+            pc(f"{T_A}/mode/set", b"solar_first", False, MODES)[2] is None
+            and pc(f"{T_A}/mode/set", b"Leftover solar", False,
+                   MODES)[2] == "leftover_solar")
+        out["select offered only"] = pc(
+            f"{T_A}/mode/set", b"Solar first", False, ("leftover_solar",))[2] \
+            is None
+        out["topic exact"] = all(
+            pc(t, b"OFF", False, MODES)[3] == "topic"
+            for t in ("goodwe_ctl/unit/../use/set", "goodwe_ctl/unit/a/b/use/set",
+                      "goodwe_ctl/unit/a/use/set/x", "x/goodwe_ctl/unit/a/use/set"))
+    except Exception as err:  # noqa: BLE001
+        out["did not raise"] = False
+        out["__error__"] = repr(err)[:70]
+    return out
+
+
+P2_SHIPPED = p2_properties(mqtt)
+check("the shipped module has every P2 property (so a red below is the "
+      "mutation)", all(v is True for v in P2_SHIPPED.values()))
+P2_MUTATIONS = [
+    ("mode entities announced with modes off",
+     [("    if not modes:\n        return []\n", "    modes = modes or ('solar_first', 'leftover_solar')\n")]),
+    ("subscribed with modes off", [("        if self.modes:\n", "        if True:\n")]),
+    ("retained commands applied", [("    if retained:\n", "    if False:\n")]),
+    ("no payload size cap", [("or len(payload) > MAX_COMMAND_BYTES", "or False")]),
+    ("bad UTF-8 replaced, not refused",
+     [('.decode("utf-8")', '.decode("utf-8", "replace")')]),
+    ("switch payload case-folded",
+     [("{USE_ON: True, USE_OFF: False}.get(text)",
+       "{USE_ON: True, USE_OFF: False}.get(text.upper())")]),
+    ("select accepts mode keys too",
+     [("if MODE_LABELS.get(k) == text", "if text in (k, MODE_LABELS.get(k))")]),
+    ("select accepts any labelled mode, offered or not",
+     [("(k for k in modes if MODE_LABELS.get(k) == text)",
+       "(k for k in MODE_LABELS if MODE_LABELS.get(k) == text)")]),
+    ("command topic loosened", [('/([a-z0-9_]+)/(mode|use)/set")', '/(.+)/(mode|use)/set")')]),
+    ("command topic matched as a prefix",
+     [("_COMMAND_RE.fullmatch(topic)", "_COMMAND_RE.search(topic)")]),
+    ("store list not filtered",
+     [("and _MODE_CONFIG_RE.fullmatch(t)\n", "and True\n")]),
+    ("current configs also removed", [("and t not in keep))", "))")]),
+    ("commands published retained by HA", [('cfg["retain"] = False', 'cfg["retain"] = True')]),
+    ("retractions not published",
+     [('            self.client.publish(topic, "", retain=True)\n', "            pass\n")]),
+]
+_p2_surv = []
+for label, pairs in P2_MUTATIONS:
+    got = properties(load(pairs))
+    got.update(p2_properties(load(pairs)))
+    broke = sorted(k for k, v in {**SHIPPED, **P2_SHIPPED}.items()
+                   if v is True and got.get(k) is not True)
+    print(f"    {'RED ' if broke else 'GREEN'} {label}"
+          + (f"  (breaks: {', '.join(broke)})" if broke else ""))
+    if not broke:
+        _p2_surv.append(label)
+check(f"all {len(P2_MUTATIONS)} P2 mutations of app/mqtt.py are red",
+      not _p2_surv)
+
+
 # ---------------------------------------------------------------------------
 print()
 if fails:

goodwe_controller/test_peak_margin.py

@@ -0,0 +1,298 @@
+"""Runnable check for SAFETY-79, the margin under the peak limit.
+`python3 test_peak_margin.py`
+
+The peak guard aims at `peak_cap_w - peak_margin_w` (default 50 W); the bill
+and every report keep the real `peak_cap_w`. DECISIONS/SAFETY-79-peak-margin.md
+has the record. Drives the REAL Controller through tools/safety79_identity.py.
+
+1. `peak_margin_w: 0` is byte-identical to release/1.0 ed739d4.
+2. The guard aims at cap - margin at N=1 and N=2, and with OPT-11 modes on an
+   Off battery covers the shortfall to cap - margin.
+3. The bill and the reports still use the real cap.
+4. Each call site in main.py passes the value its side needs.
+5. The setting: default, range, refusals in plain words.
+
+Mutations (MUTATION.md, purged copies) are listed in the pull request.
+"""
+
+import ast
+import os
+import re
+import sys
+
+HERE = os.path.dirname(os.path.abspath(__file__))
+sys.path.insert(0, os.path.join(HERE, "tools"))
+
+import safety79_identity as rig                               # noqa: E402
+from named_absence import nth, present                        # noqa: E402
+from style_rules import text_problems                         # noqa: E402
+from app import messages, peakguard                 # noqa: E402
+from app.control import maintenance_charge_floor              # noqa: E402
+from app.peakguard import (PEAK_MARGIN_RANGE_W,               # noqa: E402
+                           PEAK_OPTION_DEFAULTS, quarter_target_w)
+from app.validate import config_notes, validate_startup_config  # noqa: E402
+
+fails = []
+
+
+def check(name, cond):
+    print(f"  {'ok  ' if cond else 'FAIL'} {name}")
+    if not cond:
+        fails.append(name)
+
+
+CAP = float(rig.CAP_W)
+
+# ============================================================================ #
+print("1. peak_margin_w 0: byte-identical to release/1.0 ed739d4")
+# Printed by `python3 tools/safety79_identity.py 0` on the ed739d4 tree (the
+# tool copied in, nothing else changed; that tree ignores the key), in the CI
+# image; re-printed 2026-09-29 (review) after adding the charge headroom, the
+# shaping suspension and the two maintenance scenarios. 540 cycles each.
+ED739D4 = {
+    "n1": "82316f7d4362fae4c035a1bed5e7907cf362faba994c634d0b1c356c501f53e5",
+    "n2": "0dc24e16a8390ec42edacf735eaf20273847b33b9c67d7cdc536df0ca739e421",
+    "n1_off": "eed44f058f3fcc6e7651c5c84c6fa6313e7520035ad7cac282012cf6909668a0",
+    "n2_off": "69796c107b9f78892e04d282b754f889590acf9fe1b96f09525f5ea2005e9c0a",
+    "n1_maint":
+        "bd4a88a10aa840a360ac781ac04e30b040a5aac5426fabba1fb259d71dfeba67",
+    "n2_maint":
+        "0656bc5a4a2fa3a7455822fdd7da5f68f57fd9593f374e869a0d30ba7e429275",
+}
+_zero = rig.digests(0)
+_default = rig.digests()
+for k in ED739D4:
+    print(f"    {k:7} margin 0 {'same' if _zero[k] == ED739D4[k] else 'DIFFERS'}"
+          f" {_zero[k][:16]}, default {_default[k][:16]}")
+check("⭐ at N=1 and N=2, with and without battery modes and with a "
+      "maintenance charge, margin 0 is "
+      "byte-identical to ed739d4", _zero == ED739D4)
+check("...and the default (key absent) changes every scenario, so the "
+      "scenarios do reach the margin", all(_default[k] != ED739D4[k]
+                                          for k in ED739D4))
+
+# ============================================================================ #
+print("\n2. the guard aims at cap - margin")
+_runs = {}
+for name, spec in rig.SCENARIOS.items():
+    for m in (None, 0, 120):
+        _runs[name, m] = rig.run(**spec, margin=m, dur=3600.0)
+for name in rig.SCENARIOS:
+    g = _runs[name, None][1].peak_guard
+    print(f"    {name}: margin {_runs[name, None][1].peak_margin_w:.0f} W, "
+          f"aims at {g.target_w:.0f} W, limit {g.ceiling_w:.0f} W")
+check("test_target_is_cap_minus_margin: with the key absent every scenario "
+      "(N=1, N=2, modes on and off) aims at cap - 50 and reports the cap",
+      all(_runs[n, None][1].peak_margin_w == 50.0
+          and _runs[n, None][1].peak_guard.target_w == CAP - 50
+          and _runs[n, None][1].peak_guard.ceiling_w == CAP
+          for n in rig.SCENARIOS))
+check("...and a margin of 120 moves the aim to cap - 120, a margin of 0 to "
+      "the cap", all(_runs[n, 120][1].peak_guard.target_w == CAP - 120
+                     and _runs[n, 0][1].peak_guard.target_w == CAP
+                     for n in rig.SCENARIOS))
+
+
+def banked(ctl):
+    return {b.start.strftime("%H:%M"): round(b.offtake_avg_w, 2)
+            for b in ctl.p1.blocks}
+
+
+_b = {m: banked(_runs["n1_off", m][1]) for m in (0, None, 120)}
+print(f"    N=1, the battery Off: quarters banked {_b}")
+check("test_off_battery_n1: at N=1 with the battery Off, the guard makes it "
+      "give just enough to bank the 11:30 quarter at cap - margin (2500, "
+      "2450, 2380 W for margins 0, 50, 120)",
+      [_b[m]["11:30"] for m in (0, None, 120)]
+      == [CAP, CAP - 50, CAP - 120])
+
+
+def meter_at(rows, t):
+    row = next((r for r in rows if r[0] == t), None)
+    if row is None:
+        return None, {}
+    return rig.load(t) - sum(w for _u, w in row[2]), dict(row[2])
+
+
+_g0, _c0 = meter_at(_runs["n2_off", 0][0], 1800.0)
+_g50, _c50 = meter_at(_runs["n2_off", None][0], 1800.0)
+print(f"    N=2, B Off, 6000 W house: margin 0 -> {_c0}, meter {_g0} W; "
+      f"margin 50 -> {_c50}, meter {_g50} W")
+check("test_off_battery_n2: at N=2 with B Off and A at its rating, B covers "
+      "the shortfall to cap - margin, not to the cap",
+      _c0["b"] == 1000.0 and _g0 == CAP
+      and _c50["b"] == 1050.0 and _g50 == CAP - 50)
+
+def _peak(w, observed=True):
+    return type("R", (), {"is_observed": observed, "peak_w": w})()
+
+
+# Review (SAFETY-79 point 2): the margin applies under a month peak already
+# banked, too. Aiming AT it can bank a few W over it, which raises the bill;
+# and a banked peak between cap - margin and the cap (a quarter the old guard
+# held at the cap, say) was aimed at exactly, so one overshoot crossed the cap.
+check("test_margin_under_banked_peak: the aim is the margin under the limit "
+      "that holds - the cap, or a month peak already banked above it",
+      quarter_target_w(CAP, 50, _peak(2499.0)) == CAP - 50
+      and quarter_target_w(CAP, 50, _peak(3000.0)) == 2950.0
+      and quarter_target_w(CAP, 50, _peak(3000.0, observed=False)) == CAP - 50
+      and quarter_target_w(CAP, 50, None) == CAP - 50
+      and quarter_target_w(CAP, 0, _peak(3000.0)) == 3000.0)
+
+
+def _decide(banked_w, projected_w):
+    """Half a quarter at `projected_w`, battery idle, a month peak banked."""
+    from datetime import datetime, timezone
+    from app.tariff import Projection
+    pr = Projection(projected_w, projected_w, 450.0, 1.0, False,
+                    datetime(2026, 9, 1, 12, 0, tzinfo=timezone.utc),
+                    projected_w)
+    return peakguard.decide(pr, batt_w=0.0, cap_w=CAP, margin_w=50,
+                            reading=_peak(banked_w), achievable_w=5000.0,
+                            soc=60.0, soc_floor=20.0)
+
+
+_d1, _d2 = _decide(2499.0, 2600.0), _decide(3000.0, 3100.0)
+print(f"    banked 2499 W: floor {_d1.floor_w}, limit {_d1.ceiling_w}; "
+      f"banked 3000 W: floor {_d2.floor_w}, limit {_d2.ceiling_w}")
+check("...and decide() sizes the floor to land the quarter 50 W under it "
+      "(2449 and 2950 W), while the reported limit stays the cap or the peak",
+      _d1.floor_w == 2 * (2600.0 - 2450.0) and _d1.ceiling_w == CAP
+      and _d2.floor_w == 2 * (3100.0 - 2950.0) and _d2.ceiling_w == 3000.0)
+
+# ============================================================================ #
+print("\n3. the bill and the reports keep the real cap")
+_c = _runs["n1_off", None][1]
+_pk = _c.peak.peak().peak_w
+print(f"    month peak {_pk:.0f} W, tracker cap {_c.peak.cap_w:.0f} W, "
+      f"warning {_c.peak.warning_frac():.3f}, headroom "
+      f"{_c.peak.headroom_w():.0f} W")
+check("test_billing_uses_cap: the month-peak tracker (bill, headroom, 80 % "
+      "warning) is built on the real cap, not cap - margin",
+      _c.peak.cap_w == CAP and _c.peak.warning_frac() == _pk / CAP
+      and _c.peak.headroom_w() == CAP - _pk)
+_eng = [g for r in _runs["n1_off", None][0] for g in (r[-1][1],)
+        if g and g[0] and g[1]]
+print(f"    {len(_eng)} engaged cycles; one: {_eng[0][4] if _eng else None}")
+check("test_reports_use_cap: every engaged sentence names the real limit and "
+      "the aim beside it", _eng and all(
+          g[2] == CAP and "against a 2500 W limit (peak_cap_w), aiming at "
+          "2450 W" in g[4] for g in _eng))
+check("...and the settings note compares the month peak with the real cap: "
+      "a 2480 W peak under a 2500 W cap is no warning, 2520 W is",
+      config_notes({"peak_cap_w": 2500, "peak_margin_w": 50},
+                   month_peak_w=2480) == []
+      and [n.keys for n in config_notes({"peak_cap_w": 2500,
+                                         "peak_margin_w": 50},
+                                        month_peak_w=2520)]
+      == [("peak_cap_w",)])
+_lim = _eng[0][2] if _eng else 0.0
+_notice = messages.peak_saturated(_lim)
+check("...and the user-facing peak notice built from the decision's limit "
+      "says the cap", messages.watts(CAP) in _notice.body
+      and messages.watts(CAP - 50) not in _notice.body)
+
+# ============================================================================ #
+print("\n4. each call site in main.py")
+with open(os.path.join(HERE, "app", "main.py"), encoding="utf-8") as f:
+    _src = f.read()
+_tree = ast.parse(_src)
+
+
+def calls(name):
+    return [n for n in ast.walk(_tree) if isinstance(n, ast.Call)
+            and isinstance(n.func, (ast.Attribute, ast.Name))
+            and getattr(n.func, "attr", getattr(n.func, "id", "")) == name]
+
+
+def kw(call, key):
+    if not present(call):
+        return "absent"
+    return next((ast.get_source_segment(_src, k.value) for k in call.keywords
+                 if k.arg == key), "absent")
+
+
+def arg(call, i):
+    if not present(call) or len(call.args) <= i:
+        return "absent"
+    return ast.get_source_segment(_src, call.args[i])
+
+
+_dec = [c for c in calls("decide")
+        if getattr(c.func, "value", None) is not None
+        and getattr(c.func.value, "id", "") == "peakguard"]
+_shape = calls("shaping_suspended")
+_maint = calls("maintenance_charge_floor")
+_tracker = calls("PeakTracker")
+_d, _s, _m, _t = (nth(_dec, 0, "peakguard.decide()"),
+                  nth(_shape, 0, "shaping_suspended()"),
+                  nth(_maint, 0, "maintenance_charge_floor()"),
+                  nth(_tracker, 0, "PeakTracker()"))
+print(f"    decide(cap_w={kw(_d, 'cap_w')}, margin_w={kw(_d, 'margin_w')})")
+print(f"    shaping_suspended(..., {arg(_s, 2)})")
+print(f"    maintenance_charge_floor(..., {arg(_m, 2)})")
+print(f"    PeakTracker(cap_w={kw(_t, 'cap_w')})")
+check("test_call_sites: the guard gets the real cap AND the margin; the "
+      "forecast checks get cap - margin; the month-peak tracker gets the "
+      "real cap",
+      len(_dec) == 1 and "peak_cap_w" in kw(_d, "cap_w")
+      and kw(_d, "margin_w") == "self.peak_margin_w"
+      and len(_shape) == 1 and len(_maint) == 1
+      and arg(_s, 2) == "self.peak_aim_w()"
+      and arg(_m, 2) == "self.peak_aim_w()"
+      and len(_tracker) == 1
+      and kw(_t, "cap_w") == 'float(opts.get("peak_cap_w", 3500))')
+check("test_aim_w: `peak_aim_w()` is cap - margin, and the forecast sizing "
+      "gives a maintenance charge only what is left under it",
+      _c.peak_aim_w() == CAP - 50
+      and maintenance_charge_floor(2000.0, 2400.0, _c.peak_aim_w()) == 50.0
+      and peakguard.shaping_suspended(None, 2470.0, _c.peak_aim_w())
+      and not peakguard.shaping_suspended(None, 2470.0, CAP))
+
+# ============================================================================ #
+print("\n5. the setting")
+
+
+with open(os.path.join(HERE, "config.yaml"), encoding="utf-8") as f:
+    _hits = re.findall(r"^  peak_margin_w:\s*(\S+)", f.read(), re.M)
+print(f"    config.yaml: {_hits}")
+check("config.yaml ships 50, int(0,500), and the in-code mirror matches",
+      _hits == ["50", "int(0,500)"]
+      and PEAK_OPTION_DEFAULTS["peak_margin_w"] == 50
+      and PEAK_MARGIN_RANGE_W == (0, 500))
+
+BASE = {"setpoint_entity": "number.sp", "meter_entity": "sensor.grid",
+        "max_w": 2000, "peak_cap_w": 3500}
+_base_problems = validate_startup_config(BASE)
+
+
+def refused(value, **more):
+    opts = dict(BASE, peak_margin_w=value, **more)
+    return [p for p in validate_startup_config(opts)
+            if p not in _base_problems]
+
+
+_ok = {v: refused(v) for v in (0, 50, 500)}
+_bad = {v: refused(v) for v in (-1, 501, 12.5, "abc")}
+for v, p in _bad.items():
+    print(f"    {v!r}: {p}")
+check("test_margin_range: 0, 50 and 500 are accepted, and so is the key "
+      "absent", not any(_ok.values()) and _base_problems == [])
+check("...and below 0, above 500, not whole, or not a number is refused, "
+      "once, naming the setting and the range",
+      all(len(p) == 1 and "“Margin below the peak limit” (peak_margin_w)"
+          in p[0] and "from 0 to 500 W" in p[0] for p in _bad.values()))
+_high = refused(500, peak_cap_w=500)
+print(f"    500 under a 500 W limit: {_high}")
+check("...and a margin that leaves nothing under the limit is refused",
+      len(_high) == 1 and "not below" in _high[0])
+_style = [text_problems(p[0], keys=("peak_margin_w", "peak_cap_w"),
+                        entities=False) for p in list(_bad.values()) + [_high]]
+check(f"...in plain words (STYLE.md): {_style}", not any(_style))
+
+
+print()
+if fails:
+    print(f"{len(fails)} FAILED: {', '.join(fails)}")
+    sys.exit(1)
+print("all checks passed")

goodwe_controller/test_peakguard.py

@@ -368,7 +368,7 @@ check("...and it grows monotonically through the quarter, so the guard is at "
 BATT_NOW = 400.0
 a_edge = averager_at(CEIL, 450.0)
 p_edge = PeakTracker.project(a_edge, B + timedelta(seconds=450))
-d_edge = decide(p_edge, batt_w=BATT_NOW, cap_w=CAP_W, reading=None,
+d_edge = decide(p_edge, batt_w=BATT_NOW, cap_w=CAP_W, margin_w=0.0, reading=None,
                 achievable_w=LIVE_MAX_W, max_hold_s=MAX_HOLD_S, soc=SOC_OK, soc_floor=SOC_FLOOR)
 e_edge = excess_w(p_edge, CEIL)
 print(f"    held exactly AT the ceiling for 450 s: projected "
@@ -383,7 +383,7 @@ check("at the engagement boundary the excess is zero, so the floor is exactly "
 # than on the excess. Re-derived here rather than described: a guard that
 # released as soon as `excess <= 0` hands the actuator back at the exact instant
 # it is working, and the load springs straight back.
-released = decide(p_edge, batt_w=BATT_NOW, cap_w=CAP_W, reading=None,
+released = decide(p_edge, batt_w=BATT_NOW, cap_w=CAP_W, margin_w=0.0, reading=None,
                   achievable_w=LIVE_MAX_W, max_hold_s=MAX_HOLD_S, soc=SOC_OK, soc_floor=SOC_FLOOR)
 would_release_on_excess = excess_w(p_edge, CEIL) <= 0.0
 # ⚠️ `floor_w` IS FORMATTED DEFENSIVELY, AND MUTATION.md's OWN LEDGER IS WHY.
@@ -404,7 +404,7 @@ check("...and the claim is held there, not released. `excess` falls to zero as "
       "arithmetic read as a level", would_release_on_excess and released.engaged)
 d_gone = decide(PeakTracker.project(averager_at(500.0, 450.0),
                                     B + timedelta(seconds=450)),
-                batt_w=BATT_NOW, cap_w=CAP_W, reading=None,
+                batt_w=BATT_NOW, cap_w=CAP_W, margin_w=0.0, reading=None,
                 achievable_w=LIVE_MAX_W, max_hold_s=MAX_HOLD_S, soc=SOC_OK, soc_floor=SOC_FLOOR)
 check("...and it DOES release once the load falls under what the quarter can "
       "afford, so 'threshold the floor' is not 'never let go'",
@@ -413,7 +413,7 @@ check("...and it DOES release once the load falls under what the quarter can "
 # A hair over, and the floor is the battery plus the excess. Not a constant.
 a_over = averager_at(CEIL + 200.0, 450.0)
 p_over = PeakTracker.project(a_over, B + timedelta(seconds=450))
-d_over = decide(p_over, batt_w=BATT_NOW, cap_w=CAP_W, reading=None,
+d_over = decide(p_over, batt_w=BATT_NOW, cap_w=CAP_W, margin_w=0.0, reading=None,
                 achievable_w=LIVE_MAX_W, max_hold_s=MAX_HOLD_S, soc=SOC_OK, soc_floor=SOC_FLOOR)
 print(f"    200 W over for 450 s: {d_over.explain()}")
 check("a quarter projected over the ceiling engages, and the floor is the "
@@ -422,7 +422,7 @@ check("a quarter projected over the ceiling engages, and the floor is the "
       and near(d_over.floor_w, BATT_NOW + excess_w(p_over, CEIL), 1e-6))
 check("...and the floor moves with the battery measurement, so it is a claim "
       "about delivered power rather than a fixed setpoint",
-      not near(decide(p_over, batt_w=BATT_NOW + 250.0, cap_w=CAP_W, reading=None,
+      not near(decide(p_over, batt_w=BATT_NOW + 250.0, cap_w=CAP_W, margin_w=0.0, reading=None,
                       achievable_w=LIVE_MAX_W, max_hold_s=MAX_HOLD_S, soc=SOC_OK, soc_floor=SOC_FLOOR).floor_w,
                d_over.floor_w, 1.0))
 
@@ -433,7 +433,7 @@ intent = 0.0
 delivered_floor = []
 intent_floor = []
 for _ in range(5):
-    d_del = decide(p_over, batt_w=stuck_batt, cap_w=CAP_W, reading=None,
+    d_del = decide(p_over, batt_w=stuck_batt, cap_w=CAP_W, margin_w=0.0, reading=None,
                    achievable_w=LIVE_MAX_W, max_hold_s=MAX_HOLD_S, soc=SOC_OK, soc_floor=SOC_FLOOR)
     delivered_floor.append(d_del.floor_w)
     intent = intent + excess_w(p_over, CEIL)     # what the intent form would do
@@ -516,7 +516,7 @@ check(f"a loop already discharging harder ({BIG:.0f} W) than the floor is not "
 absurd = averager_at(LIVE_MAX_W * 4, 870.0)
 p_absurd = PeakTracker.project(absurd, B + timedelta(seconds=870))
 DEV_MAX = LIVE_MAX_W - 500.0        # a device rating tighter than max_w
-d_absurd = decide(p_absurd, batt_w=1000.0, cap_w=CAP_W, reading=None,
+d_absurd = decide(p_absurd, batt_w=1000.0, cap_w=CAP_W, margin_w=0.0, reading=None,
                   achievable_w=min(LIVE_MAX_W, DEV_MAX), max_hold_s=MAX_HOLD_S, soc=SOC_OK, soc_floor=SOC_FLOOR)
 r6 = resolve([Claim.limit("device", P_SAFETY, -DEV_MAX, DEV_MAX, "rating"),
               Claim.limit("supervised", P_SAFETY, -LIVE_MAX_W, LIVE_MAX_W, "max_w"),
@@ -599,7 +599,7 @@ print(f"    a real PeakStore on a missing volume: state={rd_broken.state}, "
       f"headroom_w()={tr_broken.headroom_w(B)}")
 check("the real store answers NOT_OBSERVED rather than a peak of 0 W",
       rd_broken.state == NOT_OBSERVED and rd_broken.peak_w is None)
-d_broken = decide(p_over, batt_w=BATT_NOW, cap_w=CAP_W, reading=rd_broken,
+d_broken = decide(p_over, batt_w=BATT_NOW, cap_w=CAP_W, margin_w=0.0, reading=rd_broken,
                   achievable_w=LIVE_MAX_W, max_hold_s=MAX_HOLD_S, soc=SOC_OK, soc_floor=SOC_FLOOR)
 print(f"    {d_broken.explain()}")
 check("AND THE GUARD STILL ENGAGES ON IT. An unreadable peak must not disarm "
@@ -627,7 +627,7 @@ print("4. blindness: three ways a projection is not evidence")
 # (a) NO OPEN BLOCK. A fresh averager projects 0 W - under every ceiling.
 fresh = QuarterAverager(phases=1, max_hold_s=MAX_HOLD_S)
 p_fresh = PeakTracker.project(fresh, B)
-d_fresh = decide(p_fresh, batt_w=0.0, cap_w=CAP_W, reading=None,
+d_fresh = decide(p_fresh, batt_w=0.0, cap_w=CAP_W, margin_w=0.0, reading=None,
                  achievable_w=LIVE_MAX_W, max_hold_s=MAX_HOLD_S, soc=SOC_OK, soc_floor=SOC_FLOOR)
 print(f"    fresh averager: projected {p_fresh.projected_w:.0f} W, "
       f"block_start={p_fresh.block_start}; {d_fresh.explain()}")
@@ -655,7 +655,7 @@ p_frozen = PeakTracker.project(frozen_av, B + timedelta(seconds=600))
 RAILED = LIVE_MAX_W                          # the loop has wound to the rail
 is_frozen = looks_frozen(FREEZE_S + 1.0, -RAILED, LIVE_MAX_W,
                          FreezeWatchdogConfig(hold_s=FREEZE_S))
-d_frozen = decide(p_frozen, batt_w=-RAILED, cap_w=CAP_W, reading=None,
+d_frozen = decide(p_frozen, batt_w=-RAILED, cap_w=CAP_W, margin_w=0.0, reading=None,
                   achievable_w=LIVE_MAX_W, max_hold_s=MAX_HOLD_S, soc=SOC_OK, soc_floor=SOC_FLOOR,
                   frozen=is_frozen)
 print(f"    meter stuck at -4000 W with the command at {-RAILED:+.0f} W: "
@@ -680,7 +680,7 @@ check("...and the same command NOT frozen is left alone, so the blind branch is
       "not simply always on",
       near(resolve(SAFE + [Claim.set("loop", P_LOOP, -RAILED, "grid"),
                            peakguard.claim(decide(
-                               p_frozen, batt_w=-RAILED, cap_w=CAP_W,
+                               p_frozen, batt_w=-RAILED, cap_w=CAP_W, margin_w=0.0,
                                reading=None, achievable_w=LIVE_MAX_W,
                                max_hold_s=MAX_HOLD_S, soc=SOC_OK, soc_floor=SOC_FLOOR, frozen=False))
                            or Claim.limit("noop", P_PEAK)]).target_w,
@@ -692,7 +692,7 @@ gapped = averager_at(4200.0, 600.0, gap_at=200.0, gap_s=GAP)
 p_gap = PeakTracker.project(gapped, B + timedelta(seconds=600))
 clean = averager_at(4200.0, 600.0)
 p_clean = PeakTracker.project(clean, B + timedelta(seconds=600))
-d_gap = decide(p_gap, batt_w=0.0, cap_w=CAP_W, reading=None,
+d_gap = decide(p_gap, batt_w=0.0, cap_w=CAP_W, margin_w=0.0, reading=None,
                achievable_w=LIVE_MAX_W, max_hold_s=MAX_HOLD_S, soc=SOC_OK, soc_floor=SOC_FLOOR)
 print(f"    a {GAP:.0f} s hole at 4200 W: uncovered {uncovered_s(p_gap):.0f} s, "
       f"projected {p_gap.projected_w:.0f} W vs {p_clean.projected_w:.0f} W with "
@@ -756,7 +756,7 @@ check("a renamed held value now raises AttributeError on the first cycle. The "
 OLD = (p_held.partial_ws if hasattr(p_held, "partial_ws")
        else HELD * 300.0)   # acc, which the old form would still have had
 old_projected = (HELD * 300.0 + 0.0 * (QUARTER_S - 300.0)) / QUARTER_S
-d_true = decide(p_held, batt_w=0.0, cap_w=CAP_W, reading=None,
+d_true = decide(p_held, batt_w=0.0, cap_w=CAP_W, margin_w=0.0, reading=None,
                 achievable_w=LIVE_MAX_W, max_hold_s=MAX_HOLD_S, soc=SOC_OK, soc_floor=SOC_FLOOR)
 print(f"    same fixture: shipped projection {p_held.projected_w:.0f} W "
       f"(engaged={d_true.engaged}), the defaulted form's "
@@ -832,7 +832,7 @@ def replay(path, *, guarded: bool, cap_w=CAP_W, max_w=LIVE_MAX_W, pack=None,
         next_cycle = t + timedelta(seconds=CYCLE_S)
         cycles += 1
         d = decide(PeakTracker.project(ing.averager, t),
-                   batt_w=batt, cap_w=cap_w, reading=tracker.peak(t),
+                   batt_w=batt, cap_w=cap_w, margin_w=0.0, reading=tracker.peak(t),
                    achievable_w=max_w, max_hold_s=MAX_HOLD_S, soc=soc,
                    soc_floor=floor_pct, frozen=False)
         if d.floor_w is not None and d.floor_w < 0.0:
@@ -1175,7 +1175,9 @@ def wired_over_limit(db, want_floor_w, opts=None, store=None, batt_w=0.0,
             int(now.timestamp() // QUARTER_S) * QUARTER_S, timezone.utc)
         span = int((now - blk).total_seconds())
         remaining = max(1.0, QUARTER_S - (now - blk).total_seconds())
-        net_w = CAP_W + want_floor_w * remaining / QUARTER_S
+        # SAFETY-79: sized against what the guard aims at, cap - margin.
+        net_w = (CAP_W - ctl.peak_margin_w
+                 + want_floor_w * remaining / QUARTER_S)
         for i in range(0, span + 1, 5):
             ctl.p1.submit(sample(blk + timedelta(seconds=i), net_w))
         # ⚠️ AND ONE TELEGRAM AT `now` ITSELF. The averager credits a sample's
@@ -1417,7 +1419,8 @@ check("B3: main.py sizes the floor from the MEASURED battery power, not from "
 # `datetime.now()`. Every number below therefore comes from ONE instant, and the
 # tolerances that used to absorb the wall clock are gone.
 _proj3 = _seen["projection"]
-_kw = dict(cap_w=CAP_W, reading=b3.peak.peak(), achievable_w=LIVE_MAX_W,
+_kw = dict(cap_w=CAP_W, margin_w=b3.peak_margin_w, reading=b3.peak.peak(),
+           achievable_w=LIVE_MAX_W,
            max_hold_s=b3.p1.averager.max_hold_s, soc=SOC_OK,
            soc_floor=SOC_FLOOR)
 _intent = _shipped_decide(_proj3, batt_w=STANDING, **_kw)
@@ -1554,7 +1557,8 @@ check("SAFETY-24 adds exactly the two keys it names and no more: §14's kW/W "
       "trap `capacity_peak_limit_kw` stays declined, and no speculative "
       "`peak_guard`/`peak_shave` block appeared beside it",
       _peak_keys == []
-      and sorted(PEAK_OPTION_DEFAULTS) == ["bms_soc_floor", "peak_soc_floor"])
+      and sorted(PEAK_OPTION_DEFAULTS) == ["bms_soc_floor", "peak_margin_w",
+                                           "peak_soc_floor"])   # + SAFETY-79
 # Graded on the parsed call syntax, not on the text: the module docstring names
 # `peak_cap_w` while arguing that it does not read it, so a grep would be
 # satisfied by deleting the explanation and would fail on writing one.
@@ -1634,13 +1638,13 @@ check("...and the defaulting form that was ACTUALLY shipped once in this "
       discharge_permitted(float(None or 100), SOC_FLOOR) is None)
 
 # --- the decision -----------------------------------------------------------
-d_ok = decide(p_hungry, batt_w=BATT_NOW, cap_w=CAP_W, reading=None,
+d_ok = decide(p_hungry, batt_w=BATT_NOW, cap_w=CAP_W, margin_w=0.0, reading=None,
               achievable_w=LIVE_MAX_W, max_hold_s=MAX_HOLD_S, soc=SOC_OK,
               soc_floor=SOC_FLOOR)
-d_lim = decide(p_hungry, batt_w=BATT_NOW, cap_w=CAP_W, reading=None,
+d_lim = decide(p_hungry, batt_w=BATT_NOW, cap_w=CAP_W, margin_w=0.0, reading=None,
                achievable_w=LIVE_MAX_W, max_hold_s=MAX_HOLD_S,
                soc=SOC_FLOOR, soc_floor=SOC_FLOOR)
-d_unread = decide(p_hungry, batt_w=BATT_NOW, cap_w=CAP_W, reading=None,
+d_unread = decide(p_hungry, batt_w=BATT_NOW, cap_w=CAP_W, margin_w=0.0, reading=None,
                   achievable_w=LIVE_MAX_W, max_hold_s=MAX_HOLD_S, soc=None,
                   soc_floor=SOC_FLOOR)
 print(f"    with battery:  {d_ok.explain()}")
@@ -1682,7 +1686,7 @@ check("floor-limited and saturated are MUTUALLY EXCLUSIVE, by construction: the
 # notification that matters.
 d_quiet = decide(PeakTracker.project(averager_at(500.0, 450.0),
                                      B + timedelta(seconds=450)),
-                 batt_w=BATT_NOW, cap_w=CAP_W, reading=None,
+                 batt_w=BATT_NOW, cap_w=CAP_W, margin_w=0.0, reading=None,
                  achievable_w=LIVE_MAX_W, max_hold_s=MAX_HOLD_S,
                  soc=SOC_FLOOR - 5.0, soc_floor=SOC_FLOOR)
 print(f"    an EMPTY pack on a quiet quarter: {d_quiet.explain()}")
@@ -1694,7 +1698,7 @@ check("an empty pack on a quarter that does not need shaving is IDLE, not "
       and d_quiet.floor_w is None)
 # And the BLIND branch is untouched: it never forces a discharge, so there is
 # nothing for an energy floor to bound there.
-d_blind_low = decide(p_frozen, batt_w=-RAILED, cap_w=CAP_W, reading=None,
+d_blind_low = decide(p_frozen, batt_w=-RAILED, cap_w=CAP_W, margin_w=0.0, reading=None,
                      achievable_w=LIVE_MAX_W, max_hold_s=MAX_HOLD_S,
                      soc=SOC_FLOOR - 5.0, soc_floor=SOC_FLOOR, frozen=True)
 check("...and an empty pack does not disturb the BLIND branch either: SAFETY-"
@@ -1751,7 +1755,7 @@ for load in (0.0, CEIL - 500.0, CEIL, CEIL + 300.0, CEIL + 3000.0,
         for bw in (-2000.0, 0.0, 400.0, 4000.0):
             for sc in (None, 0.0, BMS_FLOOR, SOC_FLOOR, SOC_FLOOR + 0.5, 100.0):
                 for fl in (SOC_FLOOR, BMS_FLOOR + 1.0, 55.0):
-                    _d = decide(_pr, batt_w=bw, cap_w=CAP_W, reading=None,
+                    _d = decide(_pr, batt_w=bw, cap_w=CAP_W, margin_w=0.0, reading=None,
                                 achievable_w=LIVE_MAX_W,
                                 max_hold_s=MAX_HOLD_S, soc=sc, soc_floor=fl)
                     SWEEP_N += 1
@@ -1798,8 +1802,8 @@ check("`soc` and `soc_floor` are REQUIRED keyword-only arguments of decide() "
       "with no default. A default on either is TEST-14's exact shape and would "
       "be the quiet fallback this card exists to remove - and a default on "
       "`soc_floor` would additionally be CFG-03's fourth defaults site",
-      _required_kw == {"batt_w", "cap_w", "reading", "achievable_w", "soc",
-                       "soc_floor"})
+      _required_kw == {"batt_w", "cap_w", "margin_w", "reading",
+                       "achievable_w", "soc", "soc_floor"})
 _req_fields = tuple(f.name for f in dataclasses.fields(PeakDecision)
                     if f.default is dataclasses.MISSING
                     and f.default_factory is dataclasses.MISSING)
@@ -1948,7 +1952,7 @@ check("...and it reaches the action log too", any("FLOOR_LIMITED" in e
                                                   for e in c24.events))
 # A missing SoC sensor, through the same real cycle. `inputs_ok` stops the loop
 # at 0 W in that state, so the guard's own answer is what is graded here.
-d_nosoc = decide(p_hungry, batt_w=0.0, cap_w=CAP_W, reading=None,
+d_nosoc = decide(p_hungry, batt_w=0.0, cap_w=CAP_W, margin_w=0.0, reading=None,
                  achievable_w=LIVE_MAX_W, max_hold_s=MAX_HOLD_S,
                  soc=None, soc_floor=SOC_FLOOR)
 check("decide() handed soc=None answers floor-limited rather than a full "
@@ -2226,7 +2230,7 @@ def replay_loop(window, *, start_soc, soc_floor, usable_wh=REF_PACK_WH):
         next_cycle = t + timedelta(seconds=CYCLE_S)
         cycles += 1
         d = decide(PeakTracker.project(ing.averager, t), batt_w=batt,
-                   cap_w=CAP_W, reading=tracker.peak(t),
+                   cap_w=CAP_W, margin_w=0.0, reading=tracker.peak(t),
                    achievable_w=LOOP_MAX_W, max_hold_s=MAX_HOLD_S, soc=soc,
                    soc_floor=soc_floor, frozen=False)
         dec = compute(prev_w, w - batt, batt, LOOP_TUNING, sat_count=sat,
@@ -5462,7 +5466,7 @@ MUT.append(("blindness() returns None - the degenerate 0 W projection is graded"
 
 # 2. the frozen term dropped from the guard's inputs.
 MUT.append(("frozen=False - SAFETY-18's term dropped",
-            not decide(p_frozen, batt_w=-RAILED, cap_w=CAP_W, reading=None,
+            not decide(p_frozen, batt_w=-RAILED, cap_w=CAP_W, margin_w=0.0, reading=None,
                        achievable_w=LIVE_MAX_W, max_hold_s=MAX_HOLD_S, soc=SOC_OK, soc_floor=SOC_FLOOR,
                        frozen=False).engaged and d_frozen.engaged,
             "part 4: 'THE BACKSTOP: SAFETY-18's runaway ... cut to 0 W'"))
@@ -5533,7 +5537,7 @@ _early = sum(1 for _t in (500.0, 1000.0, 2000.0)
              if discharge_permitted(SOC_FLOOR - 1.0, SOC_FLOOR) is not None
              and not decide(PeakTracker.project(
                  averager_at(_t, 450.0), B + timedelta(seconds=450)),
-                 batt_w=BATT_NOW, cap_w=CAP_W, reading=None,
+                 batt_w=BATT_NOW, cap_w=CAP_W, margin_w=0.0, reading=None,
                  achievable_w=LIVE_MAX_W, max_hold_s=MAX_HOLD_S,
                  soc=SOC_FLOOR - 1.0, soc_floor=SOC_FLOOR).engaged)
 MUT.append((f"the SoC gate moved above the `wanted > 0` test ({_early} quiet "
@@ -5963,7 +5967,7 @@ print("13. SAFETY-38 round 4: the ORDERED PAIR inside PeakDecision.explain()")
 _S38_PROJ = PeakTracker.project(averager_at(CAP_W - 1200.0, 450.0),
                                 B + timedelta(seconds=450))
 _S38_CEIL_W, _S38_CEIL_FROM = quarter_ceiling_w(CAP_W, obs_hi)
-_S38_IDLE = decide(_S38_PROJ, batt_w=0.0, cap_w=CAP_W, reading=obs_hi,
+_S38_IDLE = decide(_S38_PROJ, batt_w=0.0, cap_w=CAP_W, margin_w=0.0, reading=obs_hi,
                    achievable_w=LIVE_MAX_W, max_hold_s=MAX_HOLD_S,
                    soc=SOC_OK, soc_floor=SOC_FLOOR)
 # UX-08 2d: the row is in plain words now, and the pair is still ordered.
@@ -6042,7 +6046,7 @@ def _proj_at(net_w, elapsed):
 
 def _decide_on(projection, *, batt_w, soc=SOC_OK, cap_w=CAP_W,
                achievable_w=LIVE_MAX_W, reading=None, frozen=False):
-    return decide(projection, batt_w=batt_w, cap_w=cap_w, reading=reading,
+    return decide(projection, batt_w=batt_w, cap_w=cap_w, margin_w=0.0, reading=reading,
                   achievable_w=achievable_w, soc=soc, soc_floor=SOC_FLOOR,
                   max_hold_s=MAX_HOLD_S, frozen=frozen)
 

goodwe_controller/tools/replay_ac3.py

@@ -176,7 +176,7 @@ def replay(paths, *, guarded: bool, cap_w, max_w, max_hold_s, cycle_s,
             next_cycle = t + timedelta(seconds=cycle_s)
             cycles += 1
             d = decide(PeakTracker.project(ing.averager, t), batt_w=batt,
-                       cap_w=cap_w, reading=tracker.peak(t),
+                       cap_w=cap_w, margin_w=0.0, reading=tracker.peak(t),
                        achievable_w=max_w, max_hold_s=max_hold_s, soc=soc,
                        soc_floor=soc_floor, frozen=False)
             if d.engaged:

goodwe_controller/tools/safety79_identity.py

@@ -0,0 +1,158 @@
+#!/usr/bin/env python3
+"""SAFETY-79: with `peak_margin_w: 0`, is every command byte-identical to before?
+
+    python3 tools/safety79_identity.py          # prints one digest per scenario
+
+tools/opt11_identity.py's harness (the REAL Controller, read_inputs + cycle, in
+simulated time) with a P1 meter, so the peak guard runs, and a peak limit the
+house goes over. Each row hashes opt11_identity's fields plus the guard's floor,
+limit and sentence, its charge headroom (what arbitrage is sized to) and the
+maintenance shaping's suspension. The two `_maint` scenarios run a maintenance
+charge in a cheap window with a demand forecast that crosses cap - 50 .. cap,
+so `shaping_suspended()` and `maintenance_charge_floor()` are both reached.
+It uses only the API release/1.0 ed739d4 already had, so
+the same file runs on the old tree (which ignores `peak_margin_w`) and the new.
+test_peak_margin.py part 1 holds the new tree to the digests printed on ed739d4.
+"""
+
+import functools
+import hashlib
+import os
+import sys
+import tempfile
+import time
+from datetime import timedelta
+
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+
+import opt11_identity as rig                               # noqa: E402
+from app import packmode, peakguard                       # noqa: E402
+from app.p1 import make_sample, split_signed              # noqa: E402
+from app.store import unit_key                            # noqa: E402
+
+CAP_W = 2500
+
+
+def load(t):
+    """A house under the limit, over it, a solar surplus, far over, over."""
+    if t < 300:
+        return 800.0
+    if t < 900:
+        return 3200.0
+    if t < 1500:
+        return -2600.0
+    if t < 2100:
+        return 6000.0
+    return 4200.0
+
+
+def soc(u, t):
+    return 60.0         # above the peak reserve: the guard may discharge
+
+
+def _feed(c, t):
+    """One P1 telegram per cycle: the meter reads what the house and the
+    batteries' measured power leave."""
+    imp, exp = split_signed(c.hass.grid())
+    c.p1.submit(make_sample("ha_signed", imp, exp, phases=1,
+                            ingest_ts=rig.BASE + timedelta(seconds=t),
+                            ingest_mono=time.monotonic()))
+
+
+def store_off(name):
+    """A store where battery `name` is switched Off (OPT-11)."""
+    s = rig._Store()
+    s.data[unit_key(name, packmode.USE_KEY)] = False
+    return s
+
+
+def forecast(t):
+    """A demand forecast that sweeps 2300-2600 W every 10 minutes: across
+    cap - 50 and the cap, so the margin moves the forecast checks."""
+    return 2300.0 + (t % 600) / 600 * 300
+
+
+def _feed_maint(c, t):
+    """`_feed`, plus the forecast and a cheap price (now under average)."""
+    if not getattr(c.hass, "_s79", False):
+        read = c.hass.number
+        fixed = {"sensor.pnow": 0.05, "sensor.pavg": 0.20}
+
+        async def number(e, invert=False):
+            if e == "sensor.fc":
+                return forecast(c.hass.t)
+            if e in fixed:
+                return fixed[e]
+            return await read(e, invert)
+        c.hass.number, c.hass._s79 = number, True
+    _feed(c, t)
+
+
+def store_charging(n):
+    """A store where the first battery is in its maintenance CHARGE phase."""
+    s = rig._Store()
+    for k in ("phase", "phase_started"):
+        s.data[k if n == 1 else unit_key("inverter-a", k)] = (
+            "charge" if k == "phase" else rig.BASE.isoformat())
+    return s
+
+
+P1 = {"meter_source": "ha_signed", "p1_net_entity": "sensor.net",
+      "meter_phases": 1, "peak_cap_w": CAP_W}
+MODES = {"battery_modes_enabled": True}
+MAINT = {"maintenance_enabled": True, "maintenance_start_hour": 11,
+         "peak_forecast_entity": "sensor.fc", "price_now_entity": "sensor.pnow",
+         "price_avg_entity": "sensor.pavg"}
+
+SCENARIOS = {
+    "n1": dict(n=1),
+    "n2": dict(n=2),
+    # OPT-11: the only battery Off, so only the peak guard makes it give.
+    "n1_off": dict(n=1, modes=True, off="inverter-a"),
+    # OPT-11: B Off, so B covers what A cannot.
+    "n2_off": dict(n=2, modes=True, off="inverter-b"),
+    # A maintenance charge in a cheap window, with a demand forecast.
+    "n1_maint": dict(n=1, maint=True),
+    "n2_maint": dict(n=2, maint=True),
+}
+
+
+def run(n, *, margin=None, modes=False, off=None, maint=False, dur=2700.0):
+    """Rows for one scenario; `margin=None` leaves the key out."""
+    extra = dict(P1, **(MODES if modes else {}), **(MAINT if maint else {}))
+    if margin is not None:
+        extra["peak_margin_w"] = margin
+    guard = []
+
+    def on_cycle(c, t):
+        g = c.peak_guard
+        guard.append((c._maint_shape_suspended, None if g is None else (
+            g.engaged, g.floor_w, g.ceiling_w, g.ceiling_from, g.reason,
+            peakguard.charge_headroom_w(g))))
+    db = os.path.join(tempfile.mkdtemp(prefix="s79-"), "tariff.db")
+    real = rig.Controller
+    rig.Controller = functools.partial(real, peak_db_path=db)
+    try:
+        rows, c = rig.run(n, extra=extra, dur=dur, load=load, soc=soc,
+                          cut=lambda u, t: False,
+                          before=_feed_maint if maint else _feed,
+                          on_cycle=on_cycle,
+                          store=(store_off(off) if off else
+                                 store_charging(n) if maint else None))
+    finally:
+        rig.Controller = real
+    return [r + (g,) for r, g in zip(rows, guard)], c
+
+
+def digest(rows) -> str:
+    return hashlib.sha256(repr(rows).encode()).hexdigest()
+
+
+def digests(margin=None) -> dict:
+    return {k: digest(run(**v, margin=margin)[0]) for k, v in SCENARIOS.items()}
+
+
+if __name__ == "__main__":
+    m = None if len(sys.argv) < 2 else int(sys.argv[1])
+    for k, v in digests(m).items():
+        print(f"{k} {v}")

goodwe_controller/translations/en.yaml

@@ -145,6 +145,9 @@ configuration:
   peak_cap_w:
     name: "Limits · Peak limit (W)"
     description: "Your own number: the 15-minute average power the add-on keeps the grid under. Take it from your grid contract."
+  peak_margin_w:
+    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)"
     description: "Optional. A forecast of this 15-minute period's average power. Leave it empty and there is no peak limit."

goodwe_controller/translations/nl.yaml

@@ -126,6 +126,9 @@ configuration:
   peak_cap_w:
     name: "Grenzen · Piekgrens (W)"
     description: "Uw eigen getal: het gemiddelde vermogen per kwartier waaronder de add-on de afname houdt. Neem het over uit uw netcontract."
+  peak_margin_w:
+    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)"
     description: "Optioneel. Een prognose van het gemiddelde vermogen in het lopende kwartier. Laat u dit leeg, dan is er geen piekgrens."