TEL-05 review follow-ups: show unchanged_s, name a timeout, keep the poll task alive

Four follow-ups on the reviewed and approved TEL-05 work. Additive; no shipped
behaviour changes except the two failure paths below.

1. unchanged_s had no operator surface. DOCS.md told a reader "the transport
   tracks it as unchanged_s" and there was nowhere to look: main.py built the
   transport, scheduled run(), and never read the object again. The status page
   now shows it on the healthy P1 line. Still NOT thresholded and NOT folded
   into the age - that refusal was reviewed and upheld, because 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 target operating point. The whole argument for
   leaving it to a human requires the human being able to see it.

2. The "equivalent mutant" note on the content_type guard was wrong, and the
   comment is downgraded to say so. web.Response(text=...) defaults to
   text/plain, so the fake meter CAN serve valid JSON under the wrong mimetype.
   Test added; shipped behaviour was already correct.

3. A timed-out poll logged an empty reason: str(asyncio.TimeoutError()) is "",
   so the status page read "last error:" and then nothing, on a hung meter, at
   the moment the battery had just gone to 0 W. Falls back to the class name.
   Note str(err), not `err or ...` - an exception object is always truthy.

4. submit() sat outside the try in poll_once() and run() had no except, so a
   raise would kill the poll task permanently and SILENTLY - safe (the age
   climbs, the controller commands 0 W) but indistinguishable from a dead meter.
   Both wrapped; poll_s is already the retry cadence, so no backoff.

Also a comment at the parse_homewizard range(phases) slice: a 3-phase meter
configured as 1-phase understates the capacity-tariff figure. Filed separately,
not fixed here.

242 checks in test_p1.py (236 before, 6 new). test_control 55, test_arbiter 18,
test_maintenance 21, all untouched and green. Each new check proved non-vacuous:
six mutations, six named reds, no suite aborts, sources restored byte-identical.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Du77usMj8XNKNFZGmUiWDa

.gitattributes

@@ -0,0 +1,10 @@
+# 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

goodwe_controller/CHANGELOG.md

@@ -1,5 +1,115 @@
 # 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

goodwe_controller/DOCS.md

@@ -51,21 +51,67 @@ phase having charged nothing.
 ### P1 meter ingestion
 
 `meter_entity` above expects one signed sensor, which usually means a template
-someone wrote by hand. A Belgian P1 meter does not publish one: it publishes two
-**unsigned** registers, consumption and injection. Setting `meter_source` moves
-that subtraction into the add-on, where it is done once and tested, and replaces
-`meter_entity` entirely.
+someone wrote by hand. Setting `meter_source` moves the whole derivation 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 |
+| `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_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
 
@@ -106,6 +152,55 @@ 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
@@ -123,13 +218,48 @@ 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` only.** The age measures *arrival*, not change. On the
-> `ha_dsmr` path that is exactly right: a frozen meter emits no `state_changed`,
-> so nothing arrives and the age climbs. On the MQTT path a bridge that is stuck
-> republishing its last telegram keeps arriving, so the age stays near zero and a
-> frozen meter still looks fresh. Detecting *that* needs a change-detector rather
-> than an arrival-detector, and it is not in this version. Prefer `ha_dsmr` where
-> both are available.
+> **Known limit, `mqtt_p1`.** The age measures *arrival*. On the MQTT path a
+> bridge that is stuck republishing its last telegram keeps arriving, so the age
+> stays near zero and a frozen meter still looks fresh. Detecting *that* needs a
+> change-detector rather than an arrival-detector, and it is not in this version.
+
+> ⚠️ **Known limit, `ha_signed` — do not drive a watchdog off this age yet.**
+> On the HA WebSocket paths the age is stamped when a `state_changed` arrives,
+> which means it measures *time since the value last changed*, not time since
+> the meter last reported. Home Assistant offers nothing better: a repeated
+> reading produces no `state_changed`, does **not** advance `last_reported` on
+> either the REST or the WebSocket serialiser, and `state_reported` cannot be
+> subscribed to over the WebSocket at all (`Event filter is required for event
+> state_reported`). All three measured on the ENV-01 rig against the real
+> HomeWizard integration with the meter frozen: 0 `state_changed` in 70 s and no
+> timestamp movement anywhere.
+>
+> `ha_dsmr` mostly escapes this because a DSMR telegram updates several entities
+> and something in the set almost always moves. **`ha_signed` has exactly one
+> entity, so a healthy meter under a flat load is indistinguishable from a dead
+> one.** This is not hypothetical: in our own captures
+> (`sim/scenarios/ha-p1_meter_active_power-2026-08-20.json`) the real house meter
+> went **42.2 s and 97.0 s** between changes, and 23 Aug peaks at 29.1 s — all
+> past the default `meter_max_age_s` of 30.
+>
+> So `sensor.p1_sample_age_s` on `ha_signed` is safe to *read*, and it is
+> correct whenever the value is moving, but it must not yet be thresholded by
+> the ESP32 stale-input watchdog: a quiet house would trip the battery to 0 W.
+> Raising `meter_max_age_s` is **not** the fix — the two conditions produce an
+> identical signal, so a bigger number only chooses which of the two errors you
+> get. The real fix is an arrival stamp the meter itself provides — and that now
+> exists: **`meter_source: homewizard_local`**. If you have a HomeWizard P1,
+> switch to it. If your meter is only reachable through Home Assistant, this
+> limit still applies to you and the watchdog threshold still must not be armed.
+
+**Where the age is trustworthy:**
+
+| mode | the age measures | safe to threshold from firmware |
+|---|---|---|
+| `homewizard_local` | time since the meter **answered** | **yes** — every HTTP response is an arrival |
+| `ha_dsmr` | time since one of several entities changed | no — statistically usually fine, which is a masked bug, not an absent one |
+| `ha_signed` | time since the one entity changed | **no** — see above |
+| `mqtt_p1` | time since a message arrived | arrivals yes, but a stuck bridge republishing keeps arriving |
 
 ### Control
 

goodwe_controller/app/main.py

@@ -97,6 +97,10 @@ 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)))
@@ -365,10 +369,22 @@ 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:
@@ -540,6 +556,7 @@ 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()

goodwe_controller/app/p1.py

@@ -4,13 +4,22 @@ 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 meter exposes two UNSIGNED
-     registers - consumption and injection - never one signed figure. Net power
-     is `import_w - export_w`, positive = import, and that subtraction is done
+  1. The IMPORT/EXPORT DERIVATION. A Belgian P1 read over DSMR exposes two
+     UNSIGNED registers - consumption and injection. Net power 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).
@@ -47,6 +56,8 @@ _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
 
@@ -78,7 +89,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                    # SOURCE_HA | SOURCE_MQTT
+    source: str                    # one of the SOURCE_* constants above
     import_w: float                # unsigned magnitude, as the meter reports it
     export_w: float                # unsigned magnitude
     net_w: float                   # import_w - export_w  (+ import, - export)
@@ -129,6 +140,29 @@ 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,
@@ -668,6 +702,333 @@ 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
 # --------------------------------------------------------------------------- #
@@ -699,12 +1060,84 @@ 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 %s or %s - P1 ingestion disabled",
-                   source, SOURCE_HA, SOURCE_MQTT)
+        _LOG.error("meter_source %r is not one of %s - P1 ingestion disabled",
+                   source, ", ".join((SOURCE_HA, SOURCE_MQTT, SOURCE_HA_SIGNED,
+                                      SOURCE_HOMEWIZARD)))
     return None

goodwe_controller/config.yaml

@@ -1,5 +1,5 @@
 name: GoodWe RS485 Controller
-version: "0.2.1"
+version: "0.3.0"
 slug: goodwe_controller
 description: >-
   Drives a GoodWe ES/BP battery inverter over RS485 by emulating its smart
@@ -42,11 +42,24 @@ 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: ""
@@ -55,6 +68,15 @@ 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
@@ -98,20 +120,25 @@ schema:
   batt_invert: bool
   setpoint_entity: str
 
-  meter_source: list(off|ha_dsmr|mqtt_p1)
+  meter_source: list(off|ha_dsmr|mqtt_p1|ha_signed|homewizard_local)
   # ⚠️ 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)

goodwe_controller/test_p1.py

@@ -11,6 +11,7 @@ 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
@@ -18,8 +19,11 @@ 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, QuarterAverager, SOURCE_HA, SOURCE_MQTT,
-    make_sample, parse_mqtt_payload,
+    P1Error, P1Ingest, HaDsmrSource, HaSignedSource, HomeWizardLocalSource,
+    QuarterAverager,
+    SOURCE_HA, SOURCE_HA_SIGNED, SOURCE_HOMEWIZARD, SOURCE_MQTT,
+    build_source, is_enabled, make_sample, parse_homewizard, parse_mqtt_payload,
+    split_signed,
 )
 
 fails = []
@@ -56,6 +60,25 @@ 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,
@@ -548,6 +571,726 @@ 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
@@ -556,7 +1299,6 @@ 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)
@@ -564,6 +1306,8 @@ 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"]
@@ -636,6 +1380,68 @@ 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)}")