TEL-01 review fixes: stop the age sensor tripping installs that have no meter

T-1, and it was a fleet-wide trip to zero. publish() emitted p1_age unconditionally, and published_age_s counts from P1Ingest.__init__ when no sample has ever arrived. With meter_source defaulting to off, every existing install would have published sensor.p1_sample_age_s climbing without bound; the ESP32 does `has_state() && state >= max_age_s` and forces the layer-1 failsafe, so each of them would have pinned its inverter at 0 W within 30 s. Exactly the opposite of the zero-regression the off default was for. The key is now omitted from the payload AND from MQTT discovery when P1 is off, so the entity does not exist at all - which is the status quo, and what has_state() is testing for. The predicate is one function, is_enabled(), because the grid reading, the task start and the discovery announcement have to agree or this comes back. T-2, connect no longer manufactures a sample. get_states returns whatever HA currently holds, which after a Core restart is a RestoreEntity value of unknown age; stamping it with ingest_ts=now reset the age and reported a fresh meter that could have been dead for an hour. run()'s own docstring already said a reconnect must emit nothing - the code disagreed with it, and a test asserted the violation. The cache is still primed, so the first real state_changed builds a complete sample; the age just stays honest until one arrives. T-3, gaps are no longer filled with the last held value. The averager held a sample forward across any interval, so a meter dying at 5 kW and returning ten minutes later credited 5 kW x 600 s to the capacity-tariff accumulator - a fabricated peak on a permanent record. The hold is capped at max_age_s: past that the stretch is walked so block boundaries still land correctly, but nothing accumulates and elapsed does not grow, which is what finally makes the comment about a gap dragging the billed average down true. Same threshold for control and billing: a reading too old to steer by is too old to bill by. T-4, the out-of-order/duplicate guard is covered. It was untested, and the reason is worth recording: the obvious assertion passes without the guard, because the negative interval is separately refused by the covered > 0 test. What the guard prevents is the timestamp REWIND, which only shows up one sample later as a re-integrated window. The test now goes one sample later. T-6, DOCS was wrong about latency. meter_max_age_s and stale_input_s stack, so meter death to 0 W is 45 s and not 30. Documented as a table with both clocks. Also documented the T-5 asymmetry rather than papering over it: the age measures arrival, not change, so a stuck MQTT bridge republishing its last telegram still looks fresh. Correct on ha_dsmr, not detectable on mqtt_p1 without a change-detector. Written up as a known limit. Writing the T-1 test caught a second defect in the test itself: it recorded only MQTT topics, and object_id lives in the payload, so "the age sensor is not announced" had been passing for the wrong reason. test_p1.py: 99 -> 122 checks. 14 mutations run, all 14 red, files restored byte-identical - including one per fix above. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Du77usMj8XNKNFZGmUiWDa
TEL-01: P1 ingestion, with the derivation and the age the EMS owns
2026-08-24 22:16:34 +02:00 · 2026-08-24 22:16:34 +02:00
9 changed files with 44 additions and 1790 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,67 +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_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
@@ -152,55 +106,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
@@ -218,48 +123,13 @@ 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
+> **Known limit, `mqtt_p1` only.** The age measures *arrival*, not change. On the
-> bridge that is stuck republishing its last telegram keeps arriving, so the age
+> `ha_dsmr` path that is exactly right: a frozen meter emits no `state_changed`,
-> stays near zero and a frozen meter still looks fresh. Detecting *that* needs a
+> so nothing arrives and the age climbs. On the MQTT path a bridge that is stuck
-> change-detector rather than an arrival-detector, and it is not in this version.
+> 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
-> ⚠️ **Known limit, `ha_signed` — do not drive a watchdog off this age yet.**
+> than an arrival-detector, and it is not in this version. Prefer `ha_dsmr` where
-> On the HA WebSocket paths the age is stamped when a `state_changed` arrives,
+> both are available.
 > 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
@@ -272,35 +142,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 |
 | `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" |
 | `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
@@ -156,45 +156,15 @@ def compute(
        # 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.
-        #
+        if not frozen:
        # ⚠️ 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)
+            i_w = min(moved, i_w) if i_w > 0 else max(moved, i_w)
    # ⚠️ Applied EVERY cycle, frozen or not: the freeze is conditional, this
    # bound is not. It is what makes the worst-case unwind time finite and
    # knowable instead of a function of how long the error happened to stand.
-    bounded = max(-limit, min(limit, i_w))
+    i_w = max(-limit, min(limit, i_w))
    if bounded != 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
@@ -97,10 +97,6 @@ class Controller:
        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)
        # 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)))
@@ -369,22 +365,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:
@@ -556,7 +540,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()
@@ -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,
@@ -702,333 +668,6 @@ 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
 # --------------------------------------------------------------------------- #
@@ -1060,84 +699,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
@@ -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)
@@ -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)
@@ -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))
@@ -179,12 +113,6 @@ check(f"runaway with the detector defeated: integrator still bounded ({wi:.0f} W
      wi <= TD.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
 # 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,
@@ -193,22 +121,11 @@ check("integrator bound binds independently of the output clamp",
      d.i_w == 1000 and d.target_w == 1000)
 # Freeze = may not wind further in the direction it is already pushing.
-# ⚠️ 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)
 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
@@ -228,50 +145,6 @@ check("frozen at i_w=0: the freeze then clears", not z2.frozen)
 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.
@@ -323,67 +196,34 @@ 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 = prev if abs(error) < t.deadband_w else prev + t.gain * error
        want, reason = prev, "deadband"
    else:
        want = prev + t.gain * error
    target = max(-t.max_w, min(t.max_w, want))
-    if target != want:
+    target = max(prev - t.slew_w, min(prev + t.slew_w, target))
        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
+    return float(round(target / step) * step), sc
-# ⚠️ Compare EVERYTHING observable, not just the number. A previous version of
+# Exhaustive over the interesting corners, both freeze states, both signs, and
-# this sweep compared (target_w, sat_count) only and passed 3024 cases while
+# either side of the deadband. This is what makes the claim in control.py's
-# `reason` had silently lost a value - which is the kind of thing a sweep this
+# integrator comment a checked fact rather than an assertion.
 # 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)
+                    lt, lsc = legacy(prev, grid, actual, tune, sc)
-                    got = (d.target_w, d.sat_count, d.frozen,
+                    if (d.target_w, d.sat_count) != (lt, lsc):
-                           ALIAS.get(d.reason, d.reason))
+                        diffs.append((prev, grid, actual, sc, d.target_w, lt))
-                    if got != legacy(prev, grid, actual, tune, sc):
+check(f"i_w=None reproduces release/1.0 over {3*7*9*4*4} cases"
                        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,
@@ -571,726 +548,6 @@ check("the averager integrated the live stream", live.averager.elapsed_s > 0.2)
 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
@@ -1299,6 +556,7 @@ print("the age sensor must not exist when P1 is off")
 # 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.p1 import is_enabled          # noqa: E402
 from app.mqtt import SENSORS, MqttPublisher   # noqa: E402
 check("meter_source off is disabled", is_enabled({"meter_source": "off"}) is False)
@@ -1306,8 +564,6 @@ 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"]
@@ -1380,68 +636,6 @@ 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:
    print(f"{len(fails)} of {total} FAILED: {', '.join(fails)}")