TEL-01: P1 ingestion, with the derivation and the age the EMS owns

A Belgian P1 meter publishes two UNSIGNED registers, not one signed figure. Until now the add-on asked the installer to bridge that gap with a template sensor, which put the sign convention of the whole control loop in a text box. This moves it into the EMS: net = import - export, derived once, in one place, with a test that fails if anyone inverts it. Two transports behind one contract, chosen by `meter_source`: the HA WebSocket subscribing to the DSMR integration's entities, and MQTT on a configurable topic. Everything downstream reads P1Ingest, so switching is a config edit. `meter_source: off` is the default and keeps the existing meter_entity path, so no installed system changes until it opts in. The other half is the timestamp. Every accepted sample is stamped at ingest with a monotonic clock, `meter_max_age_s` is applied to it, and the age is published as sensor.p1_sample_age_s for the ESP32's stale-input watchdog. That entity is recomputed against the clock every second rather than only when a telegram lands, because HA pushes state only on change: a meter frozen at a constant reading emits nothing and looks, to anything watching the value, exactly like a meter that has died. The age tells them apart. Deliberately absent: any fallback to an inverter-side power figure. The inverter's own AC power correlates 0.998 with battery power and 0.09 with the real meter, so failing over to it means regulating against your own output. A gap stays a gap - a reconnect emits no synthetic sample, and a rejected telegram never resolves to 0 W or refreshes the timestamp. Quarter-hour averages are time-weighted over clock-aligned blocks rather than a mean of samples, so a cadence change cannot bias the capacity-tariff figure, and only offtake is accumulated so a quarter of pure export averages to 0 kW. Per-phase import is kept separately: on an unbalanced three-phase load the phase sum and the connection net are different numbers, and only one of them is billed. test_p1.py: 99 checks, runnable with a bare interpreter and no meter. Includes an end-to-end run of the HA transport against a fake Home Assistant websocket. Stacked on SAFETY-04; nothing here touches control.py. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Du77usMj8XNKNFZGmUiWDa
2026-08-24 21:52:51 +02:00
10 changed files with 103 additions and 2190 deletions
@@ -1,10 +0,0 @@
 # This add-on is deployed to a Linux container. core.autocrlf=true on the
 # authoring box gave the checkout CRLF, so a plain copy shipped CRLF files -
 # run.sh with CRLF is a "bad interpreter" failure, and any hash-based drift
 # check between repo and deployment fails for a reason that has nothing to do
 # with the code. Deploy with:
 #     git -c core.autocrlf=false archive release/1.0 goodwe_controller | tar -x
 # which is how 0.3.0 went out, byte-identical to the blobs.
 * text=auto eol=lf
 *.png binary
 *.gz  binary
@@ -1,115 +1,5 @@
 # Changelog
 ## Unreleased
 **TEL-05.** A fourth `meter_source`, `homewizard_local`, which polls a
 HomeWizard P1's **own local API** (`GET /api/v1/data`) instead of watching a
 Home Assistant entity. Set `p1_host` to the meter's address; `meter_poll_s`
 (default 5 s, the meter's own update rate) sets the cadence.
 ✅ **This is the first transport whose `sensor.p1_sample_age_s` measures when
 the meter *reported*, and therefore the first one a firmware watchdog may
 threshold.** Every HTTP response is an arrival: the meter answered, now, with
 its current reading, and whether the *number* moved is not consulted. Home
 Assistant cannot express that at all — a repeated reading emits no
 `state_changed`, advances `last_reported` on neither serialiser, and
 `state_reported` is not subscribable ("Event filter is required"). On
 `ha_signed` that made a healthy meter under a flat load indistinguishable from
 a dead one, and our own capture of this house's meter goes 42.2 s and 97.0 s
 between changes — both past the default `meter_max_age_s` of 30, i.e. a false
 trip to 0 W on a meter that is fine. If you have a HomeWizard P1, move to this
 mode.
 Everything TEL-01 established is reused, not re-implemented: ingest
 timestamping, `meter_max_age_s`, the clock-recomputed age, the plausibility
 bounds and the §20 unsigned-decode rejection, and the same `split_signed` sign
 convention `ha_signed` uses. A failed or timed-out poll submits nothing, so it
 is a *missing* reading — never 0 W — and it does not reset the age.
 Verified on the ENV-01 rig against `sim/hwsim.py`, steady and with `--fault
 freeze` injected. Still defaults to `off`.
 ⚠️ **An arrival stamp still cannot see a *frozen* meter**, and no arrival
 detector can: a meter answering `200 OK` forever with a stale number is
 arriving. The local API does expose what the HA path never had — the
 `total_power_*_kwh` registers stop advancing — and the transport tracks it as
 `unchanged_s`, but that is deliberately **not** folded into the age and not
 thresholded: this controller regulates grid power toward ~0 W, and at a
 converged −10 W the export register needs six minutes to move by its 1 Wh
 resolution while the power figure legitimately repeats. Thresholding it at 30 s
 would rebuild the false-trip limit cycle at the exact operating point we aim
 for. Freeze detection needs the low-power case solved first, separately. It is
 shown on the status page's P1 line instead — leaving it unthresholded only
 holds up if a human can read it, so now they can.
 **TEL-04.** A third `meter_source`, `ha_signed`, reading **one signed** Home
 Assistant entity: positive = import, negative = export. That is the shape a
 HomeWizard P1 publishes (`sensor.p1_meter_active_power`), and it is the meter
 actually fitted here - which neither TEL-01 transport can read, because
 `ha_dsmr` needs two unsigned registers and refuses a negative one, i.e. every
 exporting telegram. Set `p1_net_entity`, and `p1_phase_net_entities` for the
 per-phase capacity-tariff figures on a three-phase connection.
 Everything TEL-01 established is inherited rather than re-implemented - the
 new transport is a subclass of the `ha_dsmr` one overriding only which
 entities it wants and how they become a sample. So ingest timestamping,
 `meter_max_age_s`, `sensor.p1_sample_age_s` recomputed against the clock and
 republished once a second, the plausibility bounds, and `unavailable` /
 `unknown` treated as a *missing reading and never 0 W* all behave identically
 across the three sources.
 Still defaults to `off`; an existing install is unaffected until it opts in.
 ⚠️ **`sensor.p1_sample_age_s` is published on `ha_signed`, but must not yet be
 thresholded by the ESP32 stale-input watchdog.** On the HA WebSocket paths the
 age is stamped from `state_changed`, so it measures time since the value
 *changed*, not since the meter *reported* - and Home Assistant exposes no
 arrival signal for a repeated reading (no `state_changed`, no `last_reported`
 movement on either serialiser, and `state_reported` is not subscribable over
 the WebSocket). Measured on the ENV-01 rig against the real HomeWizard
 integration. `ha_dsmr` mostly escapes it because a telegram moves several
 entities at once; `ha_signed` has one, so a healthy meter under a flat load is
 indistinguishable from a dead one. Our own capture has the house meter going
 42.2 s and 97.0 s between changes. Raising `meter_max_age_s` does not fix that,
 it only chooses which error you get; the fix is an arrival stamp from the meter
 itself and is a separate ticket. Full detail in DOCS.md.
 ## 0.3.0
 **SAFETY-04.** The control law's integrator is now an explicit accumulator,
 bounded independently of the output clamp instead of inheriting whatever
 headroom the clamp happened to leave. It also freezes while the inverter is
 not tracking, rather than continuing to wind up against a command nothing is
 acting on. `integrator_max_w` (default `0`) governs the bound; `0` means
 "follow `max_w`", which is the existing behaviour.
 Behaviour is unchanged at the defaults - a 4,928-case equivalence sweep
 against the previous control law confirms it decides identically at
 `integrator_max_w: 0`.
 **TEL-01.** P1 meter ingestion, so a Belgian P1's two unsigned registers
 (consumption, injection) no longer need a hand-written signed template
 sensor: the subtraction moves into the add-on, done once and tested. Two
 transports, chosen with the new `meter_source` option: `ha_dsmr` subscribes
 to the DSMR integration over the HA WebSocket, `mqtt_p1` reads a topic.
 Defaults to `off`, which keeps the existing `meter_entity` path untouched -
 nothing changes for an install that does not opt in.
 Enabling it publishes `sensor.p1_sample_age_s`: seconds since the newest
 accepted telegram, recomputed against the clock and republished roughly once
 a second rather than only when a telegram lands. That is deliberate - Home
 Assistant only pushes a state on change, so a meter sitting at a genuinely
 constant reading would otherwise look identical to a dead one. Watching the
 age instead means a frozen meter shows a climbing age, not a flat line. The
 firmware watchdog subscribes to this exact entity id.
 Known limits, both already in DOCS.md: on `mqtt_p1`, a bridge stuck
 republishing its last telegram still "arrives", so the age cannot detect
 that particular failure - prefer `ha_dsmr` where both are available. And a
 dead P1 meter takes 45 s to reach 0 W commanded (30 s for `meter_max_age_s`
 to call the reading stale, then 15 s of `stale_input_s` on top), which is
 `meter_max_age_s` and `stale_input_s` stacking, not either one alone.
 ## 0.2.1
 `target_grid_w` (default -10 W): what the meter should rest at. The deadband
@@ -51,84 +51,21 @@ phase having charged nothing.
 ### P1 meter ingestion
 `meter_entity` above expects one signed sensor, which usually means a template
-someone wrote by hand. Setting `meter_source` moves the whole derivation into
+someone wrote by hand. A Belgian P1 meter does not publish one: it publishes two
-the add-on, where it is done once and tested, and replaces `meter_entity`
+**unsigned** registers, consumption and injection. Setting `meter_source` moves
-entirely.
+that subtraction into the add-on, where it is done once and tested, and replaces
-
+`meter_entity` entirely.
 Which mode you want depends on what your P1 reader publishes, and there are two
 shapes in the wild:
 - **Two unsigned registers**, consumption and injection, which is what a Belgian
  P1 read over DSMR gives you → `ha_dsmr`, or `mqtt_p1` for a bridge. The add-on
  subtracts them.
 - **One signed figure**, positive = import and negative = export, which is what
  a HomeWizard P1 gives you (`sensor.p1_meter_active_power`) → `ha_signed`. The
  add-on splits it. `ha_dsmr` **cannot** read this: it wants two registers and
  rejects a negative one outright, which is every exporting telegram.
 Either way, do not build the missing shape out of template sensors. The point of
 `meter_source` is that the sign convention is derived in one tested place rather
 than in YAML nobody reviews underneath a safety input.
 > ✅ **If your meter is a HomeWizard P1, use `homewizard_local`, not
 > `ha_signed`.** It reads the same meter and produces the same numbers, but it
 > polls the meter directly instead of watching a Home Assistant entity — and
 > that is the difference between an age sensor a watchdog can threshold and one
 > it cannot. See "`sensor.p1_sample_age_s`" below; `ha_signed` remains for
 > installs where the meter is only reachable through Home Assistant.
 > ⚠️ **`ha_dsmr` and `mqtt_p1` have never processed a telegram from real
 > hardware.** No meter in this installation uses either one. Both were written
 > to the assumption in `specs.md` §5.2 that a Belgian P1 exposes two unsigned
 > registers, and the meter actually fitted here does not — it is the HomeWizard
 > P1 that `ha_signed` reads. They are covered by the unit checks in `test_p1.py`
 > and by an end-to-end test against a fake Home Assistant, and nothing more.
 >
 > This is recorded because the realistic way it bites is someone debugging a
 > meter problem months from now treating those two paths as proven and looking
 > for the fault elsewhere. If you are the first person to point one at a real
 > meter, expect to find something, and please update this note when you do.
 | option | default | meaning |
 |---|---|---|
-| `meter_source` | `off` | `off` keeps `meter_entity`. `ha_dsmr` subscribes to the DSMR integration over the HA WebSocket; `mqtt_p1` reads a topic; `ha_signed` subscribes to one signed entity over the HA WebSocket; `homewizard_local` polls a HomeWizard P1's own local API, bypassing Home Assistant |
+| `meter_source` | `off` | `off` keeps `meter_entity`. `ha_dsmr` subscribes to the DSMR integration over the HA WebSocket; `mqtt_p1` reads a topic |
 | `meter_phases` | 1 | 1 or 3. Must match the telegram, or every telegram is rejected and logged |
-| `meter_max_age_s` | 30 | Beyond this the reading is stale and grid power reads as *missing*. On its own it does **not** command 0 W — see the timing note below. It is also the longest a reading is held forward into the 15-minute average |
+| `meter_max_age_s` | 30 | Beyond this the reading is stale: grid power reads as *missing*, and the existing failsafe commands 0 W |
 | `meter_poll_s` | 5 | `homewizard_local` only. Seconds between polls. **Must be well under `meter_max_age_s`** — see below |
 | `p1_host` | | `homewizard_local` only. The meter's own address, `host` or `host:port` (e.g. `192.168.2.250`) |
 | `meter_mqtt_topic` | | `mqtt_p1` only |
 | `p1_import_entity` | | The **unsigned** consumption sensor. Do not point this at a signed template |
 | `p1_export_entity` | | The **unsigned** injection sensor |
 | `p1_phase_import_entities` | `[]` | L1..L3, in order. Needed for the capacity-tariff peak on a three-phase connection |
 | `p1_phase_export_entities` | `[]` | L1..L3, in order |
 | `p1_net_entity` | | `ha_signed` only. The **signed** net-power sensor: `+` import, `-` export |
 | `p1_phase_net_entities` | `[]` | `ha_signed` only. L1..L3, in order, each signed the same way. Needed for the capacity-tariff peak on a three-phase connection. **The list length must equal `meter_phases`** |
 Both per-phase lists are checked against `meter_phases` **once at startup**: a
 list of the wrong length disables P1 ingestion with an error in the log, rather
 than letting every telegram fail its phase-count check one at a time. Leaving
 the list empty is fine and is not an error — you simply get no per-phase
 figures, and therefore no capacity-tariff peak. On a three-phase connection
 that is a much bigger omission than it looks: on a surveyed reading here the
 phases carried 2769 W of import while the connection netted 187 W, so the
 billed quantity is understated roughly fifteenfold if the phases are missing.
 #### How long a dead meter takes to reach 0 W
 `meter_max_age_s` and `stale_input_s` **stack**. They are two different clocks
 and neither one is the whole answer:
 | step | option | default |
 |---|---|---|
 | telegrams stop, P1 sample goes stale, grid power starts reading *missing* | `meter_max_age_s` | 30 s |
 | inputs have been missing long enough for the loop to command 0 W | `stale_input_s` | 15 s |
 | **total, meter death → 0 W commanded by this add-on** | | **45 s** |
 So in P1 mode `stale_input_s` is *not* "how long inputs may be missing before
 commanding 0 W" measured from the meter dying — it is measured from the moment
 the P1 sample already went stale. Size the pair together: the ESP32's own
 watchdog commands 0 W after ~30 s of silence from this add-on regardless, and
 that layer is unaffected by either option.
 There is **no fallback to an inverter-side power figure**, deliberately. The
 inverter's own AC power tracks its battery almost perfectly and the real meter
@@ -152,55 +89,6 @@ tested, and guessing a key here means guessing a kilowatt:
 it is present it is used for the age, which is what stops a retained message
 replayed on reconnect from presenting a ten-minute-old reading as current.
 #### `homewizard_local` — polling the meter instead of Home Assistant
 Set `p1_host` to the meter's address and the add-on does `GET /api/v1/data` on
 it every `meter_poll_s` seconds, reading `active_power_w` (signed, same
 convention as `ha_signed`) and the three `active_power_l{1,2,3}_w` fields. Home
 Assistant is not involved: no entity, no WebSocket, no integration to
 mis-configure. Per-phase figures are used only when the meter serves all
 `meter_phases` of them — a single-phase meter returns `null` for L2/L3, and the
 connection-level reading is still accepted on its own.
 **Why this mode exists:** every HTTP response is an *arrival*. The meter
 answered, now, with its current reading — whether or not the number moved. That
 is the signal `sensor.p1_sample_age_s` needs and the one Home Assistant cannot
 give it at all (see the note below). It is also simply fewer moving parts: the
 five-second cadence is the meter's own, rather than an integration's polling of
 it re-published as a state change.
 **Cadence.** The age is never fresher than the poll interval, so:
 | | |
 |---|---|
 | meter's own update rate | ~5.0 s (measured 4.97 s) |
 | `meter_poll_s` default | 5 s — nothing to gain below the meter's own rate |
 | `meter_max_age_s` default | 30 s, i.e. six polls of headroom |
 | `meter_poll_s >= meter_max_age_s` | **refused at startup** — every reading would be stale before its successor arrived |
 | `meter_poll_s > meter_max_age_s / 2` | warned — one missed poll makes the reading stale |
 A failed poll — timeout, connection refused, non-200, unparseable body — is a
 **missing** reading. It submits nothing, so the reading does not become 0 W, the
 last good value and its timestamp are left alone, and the age goes on climbing.
 That is exactly what a dead meter should look like.
 **What it still cannot see: a frozen meter.** A meter that answers `200 OK`
 forever with a stale number is arriving, so no arrival detector — this one
 included — can tell it from a healthy one. The local API does expose the raw
 material the HA path never had (the `total_power_*_kwh` registers stop
 advancing), and the transport tracks it as `unchanged_s`, but it is deliberately
 *not* folded into the age and *not* thresholded: this controller regulates grid
 power toward ~0 W, and at a converged −10 W the export register needs six
 minutes to move by its 1 Wh resolution while the power figure legitimately
 repeats. Thresholding that at 30 s would rebuild the false-trip limit cycle at
 the exact operating point the controller aims for. Freeze detection is a
 separate problem and needs the low-power case solved first.
 You read it yourself instead: the add-on's status page shows it on the P1 line,
 as `… 4 rejected, measurement unchanged for 312 s`. On a house drawing real
 power that figure stays in the seconds; minutes of it while the load is clearly
 not near zero is the meter to go and look at.
 #### `sensor.p1_sample_age_s`
 Published over MQTT discovery whenever a broker is available: **seconds since the
@@ -214,53 +102,6 @@ emits nothing, which is indistinguishable — to anything watching the value —
 a meter that has died. Watching the age instead separates the two: it climbs when
 telegrams stop and resets when they arrive, whatever the reading says.
 The entity is only created when `meter_source` is not `off`. With P1 ingestion
 disabled there is nothing feeding it, and an age sensor climbing with no ingester
 behind it would trip the firmware watchdog on a system that is working fine.
 > **Known limit, `mqtt_p1`.** The age measures *arrival*. On the MQTT path a
 > bridge that is stuck republishing its last telegram keeps arriving, so the age
 > stays near zero and a frozen meter still looks fresh. Detecting *that* needs a
 > change-detector rather than an arrival-detector, and it is not in this version.
 > ⚠️ **Known limit, `ha_signed` — do not drive a watchdog off this age yet.**
 > On the HA WebSocket paths the age is stamped when a `state_changed` arrives,
 > which means it measures *time since the value last changed*, not time since
 > the meter last reported. Home Assistant offers nothing better: a repeated
 > reading produces no `state_changed`, does **not** advance `last_reported` on
 > either the REST or the WebSocket serialiser, and `state_reported` cannot be
 > subscribed to over the WebSocket at all (`Event filter is required for event
 > state_reported`). All three measured on the ENV-01 rig against the real
 > HomeWizard integration with the meter frozen: 0 `state_changed` in 70 s and no
 > timestamp movement anywhere.
 >
 > `ha_dsmr` mostly escapes this because a DSMR telegram updates several entities
 > and something in the set almost always moves. **`ha_signed` has exactly one
 > entity, so a healthy meter under a flat load is indistinguishable from a dead
 > one.** This is not hypothetical: in our own captures
 > (`sim/scenarios/ha-p1_meter_active_power-2026-08-20.json`) the real house meter
 > went **42.2 s and 97.0 s** between changes, and 23 Aug peaks at 29.1 s — all
 > past the default `meter_max_age_s` of 30.
 >
 > So `sensor.p1_sample_age_s` on `ha_signed` is safe to *read*, and it is
 > correct whenever the value is moving, but it must not yet be thresholded by
 > the ESP32 stale-input watchdog: a quiet house would trip the battery to 0 W.
 > Raising `meter_max_age_s` is **not** the fix — the two conditions produce an
 > identical signal, so a bigger number only chooses which of the two errors you
 > get. The real fix is an arrival stamp the meter itself provides — and that now
 > exists: **`meter_source: homewizard_local`**. If you have a HomeWizard P1,
 > switch to it. If your meter is only reachable through Home Assistant, this
 > limit still applies to you and the watchdog threshold still must not be armed.
 **Where the age is trustworthy:**
 | mode | the age measures | safe to threshold from firmware |
 |---|---|---|
 | `homewizard_local` | time since the meter **answered** | **yes** — every HTTP response is an arrival |
 | `ha_dsmr` | time since one of several entities changed | no — statistically usually fine, which is a masked bug, not an absent one |
 | `ha_signed` | time since the one entity changed | **no** — see above |
 | `mqtt_p1` | time since a message arrived | arrivals yes, but a stuck bridge republishing keeps arriving |
 ### Control
 | option | default | meaning |
@@ -272,35 +113,12 @@ behind it would trip the firmware watchdog on a system that is working fine.
 | `target_grid_w` | -10 | What the meter should rest at. Negative = a slight export |
 | `step_w` | 10 | Quantisation |
 | `saturation_w` | 500 | Divergence that counts as "the inverter is at a limit" |
-| `saturation_cycles` | 3 | How many consecutive cycles before freezing. A cycle is one *changed* meter reading, not a fixed period - see the note below. **Do not set to 1** |
+| `saturation_cycles` | 3 | How many consecutive cycles before freezing. **Do not set to 1** |
-| `integrator_max_w` | 0 | Bound on the loop's accumulator, and 0 means "same as `max_w`". Caps how much stale error can be waiting to unwind when the sign flips. **Do not raise it above `max_w`** - the output clamp already bounds what is commanded, so the only thing extra headroom buys is more cycles of wrong-direction power after every saturation event. Lowering it below `max_w` is the useful direction |
+| `integrator_max_w` | 3000 | Bound on the loop's accumulator, separate from `max_w`. Caps how much stale error can be waiting to unwind when the sign flips. **Keep it above `max_w`, and do not set it equal to `max_w`** |
 | `heartbeat_s` | 10 | Refresh interval; must stay well under the firmware watchdog |
-| `stale_input_s` | 15 | How long inputs may be missing before commanding 0 W. In P1 mode this clock starts only *after* `meter_max_age_s` has already expired — the two stack, see "How long a dead meter takes to reach 0 W" |
+| `stale_input_s` | 15 | How long inputs may be missing before commanding 0 W |
 | `auto_start` | false | Start controlling on boot (only after commissioning) |
 #### Saturation is counted in cycles, not seconds
 The specification states the saturation window as **"> 10 s"**. This add-on counts
 **cycles** instead, and that is a deliberate, accepted deviation rather than an
 oversight - the acceptance criterion is not met as literally written.
 A cycle here is one *changed* meter reading: the controller only runs the loop when the
 meter value differs from the previous poll. At the reference P1's ~5 s update rate the
 default of 3 cycles is usually around 15 s, but there is **no guaranteed wall-clock
 window** - a meter that repeats the same value stalls the counter for as long as it
 repeats.
 Two reasons that is acceptable:
 - the control law is a pure function with no clock, which is what makes it testable
  without hardware, and a seconds-based window would have to live in the controller;
 - a stalled counter is a detection-latency limit and not a runaway risk. The condition
  that stalls it - an unchanging meter - stops the whole loop, so nothing accumulates
  while it is stalled.
 If a guaranteed window matters on your site, raise `saturation_cycles` for a fast meter,
 and treat the figure as "N meter updates" rather than "N seconds".
 #### Why `target_grid_w` is not zero
 The deadband is a one-way ratchet: any resting point inside it holds until
@@ -28,22 +28,16 @@ class Tuning:
    step_w: int = 10
    saturation_w: float = 500.0
    saturation_cycles: int = 3
-    # The integrator's own bound. None means "follow max_w", which is the
+    # ⚠️ The integrator's OWN bound, and deliberately not max_w. A commercial
-    # default and the recommended setting.
+    # controller on this same site clamped only its output and still reported
-    #
+    # 14 768 W: with the inverter switched off its integrator climbed ~130 W
-    # ⚠️ DO NOT RAISE THIS ABOVE max_w without a measurement to justify it.
+    # every 4 s past 10 kW while the output sat on the 5 kW rail, so the moment
-    # Every watt of integrator above the rail is a watt of wind that has to be
+    # the error flipped there were minutes of accumulated wind to burn off
-    # burned off before the command can start moving the other way, i.e. extra
+    # before the command moved at all. Bounding the accumulator is what makes
-    # cycles of discharge into an already-exporting meter after every
+    # recovery time finite; bounding the output only hides it.
-    # saturation event. Measured on the closed-loop sim, 4000 W load dropped to
+    # Headroom above max_w is wanted (a legitimate large error must not be
-    # 0: at integrator_max_w == max_w the command is 1000 W two cycles later; at
+    # truncated at the rail), headroom without limit is the bug.
-    # 1.5x max_w it is 1800 W. The output clamp already bounds what reaches the
+    integrator_max_w: float = 3000.0
    # wire, so headroom here buys nothing but unwind latency.
    #
    # It is a separate key because it has to be able to be SMALLER than max_w,
    # which is the only direction that buys anything: it caps unwind latency
    # below what the rail implies. Merging it into max_w would take that away.
    integrator_max_w: float | None = None
    # What the meter should rest at, in W. Negative = a slight export.
    # ⚠️ The deadband is a one-way ratchet: any resting point inside it holds
    # forever, and the meter's IMPORT register counts every positive one with
@@ -60,9 +54,9 @@ class Decision:
    sat_count: int
    frozen: bool
    reason: str
-    # The integrator AFTER this cycle, before the output clamp, the slew limit
+    # The integrator AFTER this cycle, pre-clamp-to-max_w. Carry it back in as
-    # and quantisation. Carry it back in as `i_w` next cycle; that is what keeps
+    # `i_w` next cycle; that is what keeps it a separate quantity from the
-    # it a separate quantity from the command.
+    # command, which is the whole point of the bound above.
    i_w: float = 0.0
@@ -95,19 +89,14 @@ def compute(
    # lets slew be larger than saturation_w.
    #
    # The spec states this window twice and differently: "> 10 s" (§11.2) and
-    # "3 samples" (§10.3). This counts CYCLES, and a cycle is not a unit of
+    # "3 samples" (§10.3). Cycles are authoritative here because this function
-    # time: run_control() calls cycle() only when the meter value CHANGES
+    # has no clock - it is driven one cycle per meter update by run_control(),
-    # (`if self.grid != last_grid`), so three cycles is three distinct meter
+    # which only calls cycle() when the meter value changes. At the ~5 s
-    # readings and nothing more. At the reference P1's ~5 s update rate that is
+    # HomeWizard P1 cadence the default 3 cycles is ~15 s, i.e. the stricter
-    # usually ~15 s, but there is no upper bound on it - a meter that repeats a
+    # reading of the two. On a faster meter it is not, so saturation_cycles is
-    # value stalls the counter.
+    # configurable and must be raised to keep the window over 10 s.
-    #
+    # ponytail: a seconds-based window would mean plumbing wall-clock or dt
-    # That is a detection-latency limit, not a windup hazard: the same
+    # into a pure function whose whole value is that it has neither.
    # condition that stalls the counter stalls the whole loop, so nothing
    # accumulates in the meantime either. If a wall-clock window is ever
    # required, it belongs in Controller (which has a clock) and not here.
    # ponytail: this function is worth keeping clockless; the ceiling is that
    # saturation_cycles cannot express a guaranteed number of seconds.
    saturated_now = abs(prev_w - actual_w) > tuning.saturation_w
    sat_count = min(sat_count + 1, 10) if saturated_now else 0
    frozen = sat_count >= tuning.saturation_cycles
@@ -123,78 +112,32 @@ def compute(
    # --- the integrator ----------------------------------------------------
    # This loop is in velocity form: the accumulator IS the commanded power, so
-    # "the integrator" and "the output" were one variable and could not be
+    # for years "the integrator" and "the output" were one variable and could
-    # bounded apart. `i_w` is that accumulator made explicit; main.py carries it
+    # not be bounded apart. `i_w` is that accumulator made explicit. A caller
-    # between cycles, which is what turns the two clamps into two limits.
+    # that passes nothing gets the old behaviour exactly - seeded from the last
-    #
+    # command every cycle - and main.py carries it instead, which is what turns
-    # Passing i_w=None re-seeds it from the last command every cycle. With
+    # the two clamps below into two independent limits.
    # integrator_max_w following max_w that reduces this function to the exact
    # velocity form it replaced, frozen branch included - asserted by an
    # exhaustive comparison against a transcription of the old law in
    # test_control.py, not by inspection. Break either the gate or the bound
    # below and that test is what tells you the equivalence went with it.
    if i_w is None:
        i_w = float(prev_w)
    limit = tuning.max_w if tuning.integrator_max_w is None else tuning.integrator_max_w
    if abs(error) < tuning.deadband_w:
        reason = "deadband"
    else:
-        moved = i_w + tuning.gain * error
+        step_i = tuning.gain * error
-        # ⚠️ Freeze means "may not wind FURTHER in the direction it is already
+        # ⚠️ Freeze means "may not wind FURTHER", not "may not move". A strict
-        # pushing". It may fall, cross zero, or reverse outright.
+        # freeze would strand the command at whatever it had reached until the
-        #
+        # inverter started tracking again - and the inverter is not tracking,
-        # It must NOT be encoded as "only corrections that shrink |i_w|": that
+        # that is what saturation means, so nothing would ever release it. The
-        # is unsatisfiable for BOTH signs of error whenever the correction is
+        # unwind direction is the escape route and stays open; the same rule is
-        # larger than twice the integrator, i.e. every time the integrator is
+        # applied again to the output below.
-        # near zero. The loop then sits at its last value forever, because what
+        if not frozen or abs(i_w + step_i) < abs(i_w):
-        # clears the freeze is the inverter tracking again and not-tracking is
+            i_w = i_w + step_i
        # the definition of saturation. Measured on that encoding: 0 W held
        # indefinitely into a 2 kW import, where this form recovers next cycle.
        #
        # This is the same asymmetric rule the output freeze uses below, which
        # has been in service on real hardware. It is applied here as well
        # because the requirement is that the INTEGRATOR stop accumulating, not
        # only the command.
        #
        # ⚠️ EXACTLY ZERO IS ITS OWN CASE, and it must be handled explicitly
        # rather than falling into one of the two branches. "May not wind
        # further in the direction it is already pushing" has no referent at
        # zero: nothing is wound, and neither direction is "further". Writing
        # this as `if i_w > 0 ... else ...` silently files zero under
        # rising-only and permanently blocks the first push toward charging -
        # the same deadlock as the shrink-only encoding above, mirrored in sign,
        # and reachable because main.py resets i_w to exactly 0.0 on every stop
        # and every reseed. Measured before the fix: 12 800 of 25 920 frozen
        # ticks at i_w == 0.0 held the integrator, 8 304 of them changing the
        # emitted command, worst case abandoning a 2 kW charge into a 4 kW
        # export.
        #
        # Freezing at zero would also be pointless: the freeze exists to stop
        # accumulation running away, and a first step from zero is bounded by
        # the gain, the output clamp and the slew limit like any other.
        if not frozen or i_w == 0.0:
            i_w = moved
        elif i_w > 0:
            i_w = min(moved, i_w)
        else:
            i_w = max(moved, i_w)
-    # ⚠️ Applied EVERY cycle, frozen or not: the freeze is conditional, this
+    # ⚠️ Applied EVERY cycle, frozen or not, and before the output clamp: the
-    # bound is not. It is what makes the worst-case unwind time finite and
+    # freeze is conditional, this bound is not. Order matters only in that the
-    # knowable instead of a function of how long the error happened to stand.
+    # command below is derived from the already-bounded integrator, so no
-    bounded = max(-limit, min(limit, i_w))
+    # accumulated value can reach the wire even once.
-    if bounded != i_w:
+    i_w = max(-tuning.integrator_max_w, min(tuning.integrator_max_w, i_w))
        # ⚠️ SAFETY-03 (alarm whenever the loop winds into a rail) must watch
        # for THIS, not for "clamped" below. At the default limit == max_w the
        # integrator bound is reached first and the command derived from it can
        # then never exceed max_w, so "clamped" is unreachable on a default
        # install - it survives only for a configuration that deliberately lets
        # the integrator run above the rail. Two reasons rather than one
        # because the two events want different alarms: "i-clamped" is the loop
        # winding, "clamped" is a command that came out over the rating anyway.
        reason = "i-clamped"
    i_w = bounded
    want = i_w
    # ⚠️ Maintenance shaping (charge-only, cheap-window floor) used to live
@@ -39,7 +39,7 @@ from .control import Tuning, compute, maintenance_charge_floor, peak_at_risk
 from .hass import HomeAssistant
 from .maintenance import IDLE, MaintConfig, Maintenance
 from .mqtt import MqttPublisher
-from .p1 import P1Ingest, build_source, is_enabled
+from .p1 import P1Ingest, build_source
 from . import web
 OPTIONS_PATH = "/data/options.json"
@@ -71,11 +71,7 @@ class Controller:
            step_w=int(opts.get("step_w", 10)),
            saturation_w=float(opts.get("saturation_w", 500)),
            saturation_cycles=int(opts.get("saturation_cycles", 3)),
-            # 0 / unset means "follow max_w", which is the recommended
+            integrator_max_w=float(opts.get("integrator_max_w", 3000)),
            # value. Read the note in control.py before raising it above
            # max_w: every watt above the rail is unwind latency.
            integrator_max_w=(float(opts["integrator_max_w"])
                              if opts.get("integrator_max_w") else None),
        )
        self.maint = Maintenance(
            MaintConfig(
@@ -96,11 +92,7 @@ class Controller:
        # until it opts in.
        self.p1 = P1Ingest(phases=int(opts.get("meter_phases", 1)),
                           max_age_s=float(opts.get("meter_max_age_s", 30)))
-        self.p1_enabled = is_enabled(opts)
+        self.p1_enabled = str(opts.get("meter_source", "off")) not in ("off", "")
        # Set by amain() once the transport is built, so the status page can show
        # what only the transport knows (homewizard_local's unchanged_s). Stays
        # None when P1 is off, or under a transport that has no such counter.
        self.p1_source = None
        # live state
        self.auto = bool(store.data.get("auto", opts.get("auto_start", False)))
@@ -328,26 +320,16 @@ class Controller:
            await asyncio.sleep(1)
    def publish(self) -> None:
-        values = {
+        self.mqtt.publish({
            "setpoint": self.target,
            "grid": self.grid,
            "battery": self.batt,
            "soc": self.soc,
            "phase": self.maint.phase,
            "status": "running" if self.auto else "stopped",
        }
        # ⚠️ ONLY when P1 ingestion is actually running. The ESP32's stale-input
        # watchdog subscribes to sensor.p1_sample_age_s and forces the layer-1
        # failsafe once it reaches max_age_s. With meter_source off there is no
        # ingester feeding it, so published_age_s would be time-since-startup
        # climbing without bound - i.e. every existing install would cross the
        # threshold within 30 s and pin its inverter at 0 W forever. Publishing
        # nothing leaves the entity non-existent, which is the status quo and
        # what has_state() in the firmware is checking for.
        if self.p1_enabled:
            # Recomputed here, once a second, on purpose - see P1Ingest.
-            values["p1_age"] = round(self.p1.published_age_s, 1)
+            "p1_age": round(self.p1.published_age_s, 1),
-        self.mqtt.publish(values)
+        })
    async def shutdown(self) -> None:
        """Deterministic wind-down. Do not skip this."""
@@ -369,22 +351,10 @@ class Controller:
                                    + (f" - last error: {self.p1.last_error}"
                                       if self.p1.last_error else "")})
            else:
                # ⚠️ unchanged_s is REPORTED, never thresholded and never folded
                # into the age - see HomeWizardLocalSource.unchanged_s for why
                # (at the converged -10 W this controller aims for, a 1 Wh
                # register needs ~6 minutes to move, so any limit false-trips at
                # the exact operating point we target). The whole argument for
                # leaving it unthresholded is that a human interprets it, which
                # requires a human being able to see it - so here it is. getattr:
                # only homewizard_local has one, and p1_source is None until
                # amain() builds the transport.
                unchanged = getattr(self.p1_source, "unchanged_s", None)
                out.append({"ok": True, "warn": False,
                            "text": f"P1 meter ({o.get('meter_source')}): {self.p1.net_w:g} W, "
                                    f"{age:.0f} s old, {self.p1.samples} telegrams, "
-                                    f"{self.p1.parse_errors} rejected"
+                                    f"{self.p1.parse_errors} rejected"})
                                    + (f", measurement unchanged for {unchanged:.0f} s"
                                       if unchanged is not None else "")})
        rows = [("battery SoC", self.soc, o.get("soc_entity")),
                ("battery power", self.batt, o.get("batt_entity"))]
        if not self.p1_enabled:
@@ -527,7 +497,6 @@ async def amain() -> None:
                broker.get("port", 1883) if broker else 1883,
                broker.get("username") if broker else None,
                broker.get("password") if broker else None,
                omit=() if is_enabled(opts) else ("p1_age",),
            )
        except Exception as err:  # noqa: BLE001
            _LOG.warning("MQTT unavailable (%s) - continuing without status entities", err)
@@ -556,7 +525,6 @@ async def amain() -> None:
        # unit - would be computed from a fraction of the data.
        p1_source = build_source(opts, controller.p1, session, broker)
        if p1_source is not None:
            controller.p1_source = p1_source     # so checks() can report on it
            tasks.append(asyncio.create_task(p1_source.run()))
        await stop.wait()
@@ -60,12 +60,7 @@ AVAILABILITY = f"{BASE}/availability"
 class MqttPublisher:
-    def __init__(self, host, port, username=None, password=None, omit=()):
+    def __init__(self, host, port, username=None, password=None):
        # `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
        # never fed is a worse signal than one that does not exist at all.
        self.omit = set(omit)
        self.enabled = mqtt is not None and bool(host)
        self.client = None
        if not self.enabled:
@@ -99,8 +94,6 @@ class MqttPublisher:
    def _announce(self) -> None:
        for key, object_id, name, unit, dev_class, state_class, icon in SENSORS:
            if key in self.omit:
                continue
            cfg = {
                "name": name,
                "object_id": object_id,
@@ -4,22 +4,13 @@ Everything downstream trusts this module: the safety checks, the capacity-tariff
 peak, the optimizer, the control loop's sign. So three things happen here and
 nowhere else.
-  1. The IMPORT/EXPORT DERIVATION. A Belgian P1 read over DSMR exposes two
+  1. The IMPORT/EXPORT DERIVATION. A Belgian P1 meter exposes two UNSIGNED
-     UNSIGNED registers - consumption and injection. Net power is
+     registers - consumption and injection - never one signed figure. Net power
-     `import_w - export_w`, positive = import, and that subtraction is done
+     is `import_w - export_w`, positive = import, and that subtraction is done
     exactly once, here (spec §5.2: "the derivation is the EMS's job, not a
     template the user has to write"). A second copy of it somewhere else is a
     second chance to invert the control loop.
     Some P1 readers - the HomeWizard P1 among them - publish the OTHER shape:
     one SIGNED figure, positive = import, and no unsigned registers at all.
     `split_signed()` fans that back out into the same two magnitudes, so there
     is still exactly one internal representation and one sign convention. ⚠️ It
     lives here, next to the subtraction, for the same reason the subtraction
     does: the moment a user is asked to write two template sensors that split a
     signed value, the sign convention is back in unreviewed YAML underneath a
     safety input, which is precisely what §5.2 moved into the EMS.
  2. THE INGEST TIMESTAMP. Every accepted sample is stamped on arrival. A value
     with no age is a value that cannot be trusted (§5.2), and staleness is the
     failsafe trigger (§11.2).
@@ -56,8 +47,6 @@ _LOG = logging.getLogger("goodwe.p1")
 SOURCE_HA = "ha_dsmr"
 SOURCE_MQTT = "mqtt_p1"
 SOURCE_HA_SIGNED = "ha_signed"
 SOURCE_HOMEWIZARD = "homewizard_local"
 QUARTER_S = 900
@@ -89,7 +78,7 @@ class P1Sample:
    ingest_ts: datetime            # tz-aware UTC, set at ingest
    ingest_mono: float             # time.monotonic() at ingest - see age_s()
    telegram_ts: datetime | None   # from the telegram, where the source has one
-    source: str                    # one of the SOURCE_* constants above
+    source: str                    # SOURCE_HA | SOURCE_MQTT
    import_w: float                # unsigned magnitude, as the meter reports it
    export_w: float                # unsigned magnitude
    net_w: float                   # import_w - export_w  (+ import, - export)
@@ -140,29 +129,6 @@ def _watts(value, what: str) -> float:
    return out
 def split_signed(net_w) -> tuple[float, float]:
    """One signed figure -> the (import, export) magnitudes the module speaks.
    The inverse of make_sample's subtraction, and the easy direction: no second
    register to disagree with, so there is nothing to mix a fresh reading with a
    stale one. `+` is import, `-` is export - verified in test_p1.py against real
    captured readings from this house's own meter, not against a datasheet.
    ⚠️ Exactly one of the two comes out non-zero. Splitting into `(max(v,0),
    max(-v,0))` rather than clamping keeps `import_w - export_w == v` exactly, so
    the signed value the meter published survives the round trip bit for bit -
    a control loop must not be steered by a number that changed on the way in.
    ⚠️ Validation is `_watts`, the same gate the unsigned path uses: NaN,
    infinity, non-numbers and the §20 open-question-5 unsigned-decode
    contamination (64954 for -582 W) are all refused here rather than believed.
    A signed source makes that check MORE important, not less - on this path
    64954 is not obviously wrong the way a negative "unsigned" register is.
    """
    v = _watts(net_w, "net")
    return (v, 0.0) if v >= 0 else (0.0, -v)
 def make_sample(source: str, import_w, export_w, *, phases: int,
                phase_import_w=None, phase_export_w=None,
                telegram_ts: datetime | None = None,
@@ -247,16 +213,8 @@ class QuarterAverager:
    halves credited to the two blocks, never attributed wholly to either.
    """
-    def __init__(self, phases: int = 1, max_hold_s: float = 30.0):
+    def __init__(self, phases: int = 1):
        self.phases = phases
        # ⚠️ How long one sample may be held forward before the series is
        # treated as a gap rather than a plateau. Without this the meter can die
        # while importing 5 kW, come back ten minutes later, and the hold-forward
        # credits 5 kW x 600 s to the capacity-tariff accumulator - a fabricated
        # peak, on a permanent record, from data that was never measured. Set
        # from meter_max_age_s: the point past which the reading is not trusted
        # for control is the point past which it must not be billed either.
        self.max_hold_s = float(max_hold_s)
        self._block: int | None = None       # epoch seconds of the block start
        self._acc = 0.0                      # W*s of offtake in the open block
        self._pp_acc = [0.0] * phases
@@ -315,22 +273,16 @@ class QuarterAverager:
            return closed
        cursor = self._last_t
        # Beyond this instant the held value stops being evidence of anything.
        # The stretch from here to `t` is walked so the block boundaries are
        # still crossed correctly, but nothing is accumulated and `_elapsed`
        # does not grow - which is what makes a closed block, always divided by
        # the full 900 s, actually get dragged down by the missing coverage.
        hold_end = self._last_t + self.max_hold_s
        while True:
            end = self._block + QUARTER_S
            stop = min(t, end)
-            covered = max(0.0, min(stop, hold_end) - cursor)
+            dt = stop - cursor
-            if covered > 0:
+            if dt > 0:
-                self._acc += max(self._last_net, 0.0) * covered
+                self._acc += max(self._last_net, 0.0) * dt
                if self._last_pp is not None:
                    for i, v in enumerate(self._last_pp[: self.phases]):
-                        self._pp_acc[i] += max(v, 0.0) * covered
+                        self._pp_acc[i] += max(v, 0.0) * dt
-                self._elapsed += covered
+                self._elapsed += dt
            cursor = stop
            if stop < end:
                break
@@ -369,9 +321,7 @@ class P1Ingest:
    def __init__(self, phases: int = 1, max_age_s: float = 30.0):
        self.phases = phases
        self.max_age_s = float(max_age_s)
-        # The same threshold governs control and billing: a reading too old to
+        self.averager = QuarterAverager(phases)
        # steer by is too old to bill by. See QuarterAverager.max_hold_s.
        self.averager = QuarterAverager(phases, max_hold_s=self.max_age_s)
        self.blocks: list[QuarterBlock] = []
        self.samples = 0
        self.parse_errors = 0
@@ -527,17 +477,9 @@ class HaDsmrSource:
                    continue
                payload = json.loads(msg.data)
                if payload.get("id") == 2 and payload.get("type") == "result":
                    # ⚠️ Prime the cache, but do NOT build a sample from it.
                    # get_states returns whatever HA currently holds, which
                    # after a Core restart is a RestoreEntity value of unknown
                    # age. Stamping that with ingest_ts=now resets the age to
                    # zero and reports a fresh meter that may have been dead for
                    # an hour - a synthetic sample hiding the outage from the
                    # watchdog that exists to catch it. The cache is what lets
                    # the FIRST real state_changed build a complete sample; the
                    # age stays honest until one arrives.
                    for obj in payload.get("result") or []:
                        self._absorb(obj.get("entity_id"), obj.get("state"))
                    self._schedule()
                elif payload.get("type") == "event":
                    data = (payload.get("event") or {}).get("data") or {}
                    if data.get("entity_id") not in self.ids:
@@ -702,348 +644,9 @@ class MqttP1Source:
            self.ingest.reject(err)
 # --------------------------------------------------------------------------- #
 # transport 3: Home Assistant WebSocket, one signed entity
 # --------------------------------------------------------------------------- #
 class HaSignedSource(HaDsmrSource):
    """The same websocket, subscribed to ONE signed power entity.
    For readers that publish net power as a single signed figure - a HomeWizard
    P1's `sensor.p1_meter_active_power`, positive = import - rather than the two
    unsigned DSMR registers. This is the meter actually installed at the house,
    and `ha_dsmr` cannot read it: it needs two registers and refuses a negative
    one outright, which is every exporting telegram.
    ⚠️ A subclass, not a copy. The connect / auth / subscribe / reconnect /
    `_absorb` machinery above is transport, not shape, and it has already been
    debugged once - notably "prime the cache from get_states but never build a
    sample out of it" and "a reconnect emits nothing". Only `_wanted` (which
    entity ids) and `build` (how they become a sample) differ, so only those two
    are overridden. Everything TEL-01 established therefore applies unchanged:
    ingest timestamping, meter_max_age_s, the clock-recomputed age sensor, and
    `unavailable` treated as a missing reading rather than 0 W.
    ⚠️ THE AGE ON THIS TRANSPORT MEASURES TIME SINCE THE VALUE CHANGED, not time
    since the meter reported, and on one entity those are very different things.
    Home Assistant offers no arrival signal for a repeated reading: it emits no
    `state_changed`, it does not advance `last_reported` on either serialiser,
    and `state_reported` cannot be subscribed to over the websocket at all
    ("Event filter is required for event state_reported"). All three measured on
    the ENV-01 rig against the real HomeWizard integration with the meter frozen
    - 0 state_changed in 70 s, no timestamp movement anywhere.
    `ha_dsmr` mostly escapes it because a DSMR telegram moves several entities at
    once. This transport has ONE, so a healthy meter under a flat load looks
    exactly like a dead one - and our own capture has the real meter going 42.2 s
    and 97.0 s between changes, both past the default max_age_s of 30. Hence
    DOCS.md: sensor.p1_sample_age_s is correct while the value moves and must not
    yet be thresholded by the firmware watchdog on this transport. Raising
    meter_max_age_s does not fix it, it only chooses which of the two errors you
    get. The fix is an arrival stamp from the meter itself - reading the
    HomeWizard local API rather than an HA entity - which is a separate ticket.
    ⚠️ The debounce is inherited but does nothing useful here, and that is fine:
    one telegram is one entity, so there is no burst of per-entity events to
    coalesce and no window in which a new reading sits beside a stale one. It
    costs one scheduled sleep per telegram at ~0.2 Hz. Left in place rather than
    special-cased, because a second code path through build() is a second place
    for the sign to go wrong.
    """
    def _wanted(self) -> set[str]:
        out = set()
        if self.entities.get("net"):
            out.add(self.entities["net"])
        out.update(e for e in self.entities.get("phase_net") or [] if e)
        return out
    def build(self) -> bool:
        """Assemble one sample from the cache. Returns True if one was accepted."""
        net_id = self.entities.get("net")
        if net_id not in self.cache:
            return False
        pn = [self.cache.get(e) for e in self.entities.get("phase_net") or []]
        if pn and None in pn:
            return False           # incomplete phase set: wait, do not guess
        try:
            imp, exp = split_signed(self.cache[net_id])
            pi = pe = None
            if pn:
                # ponytail: this split is arithmetically redundant today -
                # make_sample subtracts the two lists again and does not
                # sign-check per-phase figures, so handing it the signed values
                # with a zero export list produces the identical tuple. Verified:
                # mutating it that way leaves all 174 checks green, i.e. no test
                # can tell the difference, and it is recorded here rather than
                # left as a silent equivalent mutant for the next reviewer to
                # rediscover. Kept because `phase_import_w` means a MAGNITUDE:
                # a negative in it is the double-signing that make_sample refuses
                # outright for the connection-level registers, and the day that
                # check is extended per-phase the shortcut breaks the meter, not
                # the test.
                pairs = [split_signed(v) for v in pn]
                pi = [a for a, _ in pairs]
                pe = [b for _, b in pairs]
            self.ingest.submit(make_sample(
                SOURCE_HA_SIGNED, imp, exp,
                phases=self.ingest.phases,
                phase_import_w=pi, phase_export_w=pe,
                # ⚠️ No telegram_ts, for the same reason as ha_dsmr: HA's
                # last_changed is when the VALUE changed, which on a steady meter
                # is minutes ago while the telegram is current. sensor.
                # p1_sample_age_s is what covers a genuinely frozen meter.
            ))
            return True
        except P1Error as err:
            self.ingest.reject(err)
            return False
 # --------------------------------------------------------------------------- #
 # transport 4: the HomeWizard P1's own local API
 # --------------------------------------------------------------------------- #
 HOMEWIZARD_PATH = "/api/v1/data"
 HOMEWIZARD_PORT = 80
 def parse_homewizard(doc, phases: int, *, now: datetime | None = None,
                     ingest_mono: float | None = None) -> P1Sample:
    """One `GET /api/v1/data` document -> a sample. Raises P1Error.
    The meter's own JSON, of which we read four keys:
        {"active_power_w": -5710.0,          # SIGNED, + import  - export
         "active_power_l1_w": ..., "active_power_l2_w": ..., "..._l3_w": ...,
         "total_power_import_kwh": 1234.567, "total_power_export_kwh": 890.123}
    Same shape as `ha_signed` - one signed connection figure plus signed
    per-phase figures - so it goes through the same `split_signed` and the same
    `make_sample`. Nothing about the sign convention is re-derived here.
    ⚠️ No telegram timestamp, because the API carries none, and that is correct
    rather than a gap: the document was produced by the meter in the moment it
    answered, so the ingest stamp IS the measurement time. This is the whole
    reason the transport exists - see HomeWizardLocalSource.
    ⚠️ Per-phase figures are taken only when the meter serves ALL `phases` of
    them. A single-phase HomeWizard returns `active_power_l2_w: null`, and a
    partial set must not become a fabricated tuple. Unlike the HA transports
    there is no "wait for the rest" case to worry about: every field here came
    out of ONE response, so a missing phase cannot be a stale phase, and the
    connection-level reading is still a genuine, complete measurement. Dropping
    the whole sample over an absent per-phase field would turn a healthy meter
    into a climbing age, which is the exact false-trip this ticket removes.
    """
    if not isinstance(doc, dict):
        raise P1Error(f"meter response is not a JSON object ({type(doc).__name__})")
    imp, exp = split_signed(doc.get("active_power_w"))
    pi = pe = None
    # ⚠️ range(phases), not the legs the meter served: a 3-phase meter configured
    # as meter_phases: 1 yields a per-phase tuple covering ONE leg of three.
    # Control is unaffected (it uses the connection figure), but the
    # capacity-tariff peak is understated. Filed separately - not fixed here.
    legs = [doc.get(f"active_power_l{i + 1}_w") for i in range(phases)]
    if all(v is not None for v in legs):
        pairs = [split_signed(v) for v in legs]
        pi = [a for a, _ in pairs]
        pe = [b for _, b in pairs]
    return make_sample(SOURCE_HOMEWIZARD, imp, exp, phases=phases,
                       phase_import_w=pi, phase_export_w=pe,
                       ingest_ts=now, ingest_mono=ingest_mono)
 def measurement_fingerprint(doc) -> tuple:
    """The part of a response that a genuinely new measurement has to move.
    Power plus both energy registers. A meter under any real load advances a
    kWh counter (1 Wh resolution: ~10 s at 350 W), so this changes even when the
    power figure happens to repeat. Used ONLY for `unchanged_s` - see there for
    why it must not be allowed anywhere near the age.
    """
    if not isinstance(doc, dict):
        return ()
    return (doc.get("active_power_w"),
            doc.get("total_power_import_kwh"),
            doc.get("total_power_export_kwh"))
 class HomeWizardLocalSource:
    """Polls the meter's own local HTTP API instead of a Home Assistant entity.
    ⚠️ THIS IS THE ONLY TRANSPORT WHOSE AGE MEASURES ARRIVAL. That is the entire
    reason it exists, and it is not an optimisation - it is the difference
    between an age sensor a firmware watchdog may threshold and one it may not.
    Every HTTP response is a genuine arrival: the meter answered, now, with its
    current reading. Whether the NUMBER moved is irrelevant, so a healthy meter
    under a flat load resets the age exactly like a busy one. Home Assistant
    cannot provide this at all - it emits `state_changed` only on a change, does
    not advance `last_reported` on either serialiser for a repeat, and refuses a
    `state_reported` subscription outright ("Event filter is required"). Measured
    on the ENV-01 rig over 70 s of a frozen meter, and independently against the
    live house over ten repeated readings. So on `ha_signed` the age means "time
    since the value changed", and our own capture of this house's meter has
    42.2 s and 97.0 s between changes - both past the default `meter_max_age_s`
    of 30, i.e. a false trip to 0 W on a perfectly healthy meter.
    Everything else is TEL-01's pipeline unchanged: ingest timestamping,
    `meter_max_age_s`, the clock-recomputed age, the plausibility bounds, and a
    failed read treated as a MISSING reading rather than 0 W.
    ⚠️ A failed or timed-out poll submits nothing. It is a rejection, so the
    last good sample and its stamp stay exactly where they were and the age goes
    on climbing - which is precisely the signal a dead meter should produce.
    ⚠️ POLL CADENCE vs `meter_max_age_s`. The age can only be as fresh as the
    poll interval, so `meter_poll_s` must sit well under `meter_max_age_s` or the
    reading flaps in and out of staleness on a healthy meter. The real meter
    updates every ~5.0 s (NOTES.md:640 measures 4.97 s), so polling faster than
    that buys nothing but re-reads. Default 5 s against a 30 s max age; the
    startup check in build_source refuses `meter_poll_s >= meter_max_age_s` and
    warns past half of it.
    ponytail: a plain `asyncio.sleep` loop rather than a scheduler, and no
    backoff. A poll that fails costs one rejection and is retried on the next
    tick; a meter that is down stays down and the age reports it. The ceiling is
    a meter so slow that polls overlap - the request timeout is held under the
    poll interval so they cannot.
    """
    def __init__(self, session, ingest: P1Ingest, host: str, *,
                 port: int = HOMEWIZARD_PORT, poll_s: float = 5.0,
                 timeout_s: float | None = None):
        self.session = session
        self.ingest = ingest
        self.host, self.port = host, int(port)
        self.poll_s = float(poll_s)
        # ⚠️ Held under the poll interval on purpose: a timeout longer than the
        # cadence queues polls behind a hung meter, and a backlog of requests
        # all landing at once would stamp several arrivals for one measurement.
        self.timeout_s = float(timeout_s) if timeout_s else max(1.0, self.poll_s * 0.8)
        self.url = f"http://{self.host}:{self.port}{HOMEWIZARD_PATH}"
        self.connected = False
        self.polls = 0
        self.poll_errors = 0
        self._fingerprint: tuple | None = None
        self._changed_mono: float | None = None
    @property
    def unchanged_s(self) -> float | None:
        """Seconds since the meter last served a DIFFERENT measurement.
        ⚠️ Diagnostic only, and deliberately NOT folded into
        `sensor.p1_sample_age_s`. A frozen meter (SAFETY-01's six-minute
        runaway) answers 200 OK with a stale document forever, so no arrival
        detector can see it - only the document standing still can, and this is
        that signal. It is exposed rather than acted on because it is NOT safe
        as a 30 s watchdog input: this controller regulates grid power toward
        ~0 W, and at a converged -10 W the export register takes six minutes to
        advance by its 1 Wh resolution while the power figure legitimately
        repeats. Thresholding that would rebuild the very limit cycle the
        arrival stamp just removed, at the exact operating point we aim for.
        Whoever wires a freeze detector must handle the low-power case first.
        """
        if self._changed_mono is None:
            return None
        return max(0.0, time.monotonic() - self._changed_mono)
    async def run(self) -> None:
        """Long-lived task: poll, submit, sleep, forever.
        ⚠️ Nothing but cancellation may end this loop. An escaping exception
        would kill the poll task for the lifetime of the add-on, and it would do
        it QUIETLY: the failure mode is safe (no submissions, the age climbs, the
        watchdog holds the battery at 0 W) but it looks identical to a dead
        meter, so the operator goes hunting the wrong device. Retry on the next
        tick instead - poll_s is already the retry cadence, so no backoff.
        """
        _LOG.info("P1 ingest: polling %s every %.1fs", self.url, self.poll_s)
        while True:
            try:
                await self.poll_once()
            except asyncio.CancelledError:
                raise
            except Exception as err:  # noqa: BLE001 - the task must outlive it
                _LOG.warning("P1 meter poll %s: %s - retrying in %.1fs", self.url,
                             str(err) or type(err).__name__, self.poll_s)
            await asyncio.sleep(self.poll_s)
    async def poll_once(self) -> bool:
        """One GET. Returns True if a sample was accepted."""
        self.polls += 1
        try:
            async with self.session.get(
                    self.url,
                    timeout=aiohttp.ClientTimeout(total=self.timeout_s)) as resp:
                if resp.status != 200:
                    raise P1Error(f"meter returned HTTP {resp.status}")
                # content_type=None: the meter's own firmware is the authority on
                # what it serves, and refusing a reading over a Content-Type
                # header would be a fabricated outage.
                # ⚠️ Kept because a real HomeWizard firmware that ever answers
                # text/plain would otherwise read as a dead meter and hold the
                # battery at 0 W. (This was once recorded here as an equivalent
                # mutant - it is not. aiohttp's json_response always sets
                # application/json, but web.Response(text=...) defaults to
                # text/plain, so the fake meter CAN serve valid JSON under the
                # wrong header, and test_p1.py now does.)
                doc = await resp.json(content_type=None)
        except asyncio.CancelledError:
            raise
        except Exception as err:  # noqa: BLE001 - any read failure is a gap
            self.connected = False
            self.poll_errors += 1
            # ⚠️ str() of a bare asyncio.TimeoutError is the EMPTY STRING, so
            # f"...: {err}" renders "last error:" and then nothing - on the
            # status page, on a hung meter, at the moment the battery has just
            # dropped to 0 W and someone is reading that line to find out why.
            # The class name is the only thing that says "it timed out".
            # ⚠️ str(err), not `err or ...`: an exception object is ALWAYS truthy,
            # empty message or not, so the `or` would never reach the fallback.
            self.ingest.reject(
                f"meter poll {self.url}: {str(err) or type(err).__name__}")
            return False
        self.connected = True
        try:
            sample = parse_homewizard(doc, self.ingest.phases)
            # ⚠️ Submitted unconditionally, INCLUDING a document identical to the
            # last one. The response is the arrival; the number is not.
            # Inside the try on purpose: submit() outside it would put a raise on
            # a path with no handler at all, killing the poll task permanently -
            # safely (the age climbs) but silently. run() backstops the rest.
            self.ingest.submit(sample)
            self._note(doc)
        except P1Error as err:
            self.ingest.reject(err)
            return False
        return True
    def _note(self, doc) -> None:
        fp = measurement_fingerprint(doc)
        if self._changed_mono is None or fp != self._fingerprint:
            self._fingerprint = fp
            self._changed_mono = time.monotonic()
 # --------------------------------------------------------------------------- #
 # selection
 # --------------------------------------------------------------------------- #
 def is_enabled(opts: dict) -> bool:
    """Whether P1 ingestion is switched on at all.
    ⚠️ One definition, because three places depend on it and they MUST agree:
    where the grid reading comes from, whether the ingest task is started, and
    whether sensor.p1_sample_age_s is announced over MQTT discovery. An age
    sensor announced with no ingester behind it is a watchdog input nobody is
    feeding, and the ESP32 trips on it.
    """
    return str(opts.get("meter_source", "off") or "off").strip() not in ("off", "")
 def build_source(opts: dict, ingest: P1Ingest, session, broker: dict | None):
    """Return the transport named by `meter_source`, or None if disabled.
@@ -1051,7 +654,7 @@ def build_source(opts: dict, ingest: P1Ingest, session, broker: dict | None):
    changing transport is a config edit, never a code path.
    """
    source = str(opts.get("meter_source", "off") or "off").strip()
-    if not is_enabled(opts):
+    if source in ("off", ""):
        return None
    if source == SOURCE_HA:
        return HaDsmrSource(session, ingest, {
@@ -1060,84 +663,12 @@ def build_source(opts: dict, ingest: P1Ingest, session, broker: dict | None):
            "phase_import": opts.get("p1_phase_import_entities") or [],
            "phase_export": opts.get("p1_phase_export_entities") or [],
        })
    if source == SOURCE_HA_SIGNED:
        net = str(opts.get("p1_net_entity", "") or "").strip()
        phase_net = [str(e).strip() for e in
                     (opts.get("p1_phase_net_entities") or []) if str(e).strip()]
        # ⚠️ Both of these are checked ONCE here rather than per telegram. A
        # misconfigured source otherwise fails silently in the only way that
        # looks exactly like a healthy one that has not been sent anything yet:
        # no samples, a climbing age, and the firmware watchdog holding the
        # battery at 0 W with nothing in the log saying why.
        if not net:
            _LOG.error("meter_source %s needs p1_net_entity - P1 ingestion "
                       "disabled (sensor.p1_sample_age_s would otherwise be "
                       "announced with nothing feeding it)", SOURCE_HA_SIGNED)
            return None
        if phase_net and len(phase_net) != ingest.phases:
            _LOG.error("p1_phase_net_entities has %d entities but meter_phases "
                       "is %d - P1 ingestion disabled. Every telegram would be "
                       "rejected on the phase-count check.",
                       len(phase_net), ingest.phases)
            return None
        return HaSignedSource(session, ingest,
                              {"net": net, "phase_net": phase_net})
    if source == SOURCE_HOMEWIZARD:
        host = str(opts.get("p1_host", "") or "").strip()
        # A pasted browser URL is the obvious way to fill this in wrong.
        host = host.removeprefix("http://").removeprefix("https://").rstrip("/")
        host, _, port_s = host.partition(":")
        # ⚠️ Checked ONCE here, like the ha_signed entity ids and for the same
        # reason: a misconfigured source otherwise fails in the one way that
        # looks exactly like a healthy one nobody has sent anything to yet - no
        # samples, a climbing age, the watchdog holding the battery at 0 W, and
        # nothing in the log saying why.
        if not host:
            _LOG.error("meter_source %s needs p1_host (the meter's own address, "
                       "e.g. 192.168.2.250) - P1 ingestion disabled",
                       SOURCE_HOMEWIZARD)
            return None
        try:
            port = int(port_s) if port_s else HOMEWIZARD_PORT
        except ValueError:
            _LOG.error("p1_host %r has an unparseable port - P1 ingestion disabled",
                       opts.get("p1_host"))
            return None
        try:
            # ⚠️ Not `or 5`: that turns an explicit 0 into the default, and a
            # cadence of 0 is a misconfiguration that must be reported, not
            # quietly corrected into something that looks like it was asked for.
            raw = opts.get("meter_poll_s", 5)
            poll_s = float(5 if raw is None or raw == "" else raw)
        except (TypeError, ValueError):
            _LOG.error("meter_poll_s %r is not a number - P1 ingestion disabled",
                       opts.get("meter_poll_s"))
            return None
        if poll_s <= 0:
            _LOG.error("meter_poll_s must be positive - P1 ingestion disabled")
            return None
        # ⚠️ The age can never be fresher than the poll interval. At or past
        # max_age_s every reading is stale before its successor arrives, so the
        # controller would sit permanently on missing inputs while the meter is
        # perfectly healthy - refuse it rather than ship that.
        if poll_s >= ingest.max_age_s:
            _LOG.error("meter_poll_s %.1f is not under meter_max_age_s %.1f - every "
                       "reading would go stale before the next poll. P1 ingestion "
                       "disabled.", poll_s, ingest.max_age_s)
            return None
        if poll_s > ingest.max_age_s / 2:
            _LOG.warning("meter_poll_s %.1f leaves no room under meter_max_age_s "
                         "%.1f: one missed poll makes the reading stale. The meter "
                         "updates every ~5 s; 5 s against 30 s is the tested pair.",
                         poll_s, ingest.max_age_s)
        return HomeWizardLocalSource(session, ingest, host, port=port, poll_s=poll_s)
    if source == SOURCE_MQTT:
        broker = broker or {}
        return MqttP1Source(ingest, str(opts.get("meter_mqtt_topic", "")),
                            broker.get("host"), broker.get("port", 1883),
                            broker.get("username"), broker.get("password"))
    if source:
-        _LOG.error("meter_source %r is not one of %s - P1 ingestion disabled",
+        _LOG.error("meter_source %r is not %s or %s - P1 ingestion disabled",
-                   source, ", ".join((SOURCE_HA, SOURCE_MQTT, SOURCE_HA_SIGNED,
+                   source, SOURCE_HA, SOURCE_MQTT)
                                      SOURCE_HOMEWIZARD)))
    return None
@@ -1,5 +1,5 @@
 name: GoodWe RS485 Controller
-version: "0.3.0"
+version: "0.2.1"
 slug: goodwe_controller
 description: >-
  Drives a GoodWe ES/BP battery inverter over RS485 by emulating its smart
@@ -42,24 +42,11 @@ options:
  # --- P1 meter ingestion (specs §5.2 / §14 `meter:`) -------------------------
  # `off` keeps the original single meter_entity path above, so an existing
  # install is untouched until it opts in. ha_dsmr subscribes to the DSMR
-  # integration's entities over the HA WebSocket; mqtt_p1 reads the topic below;
+  # integration's entities over the HA WebSocket; mqtt_p1 reads the topic below.
  # ha_signed reads ONE signed HA entity (+ import / - export), which is what a
  # HomeWizard P1 publishes and what ha_dsmr cannot consume.
  # homewizard_local skips Home Assistant entirely and polls the meter's own
  # local API. ⚠️ It is the ONLY mode whose sensor.p1_sample_age_s measures when
  # the meter REPORTED rather than when the value last CHANGED - HA emits
  # nothing at all for a repeated reading - so it is the only mode a firmware
  # watchdog may threshold. See DOCS.md.
  meter_source: "off"
  meter_phases: 1
  meter_max_age_s: 30
  meter_mqtt_topic: ""
  # homewizard_local only. The meter's own address, `host` or `host:port`.
  p1_host: ""
  # homewizard_local only. Seconds between polls. Must be well under
  # meter_max_age_s - the age is never fresher than this interval - and there is
  # nothing to gain below the meter's own ~5.0 s update rate (NOTES.md:640).
  meter_poll_s: 5
  # The two UNSIGNED Belgian registers. The EMS derives net power from them
  # (import - export); do NOT point these at a signed template sensor.
  p1_import_entity: ""
@@ -68,15 +55,6 @@ options:
  # three-phase connection; the list length must equal meter_phases.
  p1_phase_import_entities: []
  p1_phase_export_entities: []
  # ha_signed only. ONE signed net-power sensor: positive = import from the
  # grid, negative = export to it. Do NOT split it into two template sensors -
  # the split is done in the add-on (p1.split_signed) precisely so the sign
  # convention is tested rather than living in unreviewed YAML.
  p1_net_entity: ""
  # Optional, in L1..L3 order, each one signed the same way. Same role as
  # p1_phase_import_entities: the capacity-tariff peak on a three-phase
  # connection. The list length must equal meter_phases.
  p1_phase_net_entities: []
  # --- control ---------------------------------------------------------------
  max_w: 2000
@@ -87,7 +65,7 @@ options:
  step_w: 10
  saturation_w: 500
  saturation_cycles: 3
-  integrator_max_w: 0
+  integrator_max_w: 3000
  heartbeat_s: 10
  stale_input_s: 15
  auto_start: false
@@ -120,25 +98,20 @@ schema:
  batt_invert: bool
  setpoint_entity: str
-  meter_source: list(off|ha_dsmr|mqtt_p1|ha_signed|homewizard_local)
+  meter_source: list(off|ha_dsmr|mqtt_p1)
  # ⚠️ 2 is accepted by this range but is not a real Belgian connection. A
  # telegram whose phase count disagrees is rejected at ingest and logged, so a
  # mis-set 2 shows up immediately as "0 telegrams accepted" rather than as a
  # quietly wrong number.
  meter_phases: int(1,3)
  meter_max_age_s: int(5,300)
  meter_poll_s: int(1,60)
  meter_mqtt_topic: str?
  p1_host: str?
  p1_import_entity: str?
  p1_export_entity: str?
  p1_phase_import_entities:
    - str
  p1_phase_export_entities:
    - str
  p1_net_entity: str?
  p1_phase_net_entities:
    - str
  max_w: int(100,5000)
  gain: float(0.05,1.0)
@@ -148,7 +121,7 @@ schema:
  step_w: int(1,100)
  saturation_w: int(100,2000)
  saturation_cycles: int(1,10)
-  integrator_max_w: int(0,15000)
+  integrator_max_w: int(100,15000)
  heartbeat_s: int(2,25)
  stale_input_s: int(5,120)
  auto_start: bool
@@ -24,63 +24,6 @@ def check(name, cond):
        fails.append(name)
 # ---------------------------------------------------------------------------
 # COVERAGE AUDIT - measured, not executed. Read this before adding a mechanism.
 #
 # THE INVARIANT: every mechanism in compute() must be noticed by AT LEAST TWO
 # checks when it is deleted. If you add a mechanism to compute(), re-run the
 # audit and add it to the table. If a figure here drops, a check has started
 # passing for a reason other than the one it names.
 #
 # THE TECHNIQUE, because there is no script to run: replace one mechanism in
 # control.py with a no-op, run this file, count the failures, restore. That is
 # the converse of the usual mutation - not "does a wrong value fail?" but "does
 # anyone notice when the mechanism is GONE?". It is kept as a comment rather
 # than as tooling on purpose: the only cheap way to automate it is to key on
 # source lines, which goes stale silently, and a green audit that has quietly
 # stopped testing anything is precisely the failure this ticket exists to fix.
 # A comment cannot go stale-green, because it never claims to be running.
 #
 # Measured at 389d9ec. Numbers are the lead's independent reproduction.
 #
 #   mechanism in compute()                     checks that fail when deleted
 #   ------------------------------------------ -----------------------------
 #   integrator freeze (AC 3)                                               2
 #   integrator clamp (AC 1)                                                6
 #   integrator bound follows max_w                                         4
 #   output clamp                                                           3
 #   slew limit                                                             4
 #   output freeze                                                          2
 #   deadband                                                               5
 #   quantisation                                                           2
 #   saturation detector, `saturated_now = False`                          11
 #   saturation duration (AC 2), fires instantly                            2
 #   sat counter reset on a good cycle                                      6
 #   target_grid_w bias                                                     3
 #   i_w=None seeding from prev_w                                           6
 #
 # The detector figure is for the `saturated_now = False` form specifically;
 # disabling it further down as `frozen = False` is a weaker mutation and gives
 # 10. Reproduce the same form or the number will not match.
 #
 # ⚠️ IT HAS FOUND A DEAD MECHANISM TWICE, BOTH THE SAME WAY: a clamp standing in
 # for the mechanism under test. Deleting the integrator freeze once failed
 # NOTHING, because the fixtures sat at max_w 2000 and the integrator bound
 # truncated a wound value back to exactly 2000 - the assertion passed on the
 # clamp. The output clamp was masked the same way by the integrator bound.
 # Hence: A FIXTURE MUST SIT CLEAR OF EVERY RAIL IT IS NOT TESTING. Where a test
 # names one mechanism, make that mechanism the binding one (see TCLAMP and TF).
 #
 # ⚠️ RUN MUTATIONS WITH `python -B` AND CLEAR app/__pycache__. CPython
 # invalidates a .pyc on (source mtime in whole seconds, source size), so a
 # same-second rewrite that also preserves the file size reuses stale bytecode
 # and the suite reports on code you are no longer running. It under-reported one
 # mutation here as 2 where the true figure is 6. The error is one-directional -
 # stale bytecode can only under-report - so every figure above is a lower bound
 # at worst, and the two zeros ever recorded were both confirmed by fixing them
 # and watching the count rise, which a caching artefact cannot do.
 # ---------------------------------------------------------------------------
 print("control law")
 # Deadband: inside meter noise, hold exactly - do not drift.
@@ -98,22 +41,13 @@ check("proportional step (gain 0.6)", d.target_w == 300)
 d = compute(prev_w=0, grid_w=-500, actual_w=0, tuning=T)
 check("export drives charging", d.target_w == -300)
-# Clamp.
+# Clamp
-# ⚠️ integrator_max_w is lifted clear of max_w so that the OUTPUT clamp is the
+d = compute(prev_w=1900, grid_w=1000, actual_w=1900, tuning=Tuning(max_w=2000, slew_w=5000))
 # mechanism under test. Left at the default the integrator bound truncates
 # first, these two assertions pass on that alone, and deleting the output clamp
 # fails nothing - the same masking that hid the integrator freeze.
 TCLAMP = Tuning(max_w=2000, slew_w=5000, integrator_max_w=5000)
 d = compute(prev_w=1900, grid_w=1000, actual_w=1900, tuning=TCLAMP)
 check("clamped to max_w", d.target_w == 2000)
 # Slew: from 0 with a huge error, no more than slew_w in one cycle.
 d = compute(prev_w=0, grid_w=5000, actual_w=0, tuning=Tuning(max_w=5000, slew_w=1000))
 check("slew limits one cycle", d.target_w == 1000)
 d = compute(prev_w=-1900, grid_w=-1000, actual_w=-1900, tuning=TCLAMP)
 check("clamped to -max_w", d.target_w == -2000)
 d = compute(prev_w=0, grid_w=-5000, actual_w=0, tuning=Tuning(max_w=5000, slew_w=1000))
 check("slew limits one cycle, charging", d.target_w == -1000)
 # Saturation needs DURATION: one diverging cycle must NOT freeze.
 t = Tuning(saturation_w=500, saturation_cycles=3)
@@ -144,7 +78,7 @@ print("SAFETY-04: the integrator is bounded apart from the output")
 # The historical runaway, with its real numbers. A commercial controller on
 # this site, with the inverter switched OFF, wound ~130 W every 4 s past 10 kW
 # and reported 14 768 W while its output clamp sat at 5 kW. At gain 0.6 that
-# rate is a standing error of 130/0.6 = 217 W that never resolves, because the
+# rate is a standing error of 130/0.6 ≈ 217 W that never resolves, because the
 # inverter is not there to resolve it. 150 cycles is past the ~113 it took to
 # reach 14 768 W at that rate.
 RUNAWAY_ERROR = 130.0 / 0.6
@@ -152,12 +86,12 @@ RUNAWAY_CYCLES = 150
 HISTORICAL_W = 14768.0
-def runaway(tuning, sign=1):
+def runaway(tuning):
    """Inverter off: it reports 0 W forever, the error never clears."""
    prev, i_w, sat = 0.0, 0.0, 0
    worst_i, worst_cmd = 0.0, 0.0
    for _ in range(RUNAWAY_CYCLES):
-        d = compute(prev_w=prev, grid_w=sign * RUNAWAY_ERROR, actual_w=0.0,
+        d = compute(prev_w=prev, grid_w=RUNAWAY_ERROR, actual_w=0.0,
                    tuning=tuning, sat_count=sat, i_w=i_w)
        prev, i_w, sat = d.target_w, d.i_w, d.sat_count
        worst_i = max(worst_i, abs(i_w))
@@ -165,113 +99,33 @@ def runaway(tuning, sign=1):
    return worst_i, worst_cmd
-TR = Tuning(max_w=2000)          # integrator_max_w unset => follows max_w
+TR = Tuning(max_w=2000, integrator_max_w=3000)
 wi, wc = runaway(TR)
-check(f"runaway: integrator plateaus at {wi:.0f} W (<= 2000)", wi <= TR.max_w)
+check(f"runaway: integrator plateaus at {wi:.0f} W (<= 3000)", wi <= TR.integrator_max_w)
 check(f"runaway: emitted command peaks at {wc:.0f} W (<= 2000)", wc <= TR.max_w)
 check("runaway: nowhere near the historical 14 768 W", wc < HISTORICAL_W / 4)
 # ...and with the saturation detector deliberately defeated, so that only the
-# clamp is holding. Kill one mechanism, the other still bounds it.
+# clamp is holding. This is the AC that says the two mechanisms are
-TD = Tuning(max_w=2000, saturation_w=1e9)
+# independent: kill one, the other still bounds it.
 TD = Tuning(max_w=2000, integrator_max_w=3000, saturation_w=1e9)
 wi, wc = runaway(TD)
-check(f"runaway with the detector defeated: integrator still bounded ({wi:.0f} W)",
+check(f"runaway with the detector defeated: integrator still <= 3000 ({wi:.0f} W)",
-      wi <= TD.max_w)
+      wi <= TD.integrator_max_w)
 check("runaway with the detector defeated: command still <= max_w", wc <= TD.max_w)
-# The mirror: the same runaway driving the other way. An export that never
+# The bound is not max_w. If someone "simplifies" them into one key this fails.
 # clears winds the integrator negative just as hard.
 wi, wc = runaway(Tuning(max_w=2000), sign=-1)
 check(f"runaway (export direction): integrator bounded at {wi:.0f} W", wi <= 2000)
 check("runaway (export direction): emitted command <= max_w", wc <= 2000)
 # The bound is a separate quantity, and the useful direction is BELOW max_w:
 # there it binds first and caps unwind latency tighter than the rail does.
 d = compute(prev_w=0, grid_w=6000, actual_w=0,
-            tuning=Tuning(max_w=2000, integrator_max_w=1000, slew_w=5000))
+            tuning=Tuning(max_w=2000, integrator_max_w=3000, slew_w=5000))
-check("integrator bound binds independently of the output clamp",
+check("integrator bound is separate from the output clamp",
-      d.i_w == 1000 and d.target_w == 1000)
+      d.i_w == 3000 and d.target_w == 2000)
-# Freeze = may not wind further in the direction it is already pushing.
+# Freeze = does not accumulate. Same input twice; the integrator must not move.
-# ⚠️ max_w is raised WELL above the fixtures on purpose. At the default 2000
+TF = Tuning(saturation_w=500, saturation_cycles=3)
 # the integrator bound truncates a wound value back to exactly 2000 and
 # satisfies these assertions on its own, so deleting the freeze outright
 # failed nothing - the clamp was standing in for the mechanism under test.
 # Any fixture here must sit clear of every rail, or it tests the rail.
 TF = Tuning(saturation_w=500, saturation_cycles=3, max_w=5000)
 f1 = compute(prev_w=2000, grid_w=800, actual_w=0, tuning=TF, sat_count=3, i_w=2000.0)
-check("frozen: integration does not wind further", f1.i_w == 2000.0 and f1.frozen)
+check("frozen: integration does not accumulate", f1.i_w == 2000.0 and f1.frozen)
 f2 = compute(prev_w=2000, grid_w=-800, actual_w=0, tuning=TF, sat_count=3, i_w=2000.0)
 check("frozen: unwinding is still allowed", f2.i_w < 2000.0)
 # ...and the same two on the charging side. Every freeze rule in this file has
 # a mirror, because the one that did not is the defect that got through review.
 f3 = compute(prev_w=-2000, grid_w=-800, actual_w=0, tuning=TF, sat_count=3, i_w=-2000.0)
 check("frozen (charging): integration does not wind further", f3.i_w == -2000.0)
 f4 = compute(prev_w=-2000, grid_w=800, actual_w=0, tuning=TF, sat_count=3, i_w=-2000.0)
 check("frozen (charging): unwinding is still allowed", f4.i_w > -2000.0)
 # ⚠️ REGRESSION, and the reason the first cut of SAFETY-04 was rejected. A
 # freeze encoded as "only corrections that shrink |i_w|" is unsatisfiable for
 # BOTH signs of error whenever |correction| > 2*|i_w|, so near zero the loop
 # stops moving forever - the freeze cannot clear, because clearing it needs the
 # inverter to track and not-tracking is what saturation means. Measured on that
 # encoding: 0 W held into a 2 kW import for as long as the sim ran.
 z = compute(prev_w=0, grid_w=2000, actual_w=600, tuning=T, sat_count=3, i_w=0.0)
 check("frozen at i_w=0: a 2 kW import still moves the command",
      z.frozen and z.target_w == 1000)
 # ...and the next cycle the inverter is inside saturation_w of the command, so
 # the freeze clears on its own. Deadlock would show up here as frozen=True.
 z2 = compute(prev_w=1000, grid_w=1000, actual_w=600, tuning=T,
             sat_count=z.sat_count, i_w=z.i_w)
 check("frozen at i_w=0: the freeze then clears", not z2.frozen)
 # Same stranding on the other side: a small positive integrator against export.
 z3 = compute(prev_w=100, grid_w=-1000, actual_w=800, tuning=T, sat_count=3, i_w=100.0)
 check("frozen at i_w=+100: a 1 kW export still moves the command",
      z3.frozen and z3.target_w < 0)
 z4 = compute(prev_w=-100, grid_w=1000, actual_w=-800, tuning=T, sat_count=3, i_w=-100.0)
 check("frozen at i_w=-100: a 1 kW import still moves the command",
      z4.frozen and z4.target_w > 0)
 # ⚠️ EXACTLY ZERO, BOTH DIRECTIONS. This boundary has a history: the first cut
 # deadlocked here under import, and the fix for it deadlocked here under export
 # because `if i_w > 0 ... else ...` files 0.0 under rising-only. main.py resets
 # i_w to exactly 0.0 on every stop and every reseed, so it is a normal state,
 # not a corner.
 zi = compute(prev_w=0, grid_w=2000, actual_w=600, tuning=T, sat_count=3, i_w=0.0)
 check("frozen at i_w=0.0: an import push moves the integrator",
      zi.frozen and zi.i_w > 0)
 ze = compute(prev_w=0, grid_w=-2000, actual_w=-600, tuning=T, sat_count=3, i_w=0.0)
 check("frozen at i_w=0.0: an export push moves the integrator",
      ze.frozen and ze.i_w < 0)
 # The COMMAND still holds at 0 W in that second case, and that is release/1.0's
 # rule, not a leftover: at prev_w == 0 the output freeze forbids starting to
 # charge while saturated, because commanding 0 while the inverter reports
 # hundreds of watts means something else is driving the bus. Asserted so that
 # nobody "fixes" it by accident - the integrator moving is what this ticket
 # owns, the command rule belongs to the output freeze.
 check("frozen at i_w=0.0: the output freeze still blocks a charge from 0 W",
      ze.target_w == 0.0)
 # Where prev_w is already charging the output freeze does NOT block, and there
 # the difference reaches the wire: held at 0.0 the integrator abandons the
 # charge mid-export.
 zc = compute(prev_w=-2000, grid_w=-4000, actual_w=-600, tuning=T, sat_count=3, i_w=0.0)
 check("frozen at i_w=0.0: a charge is not abandoned during heavy export",
      zc.target_w == -2000.0)
 # The general property, rather than another handful of points: while frozen the
 # integrator may be held ONLY when the correction would push it further from
 # zero on the side it already sits. Any other hold is a deadlock.
 stuck = []
 for i0 in [x * 25.0 for x in range(-80, 81)]:
    for g in [x * 100.0 for x in range(-40, 41)]:
        err = g - T.target_grid_w
        if abs(err) < T.deadband_w:
            continue
        dd = compute(prev_w=0.0, grid_w=g, actual_w=1500.0, tuning=T, sat_count=3, i_w=i0)
        if dd.i_w == i0 and not ((i0 > 0 and err > 0) or (i0 < 0 and err < 0)):
            stuck.append((i0, g))
 check(f"frozen integrator never deadlocks, over {161*81} states"
      + (f" (e.g. {stuck[0]})" if stuck else ""), not stuck)
 # False-positive guard: a normal 2 kW load step must not trip the detector,
 # because the plant needs several cycles to catch up on every one of them.
@@ -283,107 +137,6 @@ for _ in range(12):
    froze = froze or d.frozen
 check("a normal 2 kW load step does not trip the saturation freeze", not froze)
 # The convergence sim below runs WITHOUT a carried integrator. This is the same
 # 2 kW step in the configuration that actually ships, where main.py carries it.
 prev, actual, sat, i_w = 0.0, 0.0, 0, 0.0
 carried = 0
 for _ in range(12):
    d = compute(prev, 2000.0 - actual, actual, T, sat, i_w)
    prev, sat, i_w = d.target_w, d.sat_count, d.i_w
    actual = actual + 0.94 * (prev - actual)
    carried += 1
    if abs(2000.0 - actual) < T.deadband_w:
        break
 check(f"carried integrator converges in {carried} cycles (<=6)", carried <= 6)
 check("carried integrator does not overshoot the load", actual <= 2000.0 + T.deadband_w)
 # ⚠️ REGRESSION: an integrator allowed to wind past the rail buys nothing (the
 # output clamp already bounds the wire) and costs extra cycles of
 # wrong-direction power after every saturation event. 4000 W load held to
 # saturation, then dropped to 0; the figure is the command on the first cycle
 # after the drop. This is what makes the DOCS advice checkable.
 def unwind(t):
    prev, actual, sat, i_w, load = 0.0, 0.0, 0, 0.0, 4000.0
    for c in range(16):
        if c == 15:
            load = 0.0
        d = compute(prev, load - actual, actual, t, sat, i_w)
        prev, sat, i_w = d.target_w, d.sat_count, d.i_w
        actual = actual + 0.94 * (prev - actual)
    return prev
 tight, loose = unwind(Tuning(max_w=2000)), unwind(Tuning(max_w=2000, integrator_max_w=3000))
 check(f"after saturation ends the command is {tight:.0f} W (<= 1000)", tight <= 1000)
 check(f"headroom above max_w makes that worse ({loose:.0f} W) - hence the default",
      loose > tight)
 print("SAFETY-04: the i_w=None path is still release/1.0, exactly")
 def legacy(prev, grid, actual, t, sat_count):
    """release/1.0's control law, transcribed. Do not 'improve' this."""
    reason = "tracking"
    sc = min(sat_count + 1, 10) if abs(prev - actual) > t.saturation_w else 0
    frozen = sc >= t.saturation_cycles
    error = grid - t.target_grid_w
    if abs(error) < t.deadband_w:
        want, reason = prev, "deadband"
    else:
        want = prev + t.gain * error
    target = max(-t.max_w, min(t.max_w, want))
    if target != want:
        reason = "clamped"
    slewed = max(prev - t.slew_w, min(prev + t.slew_w, target))
    if slewed != target:
        reason = "slew-limited"
    target = slewed
    if frozen:
        target = min(target, prev) if prev > 0 else max(target, prev)
        reason = "saturated-freeze"
    step = max(1, int(t.step_w))
    return float(round(target / step) * step), sc, frozen, reason
 # ⚠️ Compare EVERYTHING observable, not just the number. A previous version of
 # this sweep compared (target_w, sat_count) only and passed 3024 cases while
 # `reason` had silently lost a value - which is the kind of thing a sweep this
 # broad exists to catch. `frozen` and `reason` are both in the tuple now.
 #
 # The one deliberate rename: what release/1.0 called "clamped" is now
 # "i-clamped", because the truncation happens on the integrator before the
 # command is derived from it. Aliased here rather than papered over - if any
 # OTHER reason ever diverges, this check goes red.
 ALIAS = {"i-clamped": "clamped"}
 diffs = []
 seen = set()
 for tune in (Tuning(), Tuning(target_grid_w=-10.0), Tuning(max_w=5000, slew_w=5000)):
    for prev in (-2000.0, -500.0, -100.0, 0.0, 100.0, 500.0, 2000.0):
        for grid in (-6000.0, -1000.0, -500.0, -14.0, 0.0, 14.0, 500.0, 1000.0, 6000.0):
            for actual in (-2000.0, 0.0, 600.0, 2000.0):
                for sc in (0, 2, 3, 9):
                    d = compute(prev, grid, actual, tune, sc)   # i_w defaults to None
                    seen.add(d.reason)
                    got = (d.target_w, d.sat_count, d.frozen,
                           ALIAS.get(d.reason, d.reason))
                    if got != legacy(prev, grid, actual, tune, sc):
                        diffs.append((prev, grid, actual, sc, got,
                                      legacy(prev, grid, actual, tune, sc)))
 check(f"i_w=None reproduces release/1.0 over {3*7*9*4*4} cases, reason included"
      + (f" (first diff {diffs[0]})" if diffs else ""), not diffs)
 # ...and the rename is not a quiet deletion: the signal SAFETY-03 alarms on has
 # to actually occur in that sweep, or its hook is dead.
 check("the integrator clamp reports itself as 'i-clamped'", "i-clamped" in seen)
 # "clamped" stays reachable, but only where the integrator is deliberately
 # allowed above the rail - then BOTH fire and the output clamp, which describes
 # the value actually emitted, is the one reported.
 dc = compute(prev_w=0, grid_w=6000, actual_w=0,
             tuning=Tuning(max_w=2000, integrator_max_w=3000, slew_w=5000))
 check("the output clamp still reports 'clamped' when it is the binding one",
      dc.reason == "clamped" and dc.i_w == 3000 and dc.target_w == 2000)
 print("capacity tariff")
 check("no forecast means no cap", maintenance_charge_floor(2500, None, 3500) == 2500)
 check("headroom caps the charge", maintenance_charge_floor(2500, 2000, 3500) == 1500)
@@ -11,7 +11,6 @@ computes the wrong quarter-hour figure whenever the telegram cadence changes.
 """
 import asyncio
 import json
 import sys
 import time
 from datetime import datetime, timedelta, timezone
@@ -19,11 +18,8 @@ from datetime import datetime, timedelta, timezone
 import aiohttp   # already required by app.p1, so this adds no new dependency
 from app.p1 import (
-    P1Error, P1Ingest, HaDsmrSource, HaSignedSource, HomeWizardLocalSource,
+    P1Error, P1Ingest, HaDsmrSource, QuarterAverager, SOURCE_HA, SOURCE_MQTT,
-    QuarterAverager,
+    make_sample, parse_mqtt_payload,
    SOURCE_HA, SOURCE_HA_SIGNED, SOURCE_HOMEWIZARD, SOURCE_MQTT,
    build_source, is_enabled, make_sample, parse_homewizard, parse_mqtt_payload,
    split_signed,
 )
 fails = []
@@ -60,25 +56,6 @@ def raises(name, fn):
    fails.append(name)
 def built(src):
    """`src.build()`, with any escaping exception turned into a visible value.
    ⚠️ Legibility of a RED, not leniency. build() is contracted to return a bool
    and to funnel every bad telegram through ingest.reject() - a guard that goes
    missing (say the "is the net entity cached at all" one) makes it raise
    instead. That still fails the suite, but by aborting it with a traceback at
    whichever check happened to run first, which costs the next person ten
    minutes deciding whether the suite is broken or the code is. Returning the
    exception makes it compare unequal to True/False, so the NAMED check goes red
    and says which rule died.
    """
    try:
        return src.build()
    except Exception as err:  # noqa: BLE001 - a raise here is itself the failure
        print(f"       build() raised {type(err).__name__}: {err}")
        return err
 def sample(net_import, net_export=0.0, at=BASE, phases=1, pi=None, pe=None):
    return make_sample(SOURCE_HA, net_import, net_export, phases=phases,
                       phase_import_w=pi, phase_export_w=pe,
@@ -252,47 +229,6 @@ before = a.partial_ws
 a.add(sample(1000.0, at=BASE + timedelta(seconds=5)))
 check("an out-of-order telegram is dropped, not integrated backwards",
      a.partial_ws == before and a.elapsed_s == 10.0)
 # ⚠️ The assertion above is NOT sufficient on its own, and that is the whole
 # lesson: deleting the guard still passes it, because the negative interval is
 # separately refused by the `covered > 0` test. What the guard actually prevents
 # is the REWIND - without it the held timestamp moves back to +5 s and the next
 # telegram re-integrates the 5..10 s window that was already counted. The damage
 # only becomes visible one sample later, so the test has to go one sample later.
 a.add(sample(1000.0, at=BASE + timedelta(seconds=20)))
 check("...and the held timestamp is not rewound, so the next telegram "
      "cannot double-count", a.elapsed_s == 20.0 and a.partial_ws == 20000.0)
 # A duplicate telegram (identical timestamp) is the same rule.
 a = QuarterAverager(1)
 a.add(sample(1000.0, at=BASE))
 a.add(sample(1000.0, at=BASE + timedelta(seconds=10)))
 a.add(sample(4000.0, at=BASE + timedelta(seconds=10)))
 a.add(sample(1000.0, at=BASE + timedelta(seconds=20)))
 check("a duplicate timestamp neither re-integrates nor replaces the held value",
      a.elapsed_s == 20.0 and a.partial_ws == 20000.0)
 # A gap must not be filled with the last held value. The meter dies at 5 kW and
 # returns ten minutes later; hold-forward would credit 5 kW x 600 s to the
 # capacity-tariff accumulator - a fabricated peak, on a permanent record, from
 # data nobody measured.
 a = QuarterAverager(1, max_hold_s=30.0)
 a.add(sample(5000.0, at=BASE))
 a.add(sample(5000.0, at=BASE + timedelta(seconds=600)))
 check("a 600 s gap is held for at most max_hold_s, not for the whole gap",
      a.partial_ws == 5000.0 * 30.0)
 check("the unobserved stretch does not count as elapsed time", a.elapsed_s == 30.0)
 closed = a.add(sample(5000.0, at=BASE + timedelta(seconds=900)))
 check("the outage drags the billed quarter down instead of inventing a peak",
      len(closed) == 1 and abs(closed[0].offtake_avg_w - 300000.0 / 900.0) < 1e-9)
 check("...nowhere near the 5000 W a hold-forward would have billed",
      closed[0].offtake_avg_w < 400.0)
 # The cap must not disturb a normally-spaced stream.
 a = QuarterAverager(1, max_hold_s=30.0)
 for i in range(0, 121, 5):          # a healthy 5 s telegram cadence
    a.add(sample(2000.0, at=BASE + timedelta(seconds=i)))
 check("a healthy 5 s cadence is untouched by the hold cap",
      a.elapsed_s == 120.0 and abs(a.offtake_avg_w - 2000.0) < 1e-9)
 # --------------------------------------------------------------------------- #
 print("ingest timestamp, age and staleness")
@@ -546,15 +482,9 @@ async def _e2e():
 live, wire = asyncio.run(_e2e())
 check("the websocket handshake and subscription complete", live.samples >= 1)
-# ⚠️ TWO, not three. get_states primes the cache but must NOT build a sample:
+# Six state_changed events arrived (two per telegram). The debounce is what
-# HA returns whatever it currently holds, which after a Core restart is a
+# makes that three consistent samples instead of six half-updated ones.
-# RestoreEntity value of unknown age, and stamping that with ingest_ts=now
+check("three telegrams produce three samples, not six", live.samples == 3)
 # resets the age and reports a fresh meter that may have been dead for an hour.
 # Only the two real state_changed telegrams become samples. Four state_changed
 # events arrived (two per telegram); the debounce is what makes those two
 # consistent samples rather than four half-updated ones.
 check("connecting does not manufacture a sample from cached HA state",
      live.samples == 2)
 check("the final export-dominant telegram nets negative",
      live.last.net_w == -800.0)
 check("the sample was built over the wire, tagged with its transport",
@@ -562,885 +492,9 @@ check("the sample was built over the wire, tagged with its transport",
 check("an entity we did not subscribe to is never cached",
      "sensor.something_else" not in wire.cache and len(wire.cache) == 1)
 check("a mid-stream unavailable is a parse error, not a sample",
-      live.parse_errors == 1 and live.samples == 2)
+      live.parse_errors == 1 and live.samples == 3)
 check("the last good reading survives the unavailable", live.net_w == -800.0)
-check("the averager integrated the live stream", live.averager.elapsed_s > 0.2)
+check("the averager integrated the live stream", live.averager.elapsed_s > 0.5)
 # The reason get_states still matters: it is what lets the FIRST real telegram
 # build a complete sample instead of waiting for every entity to change once.
 check("the primed cache let the first telegram build immediately",
      live.samples == 2 and live.last.import_w == 0.0)
 # --------------------------------------------------------------------------- #
 print("ha_signed: one signed entity -> the same two magnitudes")
 # The meter actually fitted at this house is a HomeWizard P1 publishing ONE
 # signed sensor. ha_dsmr cannot read it - it wants two unsigned registers and
 # refuses a negative one, which is every exporting telegram.
 check("a positive reading is import", split_signed(1500.0) == (1500.0, 0.0))
 check("a negative reading is export", split_signed(-900.0) == (0.0, 900.0))
 check("zero is a balanced reading, not a missing one",
      split_signed(0.0) == (0.0, 0.0))
 check("exactly one magnitude is ever non-zero",
      all(a == 0.0 or b == 0.0 for a, b in
          (split_signed(v) for v in (-5710.0, -1.0, 0.0, 1.0, 4384.0))))
 # ⚠️ The split must not change the number. A control loop steered by a value
 # that was rounded or clamped on the way in is steered by a different meter.
 check("the split round-trips the signed value exactly",
      all(make_sample(SOURCE_HA_SIGNED, *split_signed(v), phases=1).net_w == v
          for v in (-11763.0, -5710.0, -0.5, 0.0, 0.5, 775.0, 4384.0)))
 raises("a non-numeric signed reading is rejected", lambda: split_signed("n/a"))
 raises("a signed None is rejected, not read as zero", lambda: split_signed(None))
 raises("a signed NaN is rejected", lambda: split_signed(float("nan")))
 raises("a signed infinity is rejected", lambda: split_signed(float("inf")))
 # ⚠️ This one matters MORE on the signed path than on the unsigned one. On
 # ha_dsmr the §20 contamination is also caught by "unsigned cannot be negative";
 # here 64954 arrives as a perfectly well-formed positive signed reading and the
 # plausibility ceiling is the only thing standing in front of it.
 raises("the 64954 signed-decode contamination is still rejected",
       lambda: split_signed(64954.0))
 raises("...and its negative twin too", lambda: split_signed(-64954.0))
 # --------------------------------------------------------------------------- #
 print("ha_signed: the sign convention, against real captured readings")
 # ⚠️ Not a datasheet claim. These are literal values out of
 # sim/scenarios/ha-p1_meter_active_power-2026-08-{20,23}.json, HA recorder
 # exports of sensor.p1_meter_active_power at this house, copied here rather than
 # read from that repo so this file still runs on a laptop with nothing installed
 # (§17). If the convention were inverted, the physics below would be absurd.
 # 2026-08-23T11:46:52Z - the day's most negative reading, 13:46 local, full sun.
 s = make_sample(SOURCE_HA_SIGNED, *split_signed(-5710.0), phases=1)
 check("the midday PV peak (-5710 W) is EXPORT, not a 5.7 kW draw",
      s.net_w == -5710.0 and s.export_w == 5710.0 and s.import_w == 0.0)
 # 2026-08-19T22:00:00Z - midnight local, 20 Aug's file starts here. No sun.
 s = make_sample(SOURCE_HA_SIGNED, *split_signed(775.0), phases=1)
 check("the overnight base load (+775 W) is IMPORT",
      s.net_w == 775.0 and s.import_w == 775.0 and s.export_w == 0.0)
 # 2026-08-20T12:20:58Z - 14:20 local, the largest export in either capture.
 s = make_sample(SOURCE_HA_SIGNED, *split_signed(-11763.0), phases=1)
 check("the -11763 W midday extreme is export and survives the ceiling",
      s.net_w == -11763.0 and s.export_w == 11763.0)
 # 2026-08-23T10:18:28Z - the largest import in the healthy capture.
 s = make_sample(SOURCE_HA_SIGNED, *split_signed(4384.0), phases=1)
 check("the +4384 W peak is import", s.net_w == 4384.0 and s.import_w == 4384.0)
 # The whole convention in one line: night draws, midday feeds back.
 check("night is positive and midday is negative, which is the convention",
      split_signed(775.0)[0] > 0 and split_signed(-5710.0)[1] > 0)
 # --------------------------------------------------------------------------- #
 print("ha_signed transport: building a sample out of one entity state")
 NET = {"net": "sensor.p1_meter_active_power", "phase_net": []}
 ing = P1Ingest(phases=1, max_age_s=30.0)
 sig = HaSignedSource(None, ing, NET, token="x")
 check("nothing cached yet builds nothing", built(sig) is False and ing.last is None)
 sig._absorb("sensor.p1_meter_active_power", "1000")
 check("one signed entity is a complete telegram on its own",
      built(sig) is True and ing.net_w == 1000.0)
 check("the sample is tagged with its own transport",
      ing.last.source == SOURCE_HA_SIGNED)
 sig._absorb("sensor.p1_meter_active_power", "-2500")
 built(sig)
 check("a negative state lands as a negative net", ing.net_w == -2500.0)
 before = ing.last
 sig._absorb("sensor.p1_meter_active_power", "unavailable")
 check("an unavailable signed entity is a parse error", ing.parse_errors == 1)
 check("an unavailable entity does not build a sample", built(sig) is False)
 # ⚠️ The rule the whole ticket turns on: a missing reading is MISSING. Resolving
 # it to 0 W would read as a perfectly balanced house and defeat the staleness
 # trigger that FW-01's watchdog is built on.
 check("an unavailable entity leaves the last good sample untouched, not 0 W",
      ing.last is before and ing.net_w == -2500.0)
 sig._absorb("sensor.p1_meter_active_power", "unknown")
 check("an unknown signed entity is treated the same way", ing.parse_errors == 2)
 sig._absorb("sensor.p1_meter_active_power", "banana")
 check("a non-numeric signed state is a parse error, not 0 W",
      ing.parse_errors == 3 and ing.net_w == -2500.0)
 sig._absorb("sensor.p1_meter_active_power", "64954")
 check("64954 is refused at the signed transport too",
      built(sig) is False and ing.parse_errors == 4)
 sig._absorb("sensor.not_ours", "123")
 check("an unsubscribed entity is never cached by the signed transport",
      "sensor.not_ours" not in sig.cache)
 # A rejected reading must not make the age look fresh - the age is what the
 # firmware watchdog reads, and a rejection is exactly when it must keep climbing.
 ing = P1Ingest(phases=1, max_age_s=30.0)
 sig = HaSignedSource(None, ing, dict(NET), token="x")
 ing.submit(make_sample(SOURCE_HA_SIGNED, 1200, 0, phases=1,
                       ingest_mono=time.monotonic() - 20.0))
 sig._absorb("sensor.p1_meter_active_power", "unavailable")
 built(sig)
 check("a rejected reading does not reset the published age",
      ing.published_age_s > 19 and ing.net_w == 1200.0)
 ing.submit(make_sample(SOURCE_HA_SIGNED, 1200, 0, phases=1,
                       ingest_mono=time.monotonic() - 40.0))
 check("...and the age keeps climbing past max_age_s on its own",
      ing.stale is True and ing.net_w is None)
 # The three-phase reading the TEL-04 survey recorded at this house: L1 +2301 W,
 # L2 +468 W, L3 -2582 W, netting +187 W. A signed per-phase set splits the same
 # way, and the exporting phase must still clamp out of the billed figure.
 ing3 = P1Ingest(phases=3, max_age_s=30.0)
 NET3 = {"net": "sensor.p1_meter_active_power",
        "phase_net": ["sensor.p1_l1", "sensor.p1_l2", "sensor.p1_l3"]}
 sig3 = HaSignedSource(None, ing3, NET3, token="x")
 for eid, val in (("sensor.p1_meter_active_power", "187"), ("sensor.p1_l1", "2301")):
    sig3._absorb(eid, val)
 check("an incomplete signed phase set waits instead of guessing", built(sig3) is False)
 sig3._absorb("sensor.p1_l2", "468")
 sig3._absorb("sensor.p1_l3", "-2582")
 check("a complete signed three-phase set builds", built(sig3) is True)
 check("signed per-phase entities keep the exporting phase negative",
      ing3.last.per_phase_w == (2301.0, 468.0, -2582.0))
 check("per-phase IMPORT clamps the exporting phase to zero",
      ing3.last.per_phase_import_w == (2301.0, 468.0, 0.0))
 check("the phase import sum is 2769 W while the connection nets 187 W",
      sum(ing3.last.per_phase_import_w) == 2769.0 and ing3.last.net_w == 187.0)
 check("the signed per-phase tuple length matches meter_phases",
      len(ing3.last.per_phase_w) == ing3.phases == 3)
 sig_bad = HaSignedSource(None, P1Ingest(phases=3, max_age_s=30.0),
                         {"net": "sensor.net", "phase_net": ["sensor.a", "sensor.b"]},
                         token="x")
 for eid in ("sensor.net", "sensor.a", "sensor.b"):
    sig_bad._absorb(eid, "100")
 check("two phases delivered against meter_phases 3 is rejected, not padded",
      built(sig_bad) is False and sig_bad.ingest.last is None
      and sig_bad.ingest.parse_errors == 1)
 # --------------------------------------------------------------------------- #
 print("ha_signed transport: end to end against a fake Home Assistant")
 # The transport is a subclass, so this is what proves the INHERITED machinery -
 # auth, subscribe, the get_states priming rule, the reconnect-emits-nothing
 # rule - still behaves when only _wanted() and build() were replaced.
 async def _e2e_signed():
    from aiohttp import web
    import app.p1 as p1mod
    done = asyncio.Event()
    eid = "sensor.p1_meter_active_power"
    async def fake_ha(request):
        ws = web.WebSocketResponse()
        await ws.prepare(request)
        await ws.send_json({"type": "auth_required", "ha_version": "2026.8"})
        auth = await ws.receive_json()
        assert auth["type"] == "auth" and auth["access_token"] == "tok"
        await ws.send_json({"type": "auth_ok"})
        sub = await ws.receive_json()
        assert sub["type"] == "subscribe_events"
        await ws.send_json({"id": sub["id"], "type": "result", "success": True})
        get = await ws.receive_json()
        assert get["type"] == "get_states"
        await ws.send_json({"id": get["id"], "type": "result", "success": True, "result": [
            {"entity_id": eid, "state": "775.0"},
            {"entity_id": "sensor.something_else", "state": "hello"},
        ]})
        # Two real telegrams, both literal captured values: overnight import,
        # then the midday export peak.
        for val in ("775.0", "-5710.0"):
            await asyncio.sleep(0.5)
            await ws.send_json({"type": "event", "event": {"data": {
                "entity_id": eid,
                "new_state": {"entity_id": eid, "state": val}}}})
        await asyncio.sleep(0.5)
        await ws.send_json({"type": "event", "event": {"data": {
            "entity_id": eid,
            "new_state": {"entity_id": eid, "state": "unavailable"}}}})
        await asyncio.sleep(0.5)
        done.set()
        return ws
    srv = web.Application()
    srv.router.add_get("/ws", fake_ha)
    runner = web.AppRunner(srv)
    await runner.setup()
    site = web.TCPSite(runner, "127.0.0.1", 0)
    await site.start()
    port = site._server.sockets[0].getsockname()[1]
    p1mod.WS_URL = f"http://127.0.0.1:{port}/ws"
    ing = P1Ingest(phases=1, max_age_s=30.0)
    async with aiohttp.ClientSession() as sess:
        src = HaSignedSource(sess, ing, dict(NET), token="tok")
        task = asyncio.get_running_loop().create_task(src.run())
        try:
            await asyncio.wait_for(done.wait(), 20)
            await asyncio.sleep(0.5)
        finally:
            task.cancel()
            try:
                await task
            except asyncio.CancelledError:
                pass
    await runner.cleanup()
    return ing, src
 live, wire = asyncio.run(_e2e_signed())
 # ⚠️ TWO, not three - the same rule as the ha_dsmr e2e. get_states primes the
 # cache but must never become a sample: after a Core restart it is a
 # RestoreEntity value of unknown age, and stamping it with ingest_ts=now reports
 # a fresh meter that may have been dead for an hour.
 check("ha_signed does not manufacture a sample from cached HA state",
      live.samples == 2)
 check("the signed telegrams arrived over a real websocket",
      live.last.source == SOURCE_HA_SIGNED)
 check("the final export telegram nets negative, over the wire",
      live.last.net_w == -5710.0 and live.last.export_w == 5710.0)
 check("ha_signed subscribes to the one entity and caches nothing else",
      wire.ids == {"sensor.p1_meter_active_power"}
      and "sensor.something_else" not in wire.cache)
 check("a mid-stream unavailable signed state is a parse error, not a sample",
      live.parse_errors == 1 and live.samples == 2)
 # ⚠️ And the cached half is DROPPED, so no later telegram can be assembled out
 # of a value that stopped reporting.
 check("an unavailable entity is evicted from the cache", wire.cache == {})
 check("the last good reading survives the unavailable, and is not 0 W",
      live.net_w == -5710.0)
 check("the averager integrated the live signed stream", live.averager.elapsed_s > 0.2)
 # ⚠️ The entity FW-01 waits on. It must be a number here exactly as it is on the
 # other transports - the house P1 went 51.1 s and 36.2 s between state changes
 # overnight, and without this the watchdog false-trips the battery to 0 W.
 check("sensor.p1_sample_age_s is a live number on this transport too",
      isinstance(live.published_age_s, float) and live.published_age_s >= 0.0)
 # --------------------------------------------------------------------------- #
 print("ha_signed: selection by config")
 check("ha_signed is enabled", is_enabled({"meter_source": SOURCE_HA_SIGNED}) is True)
 # ⚠️ `sel`, not `built` - that name is the build() wrapper defined at the top of
 # this file, and rebinding it here silently disarms every check appended below
 # this line. Caught in review: an added check went `TypeError: 'HaSignedSource'
 # object is not callable` and aborted the suite, which is the exact failure the
 # wrapper exists to prevent, reintroduced by a name collision.
 sel = build_source({"meter_source": SOURCE_HA_SIGNED,
                    "p1_net_entity": "sensor.p1_meter_active_power"},
                   P1Ingest(), None, None)
 check("meter_source ha_signed selects the signed transport",
      isinstance(sel, HaSignedSource))
 check("...wired to p1_net_entity, and subscribed to exactly that one entity",
      sel.ids == {"sensor.p1_meter_active_power"})
 # ⚠️ The three modes must not bleed into each other: ha_dsmr must keep ignoring
 # p1_net_entity, or a half-configured install silently reads the wrong sensor.
 plain = build_source({"meter_source": SOURCE_HA,
                      "p1_import_entity": "sensor.i", "p1_export_entity": "sensor.e",
                      "p1_net_entity": "sensor.p1_meter_active_power"},
                     P1Ingest(), None, None)
 check("ha_dsmr still selects the unsigned transport and ignores p1_net_entity",
      type(plain) is HaDsmrSource and plain.ids == {"sensor.i", "sensor.e"})
 check("meter_source off still selects nothing",
      build_source({"meter_source": "off"}, P1Ingest(), None, None) is None)
 check("an unrecognised meter_source selects nothing rather than guessing",
      build_source({"meter_source": "ha_signd"}, P1Ingest(), None, None) is None)
 # ⚠️ Caught once at startup, not once per telegram. A source that is wired up
 # wrong otherwise fails in the one way indistinguishable from a healthy source
 # nobody has sent anything to yet: no samples, a climbing age, the watchdog
 # holding the battery at 0 W, and nothing in the log saying why.
 check("a blank p1_net_entity is refused rather than silently never receiving",
      build_source({"meter_source": SOURCE_HA_SIGNED, "p1_net_entity": ""},
                   P1Ingest(), None, None) is None)
 check("...and whitespace does not sneak past it",
      build_source({"meter_source": SOURCE_HA_SIGNED, "p1_net_entity": "   "},
                   P1Ingest(), None, None) is None)
 check("a phase list that disagrees with meter_phases is refused at startup",
      build_source({"meter_source": SOURCE_HA_SIGNED, "p1_net_entity": "sensor.n",
                    "p1_phase_net_entities": ["sensor.a", "sensor.b"]},
                   P1Ingest(phases=3), None, None) is None)
 check("a phase list that agrees with meter_phases is accepted",
      isinstance(build_source(
          {"meter_source": SOURCE_HA_SIGNED, "p1_net_entity": "sensor.n",
           "p1_phase_net_entities": ["sensor.a", "sensor.b", "sensor.c"]},
          P1Ingest(phases=3), None, None), HaSignedSource))
 check("no phase list at all is still fine - per-phase billing is optional",
      isinstance(build_source(
          {"meter_source": SOURCE_HA_SIGNED, "p1_net_entity": "sensor.n"},
          P1Ingest(phases=3), None, None), HaSignedSource))
 # --------------------------------------------------------------------------- #
 print("homewizard_local: parsing the meter's own /api/v1/data document")
 # The same JSON `sim/hwsim.py` serves and the same JSON the real meter at
 # 192.168.2.250 serves. One SIGNED connection figure plus three signed legs, so
 # it goes through the same split_signed as ha_signed - the sign convention is
 # not re-derived on this transport, it is reused.
 def hw_doc(w, l1=None, l2=None, l3=None, imp_kwh=1234.567, exp_kwh=890.123):
    return {"wifi_ssid": "sim", "smr_version": 50, "meter_model": "SIM-P1",
            "total_power_import_kwh": imp_kwh, "total_power_export_kwh": exp_kwh,
            "active_power_w": w,
            "active_power_l1_w": l1, "active_power_l2_w": l2,
            "active_power_l3_w": l3, "total_gas_m3": 0.0}
 s = parse_homewizard(hw_doc(775.0, l1=775.0), 1)
 check("the overnight base load (+775 W) is import on this transport too",
      s.net_w == 775.0 and s.import_w == 775.0 and s.export_w == 0.0)
 check("the sample is tagged homewizard_local", s.source == SOURCE_HOMEWIZARD)
 s = parse_homewizard(hw_doc(-5710.0, l1=-5710.0), 1)
 check("the midday PV peak (-5710 W) is export, not a 5.7 kW draw",
      s.net_w == -5710.0 and s.export_w == 5710.0)
 check("a single-phase document still yields its one leg", s.per_phase_w == (-5710.0,))
 # The TEL-04 three-phase survey reading, served the HomeWizard way.
 s = parse_homewizard(hw_doc(187.0, 2301.0, 468.0, -2582.0), 3)
 check("signed legs keep the exporting phase negative",
      s.per_phase_w == (2301.0, 468.0, -2582.0))
 check("per-phase import clamps the exporting leg out of the billed figure",
      s.per_phase_import_w == (2301.0, 468.0, 0.0))
 check("the legs sum to 2769 W while the connection nets 187 W",
      sum(s.per_phase_import_w) == 2769.0 and s.net_w == 187.0)
 # ⚠️ A single-phase HomeWizard serves `active_power_l2_w: null`. Unlike the HA
 # transports there is no "wait for the rest" case - every field came out of ONE
 # response, so a missing leg cannot be a STALE leg. Dropping the whole sample
 # over it would turn a healthy meter into a climbing age, which is the exact
 # false trip this transport exists to remove.
 def parsed(doc, phases=1):
    """parse_homewizard(), with an escaping exception turned into a visible value.
    Same reason as `built()` above: if the "all legs or none" guard goes missing,
    parsing raises, and without this the suite aborts on a traceback instead of
    reddening the NAMED check that says which rule died.
    """
    try:
        return parse_homewizard(doc, phases)
    except Exception as err:  # noqa: BLE001 - a raise here is itself the failure
        print(f"       parse_homewizard raised {type(err).__name__}: {err}")
        return err
 s = parsed(hw_doc(500.0, l1=500.0), 3)
 check("a meter serving fewer legs than meter_phases still gives a reading",
      getattr(s, "net_w", None) == 500.0 and getattr(s, "per_phase_w", "?") is None)
 check("...and does not fabricate a partial phase tuple",
      getattr(s, "per_phase_import_w", "?") is None)
 raises("a response that is not a JSON object is rejected",
       lambda: parse_homewizard([1, 2, 3], 1))
 raises("a document with no active_power_w is rejected, not read as 0 W",
       lambda: parse_homewizard({"total_gas_m3": 0.0}, 1))
 raises("a null active_power_w is rejected", lambda: parse_homewizard(hw_doc(None), 1))
 raises("a string active_power_w is rejected", lambda: parse_homewizard(hw_doc("775"), 1))
 raises("a NaN active_power_w is rejected",
       lambda: parse_homewizard(hw_doc(float("nan")), 1))
 raises("the 64954 signed-decode contamination is refused here too",
       lambda: parse_homewizard(hw_doc(64954.0), 1))
 raises("a non-numeric leg is rejected rather than quietly dropped",
       lambda: parse_homewizard(hw_doc(600.0, "2000", 0.0, 100.0), 3))
 # --------------------------------------------------------------------------- #
 print("homewizard_local: every HTTP response is an arrival")
 # THE WHOLE TICKET. sensor.p1_sample_age_s has to mean "time since the meter
 # REPORTED", not "time since the value CHANGED". Home Assistant cannot express
 # the first: a repeated reading emits no state_changed, advances last_reported
 # on neither serialiser, and state_reported cannot be subscribed to at all
 # ("Event filter is required"). Measured on the ENV-01 rig over 70 s of a frozen
 # meter and again against the live house over ten repeated readings. Our own
 # capture of this house's meter goes 42.2 s and 97.0 s between changes, both past
 # the default meter_max_age_s of 30 - i.e. the age sensor would command 0 W on a
 # perfectly healthy meter. Polling the meter itself removes that: the response is
 # the arrival, and the number in it is not consulted.
 class _FakeMeter:
    """hwsim in eight lines: serves whatever document it is told to, or a fault."""
    def __init__(self, doc=None):
        self.doc = doc
        self.status = 200
        self.body = None          # set to raw text to serve something unparseable
        self.delay = 0.0          # set to seconds to imitate a meter that hangs
        self.hits = 0
    async def handle(self, request):
        from aiohttp import web
        self.hits += 1
        if self.delay:
            await asyncio.sleep(self.delay)
        # ⚠️ web.Response(text=...) defaults to text/plain. That is not just the
        # "unparseable body" path: fed VALID json it serves a good document under
        # the wrong mimetype, which is the only way to reach the content_type
        # guard in poll_once - json_response can never produce it.
        if self.body is not None:
            return web.Response(text=self.body, status=self.status)
        return web.json_response(self.doc, status=self.status)
 async def _hw_rig(fn):
    """Run `fn(make_source, meter)` against a real HTTP server on localhost."""
    from aiohttp import web
    meter = _FakeMeter(hw_doc(350.0, l1=350.0))
    app = web.Application()
    app.router.add_get("/api/v1/data", meter.handle)
    runner = web.AppRunner(app)
    await runner.setup()
    site = web.TCPSite(runner, "127.0.0.1", 0)
    await site.start()
    port = site._server.sockets[0].getsockname()[1]
    async with aiohttp.ClientSession() as sess:
        try:
            return await fn(sess, port, meter)
        finally:
            await runner.cleanup()
 async def _arrivals(sess, port, meter):
    ing = P1Ingest(phases=1, max_age_s=30.0)
    src = HomeWizardLocalSource(sess, ing, "127.0.0.1", port=port, poll_s=1.0)
    out = {}
    out["first"] = await src.poll_once()
    out["stamp1"] = ing.last.ingest_mono
    # Four more polls of the IDENTICAL document - the meter has not moved a watt.
    for _ in range(4):
        await asyncio.sleep(0.05)
        await src.poll_once()
    out["ing"], out["src"], out["meter"] = ing, src, meter
    return out
 r = asyncio.run(_hw_rig(_arrivals))
 ing, src = r["ing"], r["src"]
 check("a poll of the meter's own API builds a sample",
      r["first"] is True and ing.net_w == 350.0)
 # ⚠️ THE acceptance criterion. Five identical readings, five arrivals. On
 # ha_signed this whole sequence produces exactly ONE state_changed and then
 # silence, and the age climbs to 30 s on a meter that is answering perfectly.
 check("five identical readings are five arrivals, not one",
      ing.samples == 5 and r["meter"].hits == 5 and src.polls == 5)
 check("an unchanged value still stamps a NEW arrival time",
      ing.last.ingest_mono > r["stamp1"])
 check("...so the age resets on a response carrying an unchanged number",
      ing.published_age_s < 1.0 and ing.stale is False)
 check("no arrival was ever a parse error", ing.parse_errors == 0)
 # --------------------------------------------------------------------------- #
 print("homewizard_local: frozen vs steady, the pair HA cannot separate")
 # ⚠️ Read this before changing anything here. A frozen meter (hwsim --fault
 # freeze) answers 200 OK forever with a stale document. It is ARRIVING. So the
 # age - an arrival detector, correctly - reads fresh on both, and that is not a
 # defect in the age, it is the definition of the signal. What the local API adds
 # that Home Assistant never had is the ENERGY REGISTERS: a meter under real load
 # advances total_power_import_kwh (1 Wh resolution, ~10 s at 350 W) even when the
 # power figure repeats, and a frozen one does not. That is the discriminator, and
 # it is exposed as `unchanged_s` - deliberately NOT folded into the age, because
 # this controller regulates grid power toward ~0 W and at a converged -10 W the
 # export register needs six minutes to move. Thresholding unchanged_s at 30 s
 # would rebuild the false-trip limit cycle at the exact operating point we aim
 # for. See DOCS.md and HomeWizardLocalSource.unchanged_s.
 async def _freeze_vs_steady(sess, port, meter):
    ing_f = P1Ingest(phases=1, max_age_s=30.0)
    frozen = HomeWizardLocalSource(sess, ing_f, "127.0.0.1", port=port, poll_s=1.0)
    meter.doc = hw_doc(350.0, l1=350.0)          # --fault freeze: never moves
    for i in range(4):
        if i:
            await asyncio.sleep(0.1)             # sleep BEFORE, so unchanged_s
        await frozen.poll_once()                 # is read the instant a poll lands
    ing_s = P1Ingest(phases=1, max_age_s=30.0)
    steady = HomeWizardLocalSource(sess, ing_s, "127.0.0.1", port=port, poll_s=1.0)
    kwh = 1234.567
    for i in range(4):
        if i:
            await asyncio.sleep(0.1)
        # A steady 350 W house: the power figure repeats, the register climbs.
        kwh += 0.001
        meter.doc = hw_doc(350.0, l1=350.0, imp_kwh=round(kwh, 3))
        await steady.poll_once()
    return (ing_f, frozen), (ing_s, steady)
 (ing_f, frozen), (ing_s, steady) = asyncio.run(_hw_rig(_freeze_vs_steady))
 check("a frozen meter keeps arriving, so both read the same power",
      ing_f.net_w == 350.0 and ing_s.net_w == 350.0)
 # ⚠️ Recorded as a rule, not a shortcoming: the age is an ARRIVAL detector and a
 # frozen meter genuinely is arriving. Anyone tempted to make the age catch freeze
 # is about to reintroduce the false trip on a steady house.
 check("the age cannot separate them, and is fresh on both",
      ing_f.published_age_s < 1.0 and ing_s.published_age_s < 1.0)
 check("the frozen meter's measurement stands still", frozen.unchanged_s > 0.25)
 check("...while the steady meter's energy register keeps advancing",
      steady.unchanged_s < 0.05)
 check("so the two ARE separable on the local API, which HA could not do",
      frozen.unchanged_s > steady.unchanged_s * 3)
 check("unchanged_s is None before the first response ever lands",
      HomeWizardLocalSource(None, P1Ingest(), "h").unchanged_s is None)
 # --------------------------------------------------------------------------- #
 print("homewizard_local: a failed poll is a missing reading, never 0 W")
 async def _failures(sess, port, meter):
    ing = P1Ingest(phases=1, max_age_s=30.0)
    src = HomeWizardLocalSource(sess, ing, "127.0.0.1", port=port, poll_s=1.0)
    await src.poll_once()
    # Age the good sample by hand so a reset would be unmistakable.
    ing.submit(make_sample(SOURCE_HOMEWIZARD, 350.0, 0.0, phases=1,
                           ingest_mono=time.monotonic() - 90.0))
    out = {}
    meter.status = 500
    out["http500"] = await src.poll_once()
    out["conn_after_500"] = src.connected
    meter.status, meter.body = 200, "<html>gateway</html>"
    out["notjson"] = await src.poll_once()
    meter.body = None
    meter.doc = {"total_gas_m3": 0.0}            # 200 OK, no power in it
    out["nopower"] = await src.poll_once()
    out["ing"], out["src"] = ing, src
    # And a meter that is not listening at all: `--fault down`.
    dead = HomeWizardLocalSource(sess, P1Ingest(phases=1, max_age_s=30.0),
                                 "127.0.0.1", port=1, poll_s=1.0)
    out["down"] = await dead.poll_once()
    out["dead"] = dead
    return out
 r = asyncio.run(_hw_rig(_failures))
 ing, src = r["ing"], r["src"]
 check("an HTTP 500 is not a reading", r["http500"] is False)
 # A source that was connected and then failed must SAY it is disconnected -
 # otherwise the diagnostic reads healthy while the age climbs, which is exactly
 # the "connected: True, parse_errors: 0, samples: 0" state the rig recorded.
 check("a poll that fails clears the connected flag", r["conn_after_500"] is False)
 check("a 200 OK carrying something that is not JSON is not a reading",
      r["notjson"] is False)
 check("a 200 OK with no active_power_w in it is not a reading", r["nopower"] is False)
 check("a meter that refuses the connection is not a reading",
      r["down"] is False and r["dead"].connected is False)
 check("every failure was counted as a parse error", ing.parse_errors == 3)
 check("...and the transport records the transport-level ones separately",
      src.poll_errors == 2 and src.polls == 4)
 # ⚠️ The rule the safety chain rests on. A failed poll must not manufacture a
 # balanced house, and must not reset the clock the watchdog reads.
 check("a failed poll does not reset the age", ing.published_age_s > 89)
 check("a failed poll leaves the last good value in place, and it is not 0 W",
      ing.last.net_w == 350.0 and ing.stale is True and ing.net_w is None)
 check("no failed poll ever became a sample", ing.samples == 2)
 check("a 200 OK marks the transport connected even when its body is refused",
      src.connected is True)
 # --------------------------------------------------------------------------- #
 print("homewizard_local: a wrong header, a hung meter, a submit that throws")
 # Three failure shapes that all end the same way if they are mishandled - no
 # sample, a climbing age, the battery at 0 W - and each of which would send the
 # operator hunting the wrong device.
 async def _header_and_timeout(sess, port, meter):
    out = {}
    # Valid JSON under text/plain: exactly what content_type=None is for.
    meter.body = json.dumps(hw_doc(350.0, l1=350.0))
    ing = P1Ingest(phases=1, max_age_s=30.0)
    src = HomeWizardLocalSource(sess, ing, "127.0.0.1", port=port, poll_s=1.0)
    out["mimetype"] = await src.poll_once()
    out["ing"] = ing
    meter.body = None
    # A meter that takes the connection and then does not answer.
    meter.delay = 0.6
    ing_t = P1Ingest(phases=1, max_age_s=30.0)
    slow = HomeWizardLocalSource(sess, ing_t, "127.0.0.1", port=port,
                                 poll_s=1.0, timeout_s=0.05)
    out["timeout"] = await slow.poll_once()
    out["timeout_err"] = ing_t.last_error
    meter.delay = 0.0
    return out
 r = asyncio.run(_hw_rig(_header_and_timeout))
 # ⚠️ A real firmware answering text/plain must not read as a dead meter. Drop
 # content_type=None from poll_once and this goes red with "unexpected mimetype".
 check("valid JSON under the wrong Content-Type is still a reading",
      r["mimetype"] is True and r["ing"].samples == 1 and r["ing"].net_w == 350.0)
 # ⚠️ str(asyncio.TimeoutError()) is the EMPTY STRING. Without the class-name
 # fallback the status page reads "last error:" and then nothing, on a hung
 # meter, at the moment someone is reading that line to find out why the battery
 # went to 0 W.
 check("a timed-out poll names the fault instead of logging an empty reason",
      r["timeout"] is False and "TimeoutError" in (r["timeout_err"] or ""))
 async def _submit_rejects(sess, port, meter):
    """submit() raising P1Error must be a rejection, not an escaping exception."""
    class _P1Boom(P1Ingest):
        def submit(self, sample):
            raise P1Error("register went backwards")
    ing = _P1Boom(phases=1, max_age_s=30.0)
    src = HomeWizardLocalSource(sess, ing, "127.0.0.1", port=port, poll_s=1.0)
    try:
        ok = await src.poll_once()
    except Exception as err:      # noqa: BLE001 - an escape IS the failure
        ok = err
    return ok, ing.parse_errors
 ok, errs = asyncio.run(_hw_rig(_submit_rejects))
 check("a submit that rejects the sample is handled, not left to escape",
      ok is False and errs == 1)
 async def _submit_explodes(sess, port, meter):
    """And an UNEXPECTED raise must not kill the poll task for good."""
    class _Boom(P1Ingest):
        def submit(self, sample):
            raise RuntimeError("kaboom")
    src = HomeWizardLocalSource(sess, _Boom(phases=1, max_age_s=30.0),
                                "127.0.0.1", port=port, poll_s=0.05)
    task = asyncio.create_task(src.run())
    await asyncio.sleep(0.3)
    alive = not task.done()
    task.cancel()
    # ⚠️ BaseException, and the same reason built() exists: if run() loses its
    # guard the task is already dead HOLDING the RuntimeError, and awaiting it
    # re-raises - aborting the whole suite with a traceback instead of reddening
    # the check that names the rule. The death is what `alive` records; catching
    # it here is bookkeeping, not leniency.
    try:
        await task
    except BaseException:  # noqa: BLE001
        pass
    return alive, src.polls
 # ⚠️ The silent-death case. A dead poll task fails SAFE - the age climbs and the
 # controller commands 0 W - but it looks exactly like a dead meter, so the
 # operator spends the outage power-cycling hardware that was never at fault.
 alive, polls = asyncio.run(_hw_rig(_submit_explodes))
 check("a raise inside a poll does not permanently kill the polling task",
      alive is True and polls > 1)
 # --------------------------------------------------------------------------- #
 print("homewizard_local: selection by config")
 check("homewizard_local is enabled", is_enabled({"meter_source": SOURCE_HOMEWIZARD}) is True)
 sel = build_source({"meter_source": SOURCE_HOMEWIZARD, "p1_host": "192.168.2.250"},
                   P1Ingest(), None, None)
 check("meter_source homewizard_local selects the polling transport",
      isinstance(sel, HomeWizardLocalSource))
 # ⚠️ getattr, not attribute access, for the same reason `built()` exists: a
 # startup guard that goes wrong returns None here, and `None.url` would abort the
 # suite with a traceback instead of reddening the check that names the rule.
 check("...pointed at the meter's own local API on the default port",
      getattr(sel, "url", None) == "http://192.168.2.250:80/api/v1/data")
 check("the default cadence is the meter's own ~5 s update rate",
      getattr(sel, "poll_s", None) == 5.0)
 # ⚠️ The request must not outlive the poll interval: a backlog of queued requests
 # behind a slow meter would land several arrivals for one measurement.
 check("the request timeout is held under the poll interval",
      getattr(sel, "timeout_s", 99) < getattr(sel, "poll_s", 0))
 sel = build_source({"meter_source": SOURCE_HOMEWIZARD, "p1_host": "10.0.0.9:8080"},
                   P1Ingest(), None, None)
 check("an explicit host:port is honoured, which is how the sim is reached",
      getattr(sel, "url", None) == "http://10.0.0.9:8080/api/v1/data")
 sel = build_source({"meter_source": SOURCE_HOMEWIZARD,
                    "p1_host": "http://192.168.2.250/"}, P1Ingest(), None, None)
 check("a pasted browser URL still resolves to the right host",
      getattr(sel, "url", None) == "http://192.168.2.250:80/api/v1/data")
 # ⚠️ Caught once at startup, not once per poll - a source wired up wrong
 # otherwise fails in the one way indistinguishable from a healthy one nobody has
 # polled yet: no samples, a climbing age, the watchdog at 0 W, nothing in the log.
 check("a blank p1_host is refused rather than silently never polling",
      build_source({"meter_source": SOURCE_HOMEWIZARD, "p1_host": ""},
                   P1Ingest(), None, None) is None)
 check("...and whitespace does not sneak past it",
      build_source({"meter_source": SOURCE_HOMEWIZARD, "p1_host": "   "},
                   P1Ingest(), None, None) is None)
 check("an unparseable port is refused rather than guessed",
      build_source({"meter_source": SOURCE_HOMEWIZARD, "p1_host": "meter:eighty"},
                   P1Ingest(), None, None) is None)
 # ⚠️ The age can never be fresher than the poll interval, so a cadence at or past
 # max_age_s means every reading is stale before its successor arrives: the
 # controller would sit permanently on missing inputs while the meter is fine.
 check("a poll cadence at meter_max_age_s is refused",
      build_source({"meter_source": SOURCE_HOMEWIZARD, "p1_host": "m",
                    "meter_poll_s": 30}, P1Ingest(max_age_s=30), None, None) is None)
 check("...and one past it too",
      build_source({"meter_source": SOURCE_HOMEWIZARD, "p1_host": "m",
                    "meter_poll_s": 45}, P1Ingest(max_age_s=30), None, None) is None)
 check("a cadence with headroom is accepted",
      isinstance(build_source({"meter_source": SOURCE_HOMEWIZARD, "p1_host": "m",
                               "meter_poll_s": 5}, P1Ingest(max_age_s=30),
                              None, None), HomeWizardLocalSource))
 check("a zero cadence is refused rather than spinning",
      build_source({"meter_source": SOURCE_HOMEWIZARD, "p1_host": "m",
                    "meter_poll_s": 0}, P1Ingest(max_age_s=30), None, None) is None)
 check("a non-numeric cadence is refused",
      build_source({"meter_source": SOURCE_HOMEWIZARD, "p1_host": "m",
                    "meter_poll_s": "fast"}, P1Ingest(max_age_s=30), None, None) is None)
 # The four modes must not bleed into each other.
 check("ha_signed ignores p1_host and still selects the entity transport",
      type(build_source({"meter_source": SOURCE_HA_SIGNED, "p1_net_entity": "sensor.n",
                         "p1_host": "192.168.2.250"}, P1Ingest(), None, None))
      is HaSignedSource)
 check("homewizard_local ignores p1_net_entity and needs its own host",
      build_source({"meter_source": SOURCE_HOMEWIZARD,
                    "p1_net_entity": "sensor.n"}, P1Ingest(), None, None) is None)
 # --------------------------------------------------------------------------- #
 print("the age sensor must not exist when P1 is off")
 # ⚠️ This is a fleet-wide regression guard, not a nicety. The ESP32 watchdog
 # does `id(p1_age_s).has_state() && id(p1_age_s).state >= max_age_s` and forces
 # the layer-1 failsafe. published_age_s counts from P1Ingest.__init__, so if the
 # age were published with meter_source off it would climb past 30 s on every
 # existing install within half a minute and pin the inverter at 0 W forever.
 from app.mqtt import SENSORS, MqttPublisher   # noqa: E402
 check("meter_source off is disabled", is_enabled({"meter_source": "off"}) is False)
 check("a missing meter_source is disabled", is_enabled({}) is False)
 check("an empty meter_source is disabled", is_enabled({"meter_source": ""}) is False)
 check("ha_dsmr is enabled", is_enabled({"meter_source": SOURCE_HA}) is True)
 check("mqtt_p1 is enabled", is_enabled({"meter_source": SOURCE_MQTT}) is True)
 check("homewizard_local is enabled here too",
      is_enabled({"meter_source": SOURCE_HOMEWIZARD}) is True)
 # The entity id SAFETY-01's firmware subscribes to, pinned by object_id.
 row = [s for s in SENSORS if s[0] == "p1_age"]
 check("the age sensor is declared exactly once", len(row) == 1)
 check("its object_id pins entity_id to sensor.p1_sample_age_s",
      row[0][1] == "p1_sample_age_s")
 check("it is published in seconds", row[0][3] == "s")
 class _RecordingClient:
    def __init__(self):
        self.sent = []
    def publish(self, topic, payload=None, retain=False):
        # Topic AND payload: object_id, the thing that actually pins the entity
        # id, only appears in the discovery payload. Recording topics alone made
        # the "is not announced" check pass for the wrong reason.
        self.sent.append(f"{topic} {payload}")
 def _announced(omit):
    pub = MqttPublisher(None, 1883, omit=omit)   # host None -> never connects
    pub.client = _RecordingClient()
    pub._announce()
    return " ".join(pub.client.sent)
 check("with P1 off the age sensor is never announced",
      "p1_sample_age_s" not in _announced(("p1_age",)))
 check("the other status entities are still announced with P1 off",
      "goodwe_grid_power" in _announced(("p1_age",)))
 check("with P1 on the age sensor IS announced",
      "p1_sample_age_s" in _announced(()))
 # And the publish dict itself, through the real Controller.
 from app.main import Controller            # noqa: E402
 class _Store:
    data = {}
    def set(self, *a):
        pass
    def get_time(self, *a):
        return None
 class _Pub:
    def __init__(self):
        self.last = {}
    def publish(self, values):
        self.last = values
    def close(self):
        pass
 pub_off = _Pub()
 Controller({"meter_source": "off"}, None, _Store(), pub_off).publish()
 check("with P1 off, p1_age is absent from the published payload",
      "p1_age" not in pub_off.last)
 check("...while the normal status keys are still published",
      "setpoint" in pub_off.last and "grid" in pub_off.last)
 pub_on = _Pub()
 Controller({"meter_source": SOURCE_HA}, None, _Store(), pub_on).publish()
 check("with P1 on, p1_age is published", "p1_age" in pub_on.last)
 check("...as a number, so has_state() becomes true only once we feed it",
      isinstance(pub_on.last["p1_age"], float))
 # ⚠️ And on ha_signed identically - this is the whole reason TEL-04 exists. The
 # age sensor is a hard prerequisite for the FW-01 flash, and it has to appear on
 # the transport that can actually read the meter in this house.
 pub_sig = _Pub()
 Controller({"meter_source": SOURCE_HA_SIGNED}, None, _Store(), pub_sig).publish()
 check("with ha_signed selected, p1_age is published too",
      "p1_age" in pub_sig.last and isinstance(pub_sig.last["p1_age"], float))
 # ⚠️ And on homewizard_local, where it is the one age FW-01 may actually
 # threshold - every other transport's age measures when the VALUE changed.
 pub_hw = _Pub()
 Controller({"meter_source": SOURCE_HOMEWIZARD}, None, _Store(), pub_hw).publish()
 check("with homewizard_local selected, p1_age is published",
      "p1_age" in pub_hw.last and isinstance(pub_hw.last["p1_age"], float))
 # --------------------------------------------------------------------------- #
 print("unchanged_s has somewhere an operator can read it")
 # ⚠️ The counter is deliberately NOT thresholded and NOT folded into the age -
 # at the converged -10 W this controller aims for, a 1 Wh register needs six
 # minutes to move, so any limit false-trips at the target operating point. That
 # refusal only holds up if a HUMAN can interpret the number instead, and DOCS.md
 # tells them "the transport tracks it as unchanged_s". Before this, nothing ever
 # read the transport object back: connected, polls, poll_errors and unchanged_s
 # were all write-only, and the documented signal existed nowhere an operator
 # could see it.
 def p1_rows(ctl):
    """The P1 status line(s), or the exception that stopped checks() making one.
    Same reason as built(): a status line that raises would abort the suite with
    a traceback instead of reddening the check that names the rule.
    """
    try:
        return [c["text"] for c in ctl.checks() if c["text"].startswith("P1 meter")]
    except Exception as err:  # noqa: BLE001 - a raise here is itself the failure
        print(f"       checks() raised {type(err).__name__}: {err}")
        return err
 def _fed(source):
    ctl = Controller({"meter_source": source}, None, _Store(), _Pub())
    ctl.p1.submit(make_sample(source, 350.0, 0.0, phases=1,
                              ingest_mono=time.monotonic()))
    return ctl
 hw_ctl = _fed(SOURCE_HOMEWIZARD)
 hw_src = HomeWizardLocalSource(None, hw_ctl.p1, "127.0.0.1")
 hw_src._note(hw_doc(350.0, l1=350.0))
 hw_ctl.p1_source = hw_src            # what amain() does once the transport exists
 rows = p1_rows(hw_ctl)
 check("the healthy P1 status line reports the transport's unchanged_s",
      isinstance(rows, list) and len(rows) == 1 and "unchanged" in rows[0])
 # ⚠️ getattr, not attribute access: p1_source is None until amain() builds one,
 # and ha_dsmr/ha_signed/mqtt_p1 have no such counter at all. Reaching for it
 # directly would turn the whole status page into a 500 on every other transport.
 ha_rows = p1_rows(_fed(SOURCE_HA))
 check("...and the line is unharmed on a transport that has no such counter",
      isinstance(ha_rows, list) and len(ha_rows) == 1 and "unchanged" not in ha_rows[0])
 print()
 if fails: