16 Commits
Author SHA1 Message Date
glenn schrooyenandClaude Opus 5 ad9c5772a4 TEL-04: a meter source for the meter this house actually has
TEL-01 shipped ha_dsmr and mqtt_p1. Neither fits the installed meter - a
HomeWizard P1 publishing one signed figure, where ha_dsmr requires two unsigned
registers and refuses negatives. So sensor.p1_sample_age_s could not be produced
here at all.

ha_signed is a subclass of HaDsmrSource overriding only _wanted() and build(),
so ingest timestamping, staleness, the clock-recomputed age, plausibility bounds
and unavailable-is-never-zero are inherited by construction rather than copied.
A reviewer traced every inherited member and confirmed nothing in the base
assumes two entities.

122 -> 179 checks. Sign convention asserted against real captures in
sim/scenarios/: -5710 W at 13:46 local under full sun, +775 W at midnight,
verified to the timestamp by two people independently.

WHAT THIS DOES NOT DO, documented in DOCS.md, the docstring and the CHANGELOG
rather than discovered later: the age it publishes measures time since the VALUE
CHANGED, not since the meter reported. Home Assistant exposes no arrival signal
for a repeated reading - proven on the rig against a frozen meter (no
state_changed, last_reported advancing on neither serialiser, state_reported
rejected outright) and confirmed independently against the live house, where ten
repeated values all left last_reported frozen. So this entity must NOT yet be
thresholded by the firmware watchdog. ha_dsmr has the same blind spot and
escapes only statistically, because a DSMR telegram moves several entities.

The fix is to read the meter's own API, where every response is an arrival.
That is TEL-05, and it is FW-01's real gate.

Two equivalent mutants are known and recorded: the per-phase split, documented
at the site, and the incomplete-phase-set guard, which is cosmetic - both paths
return False, one via an extra log line.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-08-25 18:00:42 +02:00
glenn schrooyenandClaude Opus 5 1b343da8e3 DOCS: say plainly that ha_dsmr and mqtt_p1 have never seen real hardware
No meter in this installation uses either transport. Both were written to
specs.md 5.2's assumption that a Belgian P1 exposes two unsigned registers,
which the meter actually fitted here does not - it is the HomeWizard P1 that
ha_signed reads. Their only coverage is test_p1.py and an end-to-end test
against a fake Home Assistant.

Deliberately not called "experimental". That word says the design is
unfinished, which is not the defect and is vaguer than the truth; these are
complete and reviewed, they have simply never had a real telegram through
them. The failure this note is guarding against is a future session debugging
a meter problem, treating those two paths as proven, and looking elsewhere.

Placed where the mode is chosen rather than in a footnote, because the choice
is the decision it should inform.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Du77usMj8XNKNFZGmUiWDa
2026-08-25 17:35:11 +02:00
glenn schrooyenandClaude Opus 5 e663e10245 TEL-04 review: unshadow the helper, validate config at startup, and record
what the rig proved about the age sensor

Review findings 1, 3, 5 and 6. Finding 2 is deliberately untouched - it is
its own ticket.

3. `built` was rebound at test_p1.py:816 by `built = build_source(...)`,
   silently disarming the build() wrapper for anything appended below it.
   Renamed to `sel`. Reproduced the reviewer's failure before fixing:
   appending a check that calls built() after that line gives
   `TypeError: 'HaSignedSource' object is not callable` and aborts at 163 of
   180; with the rename the same probe reaches 180 and passes.

5. DOCS.md now states the "length must equal meter_phases" constraint that
   config.yaml already carried, plus what leaving the list empty actually
   costs: on the surveyed reading the phases carry 2769 W of import while the
   connection nets 187 W, so the tariff quantity is understated ~15x.

6. build_source now checks the ha_signed wiring once at startup instead of
   once per telegram: a blank p1_net_entity, or a phase list whose length
   disagrees with meter_phases, logs an error and disables ingestion. Both
   otherwise fail in the single way indistinguishable from a healthy source
   nobody has fed yet - no samples, a climbing age, the watchdog holding the
   battery at 0 W, and nothing in the log.

1. THE AGE SENSOR. Measured on the ENV-01 rig against the real HomeWizard
   integration, meter frozen via hwsim's `?fault=freeze` seam (cleared in a
   finally:, rig verified restored):

     - websocket state_changed for the meter over 70 s : 0
     - last_reported advanced (REST serialiser)        : no
     - last_reported advanced (websocket serialiser)   : no
     - subscribe_events(state_reported)                : rejected,
       "Event filter is required for event state_reported"

   So Home Assistant exposes NO arrival signal for a repeated reading, and
   the proposed fix - stamp from last_reported via subscribe_entities - is
   not available. subscribe_entities listens only to EVENT_STATE_CHANGED, and
   as_compressed_state carries no last_reported at all.

   The age is therefore "time since the value changed", which on ha_dsmr is
   mostly harmless (a telegram moves several entities) and on ha_signed is
   not: one entity means a healthy meter under a flat load is
   indistinguishable from a dead one. Recorded loudly in DOCS.md, in the
   HaSignedSource docstring and in the CHANGELOG, with the measured 42.2 s
   and 97.0 s gaps from our own capture.

   meter_max_age_s is deliberately NOT widened. The two conditions produce an
   identical signal, so a larger number does not separate them - it only
   chooses which of the two errors you get, and it would disarm the watchdog
   for a genuinely dead meter as well. The honest fix is an arrival stamp the
   meter itself provides.

test_p1.py: 174 -> 179 checks, all green. Other three suites unchanged.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Du77usMj8XNKNFZGmUiWDa
2026-08-25 17:34:37 +02:00
glenn schrooyenandClaude Opus 5 632be44f6c TEL-04: make a missing build() guard fail legibly instead of aborting
Review point from the non-vacuity run. Deleting build()'s "is the net entity
cached at all" guard makes build() raise KeyError rather than return False.
That still failed the suite, but by aborting it with a traceback at whichever
check ran first - a red that costs the next person ten minutes deciding
whether the suite is broken or the code is.

New `built()` helper in test_p1.py wraps build() and turns an escaping
exception into a returned value, so the comparison against True/False fails
by name. Applied only to the ha_signed section; TEL-01's own checks are
untouched.

Mutation 4 before: 0 named checks red, aborted at check 123 of 174.
Mutation 4 after:  2 named checks red - "nothing cached yet builds nothing"
and "an unavailable entity does not build a sample" - all 174 reached.

Still 174 checks, all green, and the other nine mutations are unchanged.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Du77usMj8XNKNFZGmUiWDa
2026-08-25 17:01:33 +02:00
glenn schrooyenandClaude Opus 5 8b51a51e20 TEL-04: a third meter_source for a single signed entity
TEL-01 shipped ha_dsmr and mqtt_p1, and neither can read the meter that is
actually fitted here. The house has a HomeWizard P1 exposing ONE signed
entity, sensor.p1_meter_active_power (+ import, - export); ha_dsmr wants two
unsigned registers and refuses a negative one outright, which is every
exporting telegram. So sensor.p1_sample_age_s could not be produced at this
site, and FW-01's watchdog needs it - measured, not theoretical: the house P1
went 51.1 s and 36.2 s without a state change overnight, both past
meter_max_age_s 30, so without the age sensor the watchdog would false-trip
the battery to 0 W.

Adds meter_source: ha_signed, reading p1_net_entity (and optionally
p1_phase_net_entities in L1..L3 order for the capacity-tariff peak). The
derivation is split_signed(), sitting next to make_sample's subtraction for
the same reason it 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 exactly what TEL-01 removed.

The transport is a subclass of HaDsmrSource overriding only _wanted() and
build(), so every rule TEL-01 established is inherited rather than
re-implemented: ingest timestamping, meter_max_age_s, the clock-recomputed
sensor.p1_sample_age_s republished ~1 Hz, the plausibility ceiling, the
"prime the cache from get_states but never build a sample out of it" rule,
"a reconnect emits nothing", and unavailable/unknown treated as a MISSING
reading and never as 0 W.

Defaults to off. An existing install is unaffected until it opts in.

test_p1.py: 122 -> 174 checks. Includes an end-to-end run of the new
transport against a fake Home Assistant websocket, and the sign convention
asserted against real captured readings from
sim/scenarios/ha-p1_meter_active_power-2026-08-{20,23}.json (-5710 W at
13:46 local under full sun is export; +775 W at midnight is import).

Non-vacuity: ten mutations of the new rules, each applied alone and reverted
byte-identical. Nine turn the suite red. The tenth - splitting the per-phase
signed values rather than passing them through - is an equivalent mutant,
because make_sample subtracts the two lists again and does not sign-check
per-phase figures. That is recorded in a ponytail: comment at the site rather
than left for the next reviewer to rediscover.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Du77usMj8XNKNFZGmUiWDa
2026-08-25 16:44:22 +02:00
glenn schrooyenandClaude Opus 5 c24bc0a011 Pin LF, because the deployment target is a Linux container
core.autocrlf=true gave the checkout CRLF .py and .yaml. run.sh happened to be
LF, which is the only reason a plain copy would not have produced a "bad
interpreter" failure on the add-on's entrypoint.

0.3.0 was deployed by extracting from git with autocrlf forced off and verified
byte-identical to the blobs before copying. This makes that the default rather
than something the deployer has to remember.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-08-25 16:01:44 +02:00
glenn schrooyenandClaude Opus 5 80402c978f DEPLOY-01: 0.3.0, so the update is installable at all
release/1.0 carried SAFETY-04 and TEL-01 but still declared version 0.2.1 -
identical to what is installed and running on the house. Home Assistant keys
add-on updates off the version string, so the update would never have been
offered.

Two files. No code, no option defaults. What reaches the house at these
defaults is SAFETY-04's control law alone, and it carries a 4,928-case
equivalence proof against the previous law at integrator_max_w: 0. TEL-01 is
inert until meter_source is turned on.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-08-25 15:52:47 +02:00
glenn schrooyenandClaude Opus 5 6c980e87b0 DEPLOY-01: bump version to 0.3.0, changelog for SAFETY-04 and TEL-01
Fixes the version collision noticed while planning DEPLOY-01: release/1.0
still carried version 0.2.1, identical to what is already running on the
live system, so Home Assistant would not have offered the update at all.

- config.yaml: version 0.2.1 -> 0.3.0 (minor: TEL-01 adds a feature,
  SAFETY-04 changes the control law's internals)
- CHANGELOG.md: 0.3.0 entry for SAFETY-04 and TEL-01, in the existing voice

No code under app/ touched, no option defaults changed. Verified:
meter_source: off, integrator_max_w: 0, target_grid_w: -10 all unchanged.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Du77usMj8XNKNFZGmUiWDa
2026-08-25 10:33:17 +02:00
glenn schrooyenandClaude Opus 5 505a847d85 TEL-01: P1 ingestion, with the derivation and the sample age the EMS owns
Evidence at sign-off: 122 checks in test_p1.py, 14 mutations all red with the
tree restored byte-identical, and an end-to-end run of the HA transport against
a fake Home Assistant websocket server with a real auth handshake. Age
semantics verified live rather than from fixtures - a real 2.2 s sleep with no
telegram arriving, age climbing 2.2004 s.

The deliverable that matters beyond this ticket is sensor.p1_sample_age_s:
recomputed against a monotonic clock and republished ~1 Hz rather than stamped
per telegram, so a meter frozen at a constant value - which pushes no state
change and therefore emits nothing - still shows an age that climbs. SAFETY-01's
firmware subscribes to it and trips on has_state() && state >= max_age_s.

Two blockers on the way, both of which would have shipped. With meter_source
off - the default, chosen for zero regression - the age was published anyway
and climbed without bound, which would have crossed max_age_s within half a
minute and pinned every installed inverter at 0 W. And a reconnect emitted a
synthetic sample that reset the age, hiding an outage from the watchdog that
exists to catch it, contradicting the module's own docstring while a test
asserted the violation.

Not verifiable without hardware, and not claimed: real DSMR entity ids and
units, whether a real P1 MQTT bridge matches the documented strict schema, the
0.35 s debounce against real telegram timing, and MQTT reconnect against a real
broker.

Known limits, both documented and filed as SAFETY-12: mqtt_p1 cannot detect a
frozen bridge that keeps republishing, and a value-frozen meter stops the
control loop cycling at all - the latter pre-existing and affecting the legacy
meter_entity path today.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-08-24 23:17:34 +02:00
glenn schrooyenandClaude Opus 5 f498d5fa54 SAFETY-04: bound the integrator independently of the output
Evidence at sign-off: 55 checks in test_control.py (24 on release/1.0),
independently reproduced. 108,031 failsafe-release combinations swept across
both grid directions, 2,500 randomised carried-i_w trajectories at 200 ticks,
300 repeated-meter-value stall scenarios - zero anomalies. The historical
runaway regression uses the real 10.3 numbers and goes red when the protection
is removed.

Two rejections on the way. The first cut deadlocked the loop at small |i_w| and
let the accumulator run 50% past the rail, which ADDED windup this codebase
never had - the accumulator used to be the post-clamp command, so it could not
exceed max_w by construction. The second deadlocked at i_w == 0.0 in the export
direction, found by sweeping the boundary after three reviewers had each
covered the same half of it.

The finding worth keeping: deleting the integrator freeze outright failed 0 of
55 checks, because the integrator bound truncated to exactly the value the
fixture asserted. A neighbouring mechanism was standing in for the one under
test. The same pattern turned up again in the output clamp. test_control.py now
carries the audit table and its invariant - every mechanism in compute() must
be noticed by at least two checks when deleted.

Deliberately not met as literally written: the detector counts cycles, not the
10 s the AC specifies. compute() is clockless and cycle() runs only on a
changed meter reading, so there is no wall-clock window at all. Documented in
the code and in DOCS.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-08-24 23:17:18 +02:00
glenn schrooyenandClaude Opus 5 4bd659c499 TEL-01 review fixes: stop the age sensor tripping installs that have no meter
T-1, and it was a fleet-wide trip to zero. publish() emitted p1_age
unconditionally, and published_age_s counts from P1Ingest.__init__ when no
sample has ever arrived. With meter_source defaulting to off, every existing
install would have published sensor.p1_sample_age_s climbing without bound; the
ESP32 does `has_state() && state >= max_age_s` and forces the layer-1 failsafe,
so each of them would have pinned its inverter at 0 W within 30 s. Exactly the
opposite of the zero-regression the off default was for. The key is now omitted
from the payload AND from MQTT discovery when P1 is off, so the entity does not
exist at all - which is the status quo, and what has_state() is testing for.
The predicate is one function, is_enabled(), because the grid reading, the task
start and the discovery announcement have to agree or this comes back.

T-2, connect no longer manufactures a sample. get_states returns whatever HA
currently holds, which after a Core restart is a RestoreEntity value of unknown
age; stamping it with ingest_ts=now reset the age and reported a fresh meter
that could have been dead for an hour. run()'s own docstring already said a
reconnect must emit nothing - the code disagreed with it, and a test asserted
the violation. The cache is still primed, so the first real state_changed
builds a complete sample; the age just stays honest until one arrives.

T-3, gaps are no longer filled with the last held value. The averager held a
sample forward across any interval, so a meter dying at 5 kW and returning ten
minutes later credited 5 kW x 600 s to the capacity-tariff accumulator - a
fabricated peak on a permanent record. The hold is capped at max_age_s: past
that the stretch is walked so block boundaries still land correctly, but
nothing accumulates and elapsed does not grow, which is what finally makes the
comment about a gap dragging the billed average down true. Same threshold for
control and billing: a reading too old to steer by is too old to bill by.

T-4, the out-of-order/duplicate guard is covered. It was untested, and the
reason is worth recording: the obvious assertion passes without the guard,
because the negative interval is separately refused by the covered > 0 test.
What the guard prevents is the timestamp REWIND, which only shows up one sample
later as a re-integrated window. The test now goes one sample later.

T-6, DOCS was wrong about latency. meter_max_age_s and stale_input_s stack, so
meter death to 0 W is 45 s and not 30. Documented as a table with both clocks.

Also documented the T-5 asymmetry rather than papering over it: the age
measures arrival, not change, so a stuck MQTT bridge republishing its last
telegram still looks fresh. Correct on ha_dsmr, not detectable on mqtt_p1
without a change-detector. Written up as a known limit.

Writing the T-1 test caught a second defect in the test itself: it recorded
only MQTT topics, and object_id lives in the payload, so "the age sensor is not
announced" had been passing for the wrong reason.

test_p1.py: 99 -> 122 checks. 14 mutations run, all 14 red, files restored
byte-identical - including one per fix above.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Du77usMj8XNKNFZGmUiWDa
2026-08-24 23:16:31 +02:00
glenn schrooyenandClaude Opus 5 f47f1f0129 TEL-01: P1 ingestion, with the derivation and the age the EMS owns
A Belgian P1 meter publishes two UNSIGNED registers, not one signed figure.
Until now the add-on asked the installer to bridge that gap with a template
sensor, which put the sign convention of the whole control loop in a text box.
This moves it into the EMS: net = import - export, derived once, in one place,
with a test that fails if anyone inverts it.

Two transports behind one contract, chosen by `meter_source`: the HA WebSocket
subscribing to the DSMR integration's entities, and MQTT on a configurable
topic. Everything downstream reads P1Ingest, so switching is a config edit.
`meter_source: off` is the default and keeps the existing meter_entity path,
so no installed system changes until it opts in.

The other half is the timestamp. Every accepted sample is stamped at ingest
with a monotonic clock, `meter_max_age_s` is applied to it, and the age is
published as sensor.p1_sample_age_s for the ESP32's stale-input watchdog. That
entity is recomputed against the clock every second rather than only when a
telegram lands, because HA pushes state only on change: a meter frozen at a
constant reading emits nothing and looks, to anything watching the value,
exactly like a meter that has died. The age tells them apart.

Deliberately absent: any fallback to an inverter-side power figure. The
inverter's own AC power correlates 0.998 with battery power and 0.09 with the
real meter, so failing over to it means regulating against your own output.
A gap stays a gap - a reconnect emits no synthetic sample, and a rejected
telegram never resolves to 0 W or refreshes the timestamp.

Quarter-hour averages are time-weighted over clock-aligned blocks rather than
a mean of samples, so a cadence change cannot bias the capacity-tariff figure,
and only offtake is accumulated so a quarter of pure export averages to 0 kW.
Per-phase import is kept separately: on an unbalanced three-phase load the
phase sum and the connection net are different numbers, and only one of them
is billed.

test_p1.py: 99 checks, runnable with a bare interpreter and no meter. Includes
an end-to-end run of the HA transport against a fake Home Assistant websocket.

Stacked on SAFETY-04; nothing here touches control.py.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Du77usMj8XNKNFZGmUiWDa
2026-08-24 23:16:31 +02:00
glenn schrooyenandClaude Opus 5 680461c9bf SAFETY-04: record the coverage audit as a comment, with its invariant
The audit found a dead mechanism twice, both times a clamp standing in for the
mechanism under test, so the technique has to survive this ticket. Not as a
script: the only cheap way to automate it is to key on source lines, that goes
stale silently, and a green audit that has quietly stopped testing anything is
this ticket's own failure mode one level up. Automating it properly would mean
decomposing compute() to make its statements separately addressable, which is a
refactor of the most safety-critical function in the repo for the benefit of
test tooling.

So it goes in as a comment block next to the checks it describes, carrying the
commit it was measured at, the thirteen figures, and the invariant with the
teeth in it: every mechanism must be noticed by at least two checks when it is
deleted, and adding a mechanism means re-running the audit. A comment cannot go
stale-green, because it never claims to be running.

Also recorded: fixtures must sit clear of every rail they are not testing,
which is the rule both misses violated; and the `python -B` / clear-pycache
discipline, with the reason (CPython invalidates on source mtime-in-seconds
plus size, so a same-second same-size rewrite reuses stale bytecode) and the
reason it casts no doubt on the figures (the error is one-directional, so every
number is a lower bound).

Figures are the lead's independent reproduction. I re-measured the one that
differed: the detector is 11 for `saturated_now = False` and 10 for the weaker
`frozen = False` form, so the table names the form.

test_control.py stays at 55 checks, all passing.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Du77usMj8XNKNFZGmUiWDa
2026-08-24 23:07:58 +02:00
glenn schrooyenandClaude Opus 5 389d9ecd6d SAFETY-04: stop the clamps standing in for the mechanisms under test
AC 3 - "integration freezes while saturated" - had no non-vacuous test.
Deleting the integrator freeze outright failed 0 of 55 checks: the two checks
that name it used a fixture at max_w 2000 with the integrator bound following
it, so a wound value was truncated back to exactly 2000 and the assertion
passed on the clamp instead. Fixture lifted to max_w 5000, clear of every rail.
Deleting the freeze now fails 2.

Then swept the whole function for the same pattern, one mechanism at a time:
delete it, count which checks notice. It found a second instance - the OUTPUT
clamp. `clamped to max_w` and `clamped to -max_w` were both satisfied by the
integrator bound truncating first, so removing the output clamp failed only the
reason-string check. Those two fixtures now set integrator_max_w above max_w so
the mechanism they name is the binding one; the output clamp goes from 1 failure
to 3.

Every mechanism in compute() is now caught by a check that names it: freeze 2,
integrator clamp 6, bound-follows-max_w 4, output clamp 3, slew 4, output freeze
2, deadband 5, quantise 2, detector 10, duration 2, counter reset 6, grid bias 3,
None-seeding 6. No mechanism at zero.

Method note: the audit disables bytecode caching. Rewriting control.py inside
one second leaves a stale app/__pycache__ entry and silently under-reports -
it under-reported one mutation as 2 failures where the true figure is 6.

test_control.py stays at 55 checks, all passing.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Du77usMj8XNKNFZGmUiWDa
2026-08-24 22:59:29 +02:00
glenn schrooyenandClaude Opus 5 53d301b920 SAFETY-04: exactly zero is its own case in the freeze tie-break
`min(moved, i_w) if i_w > 0 else max(moved, i_w)` files i_w == 0.0 under
rising-only, so the first push toward charging from exactly zero was blocked
permanently - the S-1 deadlock again, mirrored in sign. main.py resets i_w to
exactly 0.0 on every stop and every reseed, so it is a normal state.

Zero is now handled explicitly and both directions are allowed: nothing is
wound, so "may not wind further" has no referent, and a first step from zero is
bounded by the gain, the output clamp and the slew limit like any other.

Measured before the fix, at i_w == 0.0 and frozen: 12 800 of 25 920 ticks held
the integrator and 8 304 of those changed the emitted command, worst case
abandoning a 2 kW charge into a 4 kW export. Note this is NOT the same as the
reported symptom: at prev_w == 0 the command holds at 0 W either way, because
the output freeze forbids starting a charge while saturated, and that rule is
release/1.0's and unchanged. There is now a test asserting it deliberately.

Tests. The durable part is a property rather than more points: over 13 041
frozen states the integrator may be held ONLY by a correction pushing it
further from zero on the side it already sits, and any other hold fails. Both
signs at exactly 0.0. Mirrors added everywhere the suite tested one direction
of two - freeze wind/unwind while charging, i_w=-100, the export-direction
runaway, the negative clamp and slew.

DOCS: the cycles-vs-seconds deviation is now written down as a deviation - the
"> 10 s" criterion is not met as literally written, a cycle is one CHANGED
meter reading, and there is no guaranteed wall-clock window.

test_control.py: 43 -> 55 checks, all passing.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Du77usMj8XNKNFZGmUiWDa
2026-08-24 22:49:01 +02:00
glenn schrooyenandClaude Opus 5 7123aa00a4 SAFETY-04: revive the clamp reason, and compare reasons in the sweep
`want = i_w` after the integrator bound, so at the default limit == max_w the
output clamp can never fire and `reason == "clamped"` had become unreachable.
Observability only today - nothing gates on the string - but SAFETY-03 exists
to alarm on exactly that engagement, so its hook was dead before it was built.

The integrator bound now reports "i-clamped", and that is the signal SAFETY-03
must watch: it is the one that fires on a default install. "clamped" stays
reachable for a configuration that lets the integrator run above the rail,
where both fire and the output clamp - which describes the value actually
emitted - is the one reported. Two names because the two events want different
alarms: the loop winding, versus a command that came out over the rating.

The real fix is the second half. The equivalence sweep compared
(target_w, sat_count), which is how a dead reason survived 3024 cases. It now
compares (target_w, sat_count, frozen, reason) and it catches this defect:
dropping the emit turns it red. Deliberate rename aliased explicitly, so any
OTHER reason divergence still fails.

Result of adding reason to the tuple: 105 of 3024 cases differ, and every one
of them is the i-clamped/clamped rename. Zero value divergences, `frozen`
included. Nothing else surfaced.

test_control.py: 41 -> 43 checks, all passing.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Du77usMj8XNKNFZGmUiWDa
2026-08-24 22:29:10 +02:00
10 changed files with 2419 additions and 35 deletions
+10
View File
@@ -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
+70
View File
@@ -1,5 +1,75 @@
# Changelog
## Unreleased
**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
+169 -2
View File
@@ -48,6 +48,150 @@ Use the ESP32's readings rather than the inverter's cloud or dongle sensors:
those serve cached values, and a stale reading here ends the maintenance charge
phase having charged nothing.
### P1 meter ingestion
`meter_entity` above expects one signed sensor, which usually means a template
someone wrote by hand. Setting `meter_source` moves the whole derivation into
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.
> ⚠️ **`ha_dsmr` and `mqtt_p1` have never processed a telegram from real
> hardware.** No meter in this installation uses either one. Both were written
> to the assumption in `specs.md` §5.2 that a Belgian P1 exposes two unsigned
> registers, and the meter actually fitted here does not — it is the HomeWizard
> P1 that `ha_signed` reads. They are covered by the unit checks in `test_p1.py`
> and by an end-to-end test against a fake Home Assistant, and nothing more.
>
> This is recorded because the realistic way it bites is someone debugging a
> meter problem months from now treating those two paths as proven and looking
> for the fault elsewhere. If you are the first person to point one at a real
> meter, expect to find something, and please update this note when you do.
| option | default | meaning |
|---|---|---|
| `meter_source` | `off` | `off` keeps `meter_entity`. `ha_dsmr` subscribes to the DSMR integration over the HA WebSocket; `mqtt_p1` reads a topic; `ha_signed` subscribes to one signed entity over the HA WebSocket |
| `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_mqtt_topic` | | `mqtt_p1` only |
| `p1_import_entity` | | The **unsigned** consumption sensor. Do not point this at a signed template |
| `p1_export_entity` | | The **unsigned** injection sensor |
| `p1_phase_import_entities` | `[]` | L1..L3, in order. Needed for the capacity-tariff peak on a three-phase connection |
| `p1_phase_export_entities` | `[]` | L1..L3, in order |
| `p1_net_entity` | | `ha_signed` only. The **signed** net-power sensor: `+` import, `-` export |
| `p1_phase_net_entities` | `[]` | `ha_signed` only. L1..L3, in order, each signed the same way. Needed for the capacity-tariff peak on a three-phase connection. **The list length must equal `meter_phases`** |
Both per-phase lists are checked against `meter_phases` **once at startup**: a
list of the wrong length disables P1 ingestion with an error in the log, rather
than letting every telegram fail its phase-count check one at a time. Leaving
the list empty is fine and is not an error — you simply get no per-phase
figures, and therefore no capacity-tariff peak. On a three-phase connection
that is a much bigger omission than it looks: on a surveyed reading here the
phases carried 2769 W of import while the connection netted 187 W, so the
billed quantity is understated roughly fifteenfold if the phases are missing.
#### How long a dead meter takes to reach 0 W
`meter_max_age_s` and `stale_input_s` **stack**. They are two different clocks
and neither one is the whole answer:
| step | option | default |
|---|---|---|
| telegrams stop, P1 sample goes stale, grid power starts reading *missing* | `meter_max_age_s` | 30 s |
| inputs have been missing long enough for the loop to command 0 W | `stale_input_s` | 15 s |
| **total, meter death → 0 W commanded by this add-on** | | **45 s** |
So in P1 mode `stale_input_s` is *not* "how long inputs may be missing before
commanding 0 W" measured from the meter dying — it is measured from the moment
the P1 sample already went stale. Size the pair together: the ESP32's own
watchdog commands 0 W after ~30 s of silence from this add-on regardless, and
that layer is unaffected by either option.
There is **no fallback to an inverter-side power figure**, deliberately. The
inverter's own AC power tracks its battery almost perfectly and the real meter
hardly at all, so a controller that failed over to it would be regulating
against its own output while looking healthy.
The `mqtt_p1` payload is one JSON object per telegram, and the schema is strict —
a key it does not recognise is a telegram from something other than what was
tested, and guessing a key here means guessing a kilowatt:
```json
{"import_w": 1234.0,
"export_w": 0.0,
"phases": [{"import_w": 500, "export_w": 0},
{"import_w": 400, "export_w": 0},
{"import_w": 334, "export_w": 0}],
"timestamp": "2026-08-24T18:00:05+02:00"}
```
`phases` and `timestamp` are optional; `timestamp` must carry a UTC offset. Where
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.
#### `sensor.p1_sample_age_s`
Published over MQTT discovery whenever a broker is available: **seconds since the
newest accepted telegram**, refreshed every second rather than only when a
telegram lands. The ESP32's stale-input watchdog subscribes to this exact entity
id, so do not rename it.
The reason it is recomputed against the clock is that Home Assistant only pushes
a state when the state *changes*. A meter sitting at a genuinely constant reading
emits nothing, which is indistinguishable — to anything watching the value — from
a meter that has died. Watching the age instead separates the two: it climbs when
telegrams stop and resets when they arrive, whatever the reading says.
The entity is only created when `meter_source` is not `off`. With P1 ingestion
disabled there is nothing feeding it, and an age sensor climbing with no ingester
behind it would trip the firmware watchdog on a system that is working fine.
> **Known limit, `mqtt_p1`.** The age measures *arrival*. On the MQTT path a
> bridge that is stuck republishing its last telegram keeps arriving, so the age
> stays near zero and a frozen meter still looks fresh. Detecting *that* needs a
> change-detector rather than an arrival-detector, and it is not in this version.
> ⚠️ **Known limit, `ha_signed` — do not drive a watchdog off this age yet.**
> On the HA WebSocket paths the age is stamped when a `state_changed` arrives,
> which means it measures *time since the value last changed*, not time since
> the meter last reported. Home Assistant offers nothing better: a repeated
> reading produces no `state_changed`, does **not** advance `last_reported` on
> either the REST or the WebSocket serialiser, and `state_reported` cannot be
> subscribed to over the WebSocket at all (`Event filter is required for event
> state_reported`). All three measured on the ENV-01 rig against the real
> HomeWizard integration with the meter frozen: 0 `state_changed` in 70 s and no
> timestamp movement anywhere.
>
> `ha_dsmr` mostly escapes this because a DSMR telegram updates several entities
> and something in the set almost always moves. **`ha_signed` has exactly one
> entity, so a healthy meter under a flat load is indistinguishable from a dead
> one.** This is not hypothetical: in our own captures
> (`sim/scenarios/ha-p1_meter_active_power-2026-08-20.json`) the real house meter
> went **42.2 s and 97.0 s** between changes, and 23 Aug peaks at 29.1 s — all
> past the default `meter_max_age_s` of 30.
>
> So `sensor.p1_sample_age_s` on `ha_signed` is safe to *read*, and it is
> correct whenever the value is moving, but it must not yet be thresholded by
> the ESP32 stale-input watchdog: a quiet house would trip the battery to 0 W.
> Raising `meter_max_age_s` is **not** the fix — the two conditions produce an
> identical signal, so a bigger number only chooses which of the two errors you
> get. The real fix is an arrival stamp the meter itself provides, i.e. reading
> the HomeWizard local API directly rather than through an HA entity.
### Control
| option | default | meaning |
@@ -59,12 +203,35 @@ phase having charged nothing.
| `target_grid_w` | -10 | What the meter should rest at. Negative = a slight export |
| `step_w` | 10 | Quantisation |
| `saturation_w` | 500 | Divergence that counts as "the inverter is at a limit" |
| `saturation_cycles` | 3 | How many consecutive cycles before freezing. **Do not set to 1** |
| `saturation_cycles` | 3 | How many consecutive cycles before freezing. A cycle is one *changed* meter reading, not a fixed period - see the note below. **Do not set to 1** |
| `integrator_max_w` | 0 | Bound on the loop's accumulator, and 0 means "same as `max_w`". Caps how much stale error can be waiting to unwind when the sign flips. **Do not raise it above `max_w`** - the output clamp already bounds what is commanded, so the only thing extra headroom buys is more cycles of wrong-direction power after every saturation event. Lowering it below `max_w` is the useful direction |
| `heartbeat_s` | 10 | Refresh interval; must stay well under the firmware watchdog |
| `stale_input_s` | 15 | How long inputs may be missing before commanding 0 W |
| `stale_input_s` | 15 | How long inputs may be missing before commanding 0 W. In P1 mode this clock starts only *after* `meter_max_age_s` has already expired — the two stack, see "How long a dead meter takes to reach 0 W" |
| `auto_start` | false | Start controlling on boot (only after commissioning) |
#### Saturation is counted in cycles, not seconds
The specification states the saturation window as **"> 10 s"**. This add-on counts
**cycles** instead, and that is a deliberate, accepted deviation rather than an
oversight - the acceptance criterion is not met as literally written.
A cycle here is one *changed* meter reading: the controller only runs the loop when the
meter value differs from the previous poll. At the reference P1's ~5 s update rate the
default of 3 cycles is usually around 15 s, but there is **no guaranteed wall-clock
window** - a meter that repeats the same value stalls the counter for as long as it
repeats.
Two reasons that is acceptable:
- the control law is a pure function with no clock, which is what makes it testable
without hardware, and a seconds-based window would have to live in the controller;
- a stalled counter is a detection-latency limit and not a runaway risk. The condition
that stalls it - an unchanging meter - stops the whole loop, so nothing accumulates
while it is stalled.
If a guaranteed window matters on your site, raise `saturation_cycles` for a fast meter,
and treat the figure as "N meter updates" rather than "N seconds".
#### Why `target_grid_w` is not zero
The deadband is a one-way ratchet: any resting point inside it holds until
+33 -3
View File
@@ -156,15 +156,45 @@ def compute(
# has been in service on real hardware. It is applied here as well
# because the requirement is that the INTEGRATOR stop accumulating, not
# only the command.
if not frozen:
#
# ⚠️ EXACTLY ZERO IS ITS OWN CASE, and it must be handled explicitly
# rather than falling into one of the two branches. "May not wind
# further in the direction it is already pushing" has no referent at
# zero: nothing is wound, and neither direction is "further". Writing
# this as `if i_w > 0 ... else ...` silently files zero under
# rising-only and permanently blocks the first push toward charging -
# the same deadlock as the shrink-only encoding above, mirrored in sign,
# and reachable because main.py resets i_w to exactly 0.0 on every stop
# and every reseed. Measured before the fix: 12 800 of 25 920 frozen
# ticks at i_w == 0.0 held the integrator, 8 304 of them changing the
# emitted command, worst case abandoning a 2 kW charge into a 4 kW
# export.
#
# Freezing at zero would also be pointless: the freeze exists to stop
# accumulation running away, and a first step from zero is bounded by
# the gain, the output clamp and the slew limit like any other.
if not frozen or i_w == 0.0:
i_w = moved
elif i_w > 0:
i_w = min(moved, i_w)
else:
i_w = min(moved, i_w) if i_w > 0 else max(moved, i_w)
i_w = max(moved, i_w)
# ⚠️ Applied EVERY cycle, frozen or not: the freeze is conditional, this
# bound is not. It is what makes the worst-case unwind time finite and
# knowable instead of a function of how long the error happened to stand.
i_w = max(-limit, min(limit, i_w))
bounded = max(-limit, min(limit, i_w))
if bounded != i_w:
# ⚠️ SAFETY-03 (alarm whenever the loop winds into a rail) must watch
# for THIS, not for "clamped" below. At the default limit == max_w the
# integrator bound is reached first and the command derived from it can
# then never exceed max_w, so "clamped" is unreachable on a default
# install - it survives only for a configuration that deliberately lets
# the integrator run above the rail. Two reasons rather than one
# because the two events want different alarms: "i-clamped" is the loop
# winding, "clamped" is a command that came out over the rating anyway.
reason = "i-clamped"
i_w = bounded
want = i_w
# ⚠️ Maintenance shaping (charge-only, cheap-window floor) used to live
+72 -13
View File
@@ -39,6 +39,7 @@ from .control import Tuning, compute, maintenance_charge_floor, peak_at_risk
from .hass import HomeAssistant
from .maintenance import IDLE, MaintConfig, Maintenance
from .mqtt import MqttPublisher
from .p1 import P1Ingest, build_source, is_enabled
from . import web
OPTIONS_PATH = "/data/options.json"
@@ -90,6 +91,13 @@ class Controller:
store,
)
# P1 ingestion (TEL-01). `meter_source: off` keeps the original
# single-entity meter_entity path, so an existing install is unchanged
# until it opts in.
self.p1 = P1Ingest(phases=int(opts.get("meter_phases", 1)),
max_age_s=float(opts.get("meter_max_age_s", 30)))
self.p1_enabled = is_enabled(opts)
# live state
self.auto = bool(store.data.get("auto", opts.get("auto_start", False)))
self.target = 0.0
@@ -121,8 +129,18 @@ class Controller:
# -- io ------------------------------------------------------------------
async def read_inputs(self) -> None:
o = self.o
self.grid = await self.hass.number(o.get("meter_entity", ""),
bool(o.get("meter_invert")))
if self.p1_enabled:
# ⚠️ P1 is the only authoritative measurement of what the utility
# sees (§5.1). When it is stale this is None, which falls into the
# existing "inputs missing -> command 0 W" path below. There is
# deliberately NO fallback to an inverter-side figure: the
# inverter's own AC power correlates 0.998 with battery power and
# 0.09 with the real meter, so a controller that failed over to it
# would be regulating against its own output.
self.grid = self.p1.net_w
else:
self.grid = await self.hass.number(o.get("meter_entity", ""),
bool(o.get("meter_invert")))
self.soc = await self.hass.number(o.get("soc_entity", ""))
self.batt = await self.hass.number(o.get("batt_entity", ""),
bool(o.get("batt_invert")))
@@ -306,14 +324,26 @@ class Controller:
await asyncio.sleep(1)
def publish(self) -> None:
self.mqtt.publish({
values = {
"setpoint": self.target,
"grid": self.grid,
"battery": self.batt,
"soc": self.soc,
"phase": self.maint.phase,
"status": "running" if self.auto else "stopped",
})
}
# ⚠️ ONLY when P1 ingestion is actually running. The ESP32's stale-input
# watchdog subscribes to sensor.p1_sample_age_s and forces the layer-1
# failsafe once it reaches max_age_s. With meter_source off there is no
# ingester feeding it, so published_age_s would be time-since-startup
# climbing without bound - i.e. every existing install would cross the
# threshold within 30 s and pin its inverter at 0 W forever. Publishing
# nothing leaves the entity non-existent, which is the status quo and
# what has_state() in the firmware is checking for.
if self.p1_enabled:
# Recomputed here, once a second, on purpose - see P1Ingest.
values["p1_age"] = round(self.p1.published_age_s, 1)
self.mqtt.publish(values)
async def shutdown(self) -> None:
"""Deterministic wind-down. Do not skip this."""
@@ -326,11 +356,27 @@ class Controller:
def checks(self) -> list:
o = self.o
out = []
for label, value, entity in (
("grid power", self.grid, o.get("meter_entity")),
("battery SoC", self.soc, o.get("soc_entity")),
("battery power", self.batt, o.get("batt_entity")),
):
if self.p1_enabled:
age = self.p1.published_age_s
if self.p1.stale:
out.append({"ok": False, "warn": False,
"text": f"P1 meter ({o.get('meter_source')}): no reading for "
f"{age:.0f} s (limit {self.p1.max_age_s:.0f} s)"
+ (f" - last error: {self.p1.last_error}"
if self.p1.last_error else "")})
else:
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"})
rows = [("battery SoC", self.soc, o.get("soc_entity")),
("battery power", self.batt, o.get("batt_entity"))]
if not self.p1_enabled:
# In P1 mode the check above replaces this one; leaving both in
# would report "no entity configured" for a meter_entity that is
# correctly unused, i.e. a permanent false NOT READY.
rows.insert(0, ("grid power", self.grid, o.get("meter_entity")))
for label, value, entity in rows:
if not entity:
out.append({"ok": False, "warn": False, "text": f"{label}: no entity configured"})
elif value is None:
@@ -457,6 +503,7 @@ async def amain() -> None:
# observability, and the battery does not care. Caught broadly and on
# purpose: this crashed the add-on once already (paho 1.x vs 2.x) and
# took the control loop down with it.
broker = None
try:
broker = await hass.mqtt_service()
pub = MqttPublisher(
@@ -464,6 +511,7 @@ async def amain() -> None:
broker.get("port", 1883) if broker else 1883,
broker.get("username") if broker else None,
broker.get("password") if broker else None,
omit=() if is_enabled(opts) else ("p1_age",),
)
except Exception as err: # noqa: BLE001
_LOG.warning("MQTT unavailable (%s) - continuing without status entities", err)
@@ -484,13 +532,24 @@ async def amain() -> None:
with contextlib.suppress(NotImplementedError):
loop.add_signal_handler(sig, stop.set)
task = asyncio.create_task(controller.run_control())
tasks = [asyncio.create_task(controller.run_control())]
# P1 ingestion runs as its own long-lived task. ⚠️ It must not be driven
# off the control loop: telegrams arrive every ~5 s and the loop would
# decimate them, so the 15-minute average - the capacity-tariff billing
# 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:
tasks.append(asyncio.create_task(p1_source.run()))
await stop.wait()
await controller.shutdown()
task.cancel()
with contextlib.suppress(asyncio.CancelledError):
await task
for task in tasks:
task.cancel()
for task in tasks:
with contextlib.suppress(asyncio.CancelledError):
await task
await runner.cleanup()
_LOG.info("stopped")
+16 -1
View File
@@ -45,6 +45,14 @@ SENSORS = [
("soc", "goodwe_battery_soc", "Battery SoC", "%", "battery", "measurement", None),
("phase", "goodwe_maintenance_phase", "Maintenance phase", None, None, None, "mdi:battery-sync"),
("status", "goodwe_controller_status", "Controller status", None, None, None, "mdi:heart-pulse"),
# ⚠️ This one deliberately breaks the goodwe_ prefix above: the entity id
# must be exactly `sensor.p1_sample_age_s`, because SAFETY-01's firmware
# watchdog subscribes to that literal id and the ENV-01 simulation rig
# asserts on it. Renaming it silently disarms a safety layer. It is seconds
# since the newest accepted P1 telegram, republished every second so that a
# meter frozen at a constant value still shows a climbing age - which is the
# false-trip that this entity exists to remove.
("p1_age", "p1_sample_age_s", "P1 sample age", "s", "duration", "measurement", None),
]
BASE = "goodwe_ctl"
@@ -52,7 +60,12 @@ AVAILABILITY = f"{BASE}/availability"
class MqttPublisher:
def __init__(self, host, port, username=None, password=None):
def __init__(self, host, port, username=None, password=None, omit=()):
# `omit` drops sensor keys from discovery entirely. ⚠️ Announcing a
# sensor that nothing will ever publish to is not harmless here:
# p1_sample_age_s is a watchdog input, and an entity that exists but is
# never fed is a worse signal than one that does not exist at all.
self.omit = set(omit)
self.enabled = mqtt is not None and bool(host)
self.client = None
if not self.enabled:
@@ -86,6 +99,8 @@ class MqttPublisher:
def _announce(self) -> None:
for key, object_id, name, unit, dev_class, state_class, icon in SENSORS:
if key in self.omit:
continue
cfg = {
"name": name,
"object_id": object_id,
+862
View File
@@ -0,0 +1,862 @@
"""P1 meter ingestion - the only authoritative measurement of real grid exchange.
Everything downstream trusts this module: the safety checks, the capacity-tariff
peak, the optimizer, the control loop's sign. So three things happen here and
nowhere else.
1. The IMPORT/EXPORT DERIVATION. A Belgian P1 read over DSMR exposes two
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).
3. VALIDATION. This is untrusted external data at the edge of a safety chain.
A malformed telegram must not become a plausible-looking number, and it
must never resolve to 0 W - a fabricated zero is indistinguishable from a
balanced house and defeats the very staleness trigger this module feeds.
⚠️ There is deliberately NO fallback to an inverter-side power figure. The
inverter's own AC power correlates 0.998 with battery power and 0.09 with the
real meter (§5.1) - regulating on it means regulating against your own output.
When the transport dies the correct behaviour is a gap: no sample, a growing
age, and the existing "inputs missing -> command 0 W" path in main.py.
"""
import asyncio
import json
import logging
import math
import os
import time
from dataclasses import dataclass
from datetime import datetime, timezone
import aiohttp
try:
import paho.mqtt.client as mqtt
except ImportError: # pragma: no cover - container always has it
mqtt = None
_LOG = logging.getLogger("goodwe.p1")
SOURCE_HA = "ha_dsmr"
SOURCE_MQTT = "mqtt_p1"
SOURCE_HA_SIGNED = "ha_signed"
QUARTER_S = 900
# ⚠️ Plausibility ceiling, not a clamp - anything above it is rejected as an
# anomaly rather than averaged in. Chosen to sit above the largest Belgian
# residential connection (3x63 A ~ 43 kW) and BELOW 65535: §20 open question 5
# records an HA sensor reporting 64954 for -582 W, i.e. an unsigned 16-bit
# register decoded without its sign. That corruption reads as a perfectly
# plausible 65 kW if you only bound it at "some big number".
PLAUSIBLE_MAX_W = 50_000.0
# --------------------------------------------------------------------------- #
# the sample
# --------------------------------------------------------------------------- #
class P1Error(ValueError):
"""A telegram that must be rejected rather than believed."""
@dataclass(frozen=True)
class P1Sample:
"""One telegram, validated, derived and stamped.
Frozen on purpose: this object is handed to readers on other tasks (and,
for the MQTT transport, produced on paho's network thread). Immutability is
what makes "read the latest sample" safe without a lock.
"""
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_HA_SIGNED
import_w: float # unsigned magnitude, as the meter reports it
export_w: float # unsigned magnitude
net_w: float # import_w - export_w (+ import, - export)
per_phase_w: tuple[float, ...] | None # signed net, len == phases
per_phase_import_w: tuple[float, ...] | None # offtake only, for the tariff
def age_s(self, now_mono: float | None = None,
now_utc: datetime | None = None) -> float:
"""Seconds since this sample was ingested, never negative.
⚠️ Measured with time.monotonic(), not the wall clock. An NTP step on a
Pi that just booted moves the wall clock by minutes; using it here would
either fake a stale meter or, worse, hide a real one.
Where the telegram carries its own timestamp we take the WORSE of the
two ages. That is what stops an MQTT retained message - replayed on
reconnect with a fresh receive time - from presenting a ten-minute-old
reading as brand new.
"""
now_mono = time.monotonic() if now_mono is None else now_mono
age = max(0.0, now_mono - self.ingest_mono)
if self.telegram_ts is not None:
now_utc = datetime.now(timezone.utc) if now_utc is None else now_utc
age = max(age, (now_utc - self.telegram_ts).total_seconds())
return max(0.0, age)
def _watts(value, what: str) -> float:
"""Parse one power figure, or raise. Never returns a substituted default.
⚠️ Strings are refused even when float() would happily take them. A JSON
telegram carrying "1200" where a number belongs is a payload from a source
that is not the one we validated against, and the next surprise it has may
not be a benign one. Transports that legitimately deal in text (HA entity
states are always strings) convert before they get here, so this stays the
strict edge for structured payloads.
"""
if isinstance(value, (bool, str, bytes)) or value is None:
raise P1Error(f"{what}: not a number ({value!r})")
try:
out = float(value)
except (TypeError, ValueError):
raise P1Error(f"{what}: not a number ({value!r})") from None
if not math.isfinite(out):
raise P1Error(f"{what}: not finite ({value!r})")
if abs(out) > PLAUSIBLE_MAX_W:
raise P1Error(f"{what}: {out:g} W is outside plausible meter range")
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,
ingest_ts: datetime | None = None,
ingest_mono: float | None = None) -> P1Sample:
"""Validate, derive net power, stamp. Raises P1Error on anything doubtful.
`import_w`/`export_w` are the two unsigned Belgian registers. Per-phase
figures are equally unsigned and equally split, so each phase gets the same
derivation.
"""
imp = _watts(import_w, "import")
exp = _watts(export_w, "export")
# ⚠️ Both registers are magnitudes. A negative one means the upstream
# already applied a sign we are about to apply again - reject it rather
# than silently double-signing the control loop.
if imp < 0 or exp < 0:
raise P1Error(f"unsigned registers cannot be negative (import={imp:g} export={exp:g})")
per_phase = per_phase_import = None
if phase_import_w is not None or phase_export_w is not None:
pi = list(phase_import_w or [])
pe = list(phase_export_w or [0.0] * len(pi))
if len(pi) != phases or len(pe) != phases:
raise P1Error(
f"phase count mismatch: telegram has {len(pi)} import / {len(pe)} export "
f"phases, meter_phases is {phases}")
vals = [_watts(a, f"L{i + 1} import") - _watts(b, f"L{i + 1} export")
for i, (a, b) in enumerate(zip(pi, pe))]
per_phase = tuple(vals)
per_phase_import = tuple(max(v, 0.0) for v in vals)
if telegram_ts is not None and telegram_ts.tzinfo is None:
raise P1Error("telegram timestamp has no timezone")
return P1Sample(
ingest_ts=ingest_ts or datetime.now(timezone.utc),
ingest_mono=time.monotonic() if ingest_mono is None else ingest_mono,
telegram_ts=telegram_ts,
source=source,
import_w=imp,
export_w=exp,
net_w=imp - exp,
per_phase_w=per_phase,
per_phase_import_w=per_phase_import,
)
# --------------------------------------------------------------------------- #
# the 15-minute average
# --------------------------------------------------------------------------- #
@dataclass(frozen=True)
class QuarterBlock:
start: datetime # UTC, aligned to :00/:15/:30/:45
offtake_avg_w: float # billed figure: net offtake only
per_phase_offtake_avg_w: tuple[float, ...] | None
class QuarterAverager:
"""Time-weighted average of net offtake over clock-aligned 15-min blocks.
Samples arrive irregularly (~1-10 s), so a plain mean over samples would
weight a burst of fast telegrams the same as a slow one and produce a figure
that is not the billed quantity. Each sample's value is therefore HELD until
the next arrives and integrated over that interval: sum(value * dt) / dt.
⚠️ Only OFFTAKE is accumulated (§9.1) - the capacity tariff bills the highest
quarter-hour average offtake, and a quarter of pure export averages to 0 kW,
not to a negative one. The signed series stays available for control; this
accumulator is for the meter's bill.
⚠️ Blocks are found by flooring epoch seconds to 900. That IS clock-aligned
and DST-proof for Belgium, because every offset in that tz is a whole number
of hours, so a 900 s grid in UTC lands on :00/:15/:30/:45 local before and
after a transition - no tz database, no DST special case.
ponytail: the ceiling is a timezone with a sub-hour offset (India +05:30,
Nepal, Chatham). Those need real tz-aware boundary maths; upgrade path is to
compute the boundary with zoneinfo instead of the modulo, everything else
here is unchanged.
A sample that straddles a boundary is split at the boundary and its two
halves credited to the two blocks, never attributed wholly to either.
"""
def __init__(self, phases: int = 1, max_hold_s: float = 30.0):
self.phases = phases
# ⚠️ How long one sample may be held forward before the series is
# treated as a gap rather than a plateau. Without this the meter can die
# while importing 5 kW, come back ten minutes later, and the hold-forward
# credits 5 kW x 600 s to the capacity-tariff accumulator - a fabricated
# peak, on a permanent record, from data that was never measured. Set
# from meter_max_age_s: the point past which the reading is not trusted
# for control is the point past which it must not be billed either.
self.max_hold_s = float(max_hold_s)
self._block: int | None = None # epoch seconds of the block start
self._acc = 0.0 # W*s of offtake in the open block
self._pp_acc = [0.0] * phases
self._elapsed = 0.0 # seconds integrated in the open block
self._last_t: float | None = None # epoch seconds of the held sample
self._last_net = 0.0
self._last_pp: tuple[float, ...] | None = None
# -- reading ------------------------------------------------------------
@property
def block_start(self) -> datetime | None:
if self._block is None:
return None
return datetime.fromtimestamp(self._block, timezone.utc)
@property
def elapsed_s(self) -> float:
"""Seconds already integrated into the open block.
Exposed alongside the partial accumulator because SAFETY-07 projects the
end-of-quarter average and cannot do that from a finished average.
"""
return self._elapsed
@property
def partial_ws(self) -> float:
"""Offtake watt-seconds accumulated in the open block so far."""
return self._acc
@property
def offtake_avg_w(self) -> float:
"""Average offtake over the part of the open block seen so far."""
return self._acc / self._elapsed if self._elapsed > 0 else 0.0
@property
def per_phase_offtake_avg_w(self) -> tuple[float, ...] | None:
if self._last_pp is None or self._elapsed <= 0:
return None
return tuple(a / self._elapsed for a in self._pp_acc)
# -- writing ------------------------------------------------------------
def add(self, sample: P1Sample) -> list[QuarterBlock]:
"""Integrate up to this sample, then hold its value. Returns any blocks
that closed in the process (usually none, occasionally one)."""
t = sample.ingest_ts.timestamp()
closed: list[QuarterBlock] = []
if self._last_t is None:
self._block = int(t // QUARTER_S) * QUARTER_S
self._last_t, self._last_net = t, sample.net_w
self._last_pp = sample.per_phase_w
return closed
if t <= self._last_t:
# Out-of-order or duplicate arrival: integrating a negative dt would
# subtract energy that really happened. Drop it, keep the held value.
return closed
cursor = self._last_t
# Beyond this instant the held value stops being evidence of anything.
# The stretch from here to `t` is walked so the block boundaries are
# still crossed correctly, but nothing is accumulated and `_elapsed`
# does not grow - which is what makes a closed block, always divided by
# the full 900 s, actually get dragged down by the missing coverage.
hold_end = self._last_t + self.max_hold_s
while True:
end = self._block + QUARTER_S
stop = min(t, end)
covered = max(0.0, min(stop, hold_end) - cursor)
if covered > 0:
self._acc += max(self._last_net, 0.0) * covered
if self._last_pp is not None:
for i, v in enumerate(self._last_pp[: self.phases]):
self._pp_acc[i] += max(v, 0.0) * covered
self._elapsed += covered
cursor = stop
if stop < end:
break
closed.append(QuarterBlock(
start=datetime.fromtimestamp(self._block, timezone.utc),
# A closed block is always divided by the full 900 s, never by
# the seconds we happened to observe - a gap in coverage must
# drag the billed average down, not be averaged away.
offtake_avg_w=self._acc / QUARTER_S,
per_phase_offtake_avg_w=(
tuple(a / QUARTER_S for a in self._pp_acc)
if self._last_pp is not None else None),
))
self._block = end
self._acc = 0.0
self._pp_acc = [0.0] * self.phases
self._elapsed = 0.0
self._last_t, self._last_net = t, sample.net_w
self._last_pp = sample.per_phase_w
return closed
# --------------------------------------------------------------------------- #
# what the rest of the add-on talks to
# --------------------------------------------------------------------------- #
class P1Ingest:
"""Holds the latest sample and the rolling quarter-hour average.
⚠️ Staleness is DERIVED from the stored sample, not carried as a separate
flag. That is what makes "flag before the value is visible" free: there is
one immutable object and a single attribute rebind to publish it, so a
reader can never see a fresh value with a stale flag or the reverse.
"""
def __init__(self, phases: int = 1, max_age_s: float = 30.0):
self.phases = phases
self.max_age_s = float(max_age_s)
# The same threshold governs control and billing: a reading too old to
# steer by is too old to bill by. See QuarterAverager.max_hold_s.
self.averager = QuarterAverager(phases, max_hold_s=self.max_age_s)
self.blocks: list[QuarterBlock] = []
self.samples = 0
self.parse_errors = 0
self.last_error: str | None = None
self.started_mono = time.monotonic()
self._last: P1Sample | None = None
@property
def last(self) -> P1Sample | None:
return self._last
def submit(self, sample: P1Sample) -> None:
self._last = sample
self.samples += 1
for block in self.averager.add(sample):
self.blocks.append(block)
del self.blocks[:-96] # a day of quarters; STATE-01 owns real retention
def reject(self, err: Exception | str) -> None:
"""A malformed telegram or an unavailable entity.
⚠️ The last good sample and ITS timestamp are left untouched. The reading
does not become 0 W and it does not become fresh - the age keeps growing,
which is precisely the signal a rejected telegram should produce.
"""
self.parse_errors += 1
self.last_error = str(err)
_LOG.warning("P1 telegram rejected: %s", err)
# -- what consumers read -------------------------------------------------
def age_s(self) -> float | None:
"""Age of the newest accepted sample, or None if there has never been one."""
return None if self._last is None else self._last.age_s()
@property
def published_age_s(self) -> float:
"""The figure behind `sensor.p1_sample_age_s`.
Seconds since the newest accepted telegram, or since this ingester
started when none has ever arrived.
⚠️ Always a number and never `unknown`, because SAFETY-01's firmware
watchdog subscribes to it: an entity that simply stops existing is
indistinguishable, from the firmware's side, from a meter that is fine.
And ⚠️ it is recomputed against the clock on every publish rather than
stamped once per telegram, so a meter that freezes at a constant reading
still produces a visibly climbing age. That is the whole point of this
entity - HA pushes state changes, so a genuinely constant P1 value emits
nothing at all, and a watchdog watching the value would sit there
believing the last update was recent.
"""
age = self.age_s()
return max(0.0, time.monotonic() - self.started_mono) if age is None else age
@property
def stale(self) -> bool:
"""True when there is no sample, or the newest one is past max_age_s."""
age = self.age_s()
return age is None or age > self.max_age_s
@property
def net_w(self) -> float | None:
"""Signed net grid power, or None when stale. Never a substituted zero."""
return None if self.stale else self._last.net_w
@property
def per_phase_import_w(self) -> tuple[float, ...] | None:
if self.stale or self._last is None:
return None
return self._last.per_phase_import_w
# --------------------------------------------------------------------------- #
# transport 1: Home Assistant WebSocket (the DSMR integration's entities)
# --------------------------------------------------------------------------- #
WS_URL = "ws://supervisor/core/websocket"
BAD_STATES = ("unknown", "unavailable", "none", "")
class HaDsmrSource:
"""Subscribes to state_changed for the configured DSMR entities.
⚠️ WebSocket, not REST polling. REST returns states, but polling at the
30 s planning tick decimates a 5 s telegram stream and the quarter-hour
average would then be computed from a sixth of the data (§5.3). "Consume
every telegram" means event-driven.
⚠️ One telegram updates several entities, and HA emits one state_changed per
entity. Building a sample on each event would mix a new import reading with
a stale export one for a few milliseconds every 5 s. A short debounce
coalesces the burst back into the single telegram it came from.
"""
DEBOUNCE_S = 0.35
def __init__(self, session: aiohttp.ClientSession, ingest: P1Ingest,
entities: dict, token: str | None = None):
self.session = session
self.ingest = ingest
self.entities = entities # {"import": id, "export": id, "phase_import": [...], ...}
self.token = token or os.environ.get("SUPERVISOR_TOKEN", "")
self.ids = self._wanted()
self.cache: dict[str, float] = {}
self.connected = False
self._pending: asyncio.Task | None = None
def _wanted(self) -> set[str]:
out = set()
for key in ("import", "export"):
if self.entities.get(key):
out.add(self.entities[key])
for key in ("phase_import", "phase_export"):
out.update(e for e in self.entities.get(key) or [] if e)
return out
async def run(self) -> None:
"""Long-lived task: connect, subscribe, reconnect with backoff, forever.
⚠️ A reconnect emits nothing. A gap must stay a gap - a synthetic sample
on reconnect would reset the age and hide the outage from the very
watchdog that exists to catch it.
"""
backoff = 1.0
while True:
try:
await self._session_once()
backoff = 1.0
except asyncio.CancelledError:
raise
except Exception as err: # noqa: BLE001 - any transport fault retries
_LOG.warning("P1 HA websocket: %s - reconnecting in %.0fs", err, backoff)
finally:
self.connected = False
await asyncio.sleep(backoff)
backoff = min(backoff * 2, 30.0)
async def _session_once(self) -> None:
async with self.session.ws_connect(WS_URL, heartbeat=30) as ws:
hello = await ws.receive_json()
if hello.get("type") == "auth_required":
await ws.send_json({"type": "auth", "access_token": self.token})
reply = await ws.receive_json()
if reply.get("type") != "auth_ok":
raise RuntimeError(f"auth rejected: {reply.get('message', reply)}")
await ws.send_json({"id": 1, "type": "subscribe_events",
"event_type": "state_changed"})
await ws.send_json({"id": 2, "type": "get_states"})
self.connected = True
_LOG.info("P1 ingest: subscribed to %s", ", ".join(sorted(self.ids)))
async for msg in ws:
if msg.type is not aiohttp.WSMsgType.TEXT:
continue
payload = json.loads(msg.data)
if payload.get("id") == 2 and payload.get("type") == "result":
# ⚠️ Prime the cache, but do NOT build a sample from it.
# get_states returns whatever HA currently holds, which
# after a Core restart is a RestoreEntity value of unknown
# age. Stamping that with ingest_ts=now resets the age to
# zero and reports a fresh meter that may have been dead for
# an hour - a synthetic sample hiding the outage from the
# watchdog that exists to catch it. The cache is what lets
# the FIRST real state_changed build a complete sample; the
# age stays honest until one arrives.
for obj in payload.get("result") or []:
self._absorb(obj.get("entity_id"), obj.get("state"))
elif payload.get("type") == "event":
data = (payload.get("event") or {}).get("data") or {}
if data.get("entity_id") not in self.ids:
continue
new = data.get("new_state") or {}
self._absorb(data.get("entity_id"), new.get("state"))
self._schedule()
raise RuntimeError("websocket closed")
def _absorb(self, entity_id: str | None, state) -> None:
if not entity_id or entity_id not in self.ids:
return
raw = str(state).strip().lower()
if raw in BAD_STATES:
# ⚠️ An `unavailable` DSMR entity is a missing reading, not 0 W.
# Forget the cached value so no sample can be built from a mixture
# of a live register and one that stopped reporting.
self.cache.pop(entity_id, None)
self.ingest.reject(f"{entity_id} is {raw}")
return
try:
self.cache[entity_id] = float(raw)
except ValueError:
self.cache.pop(entity_id, None)
self.ingest.reject(f"{entity_id} is not numeric: {raw!r}")
def _schedule(self) -> None:
if self._pending and not self._pending.done():
return
self._pending = asyncio.get_running_loop().create_task(self._after_debounce())
async def _after_debounce(self) -> None:
await asyncio.sleep(self.DEBOUNCE_S)
self.build()
def build(self) -> bool:
"""Assemble one sample from the cache. Returns True if one was accepted."""
imp_id, exp_id = self.entities.get("import"), self.entities.get("export")
if imp_id not in self.cache or exp_id not in self.cache:
return False
pi = [self.cache.get(e) for e in self.entities.get("phase_import") or []]
pe = [self.cache.get(e) for e in self.entities.get("phase_export") or []]
if pi and (None in pi or (pe and None in pe)):
return False # incomplete phase set: wait, do not guess
try:
self.ingest.submit(make_sample(
SOURCE_HA, self.cache[imp_id], self.cache[exp_id],
phases=self.ingest.phases,
phase_import_w=pi or None,
phase_export_w=pe or None,
# ⚠️ No telegram_ts: HA's last_changed is when the STATE changed,
# which for a constant reading is minutes ago even though the
# telegram is current. Using it as a telegram time would fake
# staleness on a genuinely steady meter.
))
return True
except P1Error as err:
self.ingest.reject(err)
return False
# --------------------------------------------------------------------------- #
# transport 2: MQTT
# --------------------------------------------------------------------------- #
def parse_mqtt_payload(raw: bytes | str, phases: int, *,
now: datetime | None = None) -> P1Sample:
"""One JSON telegram from the configured topic. Raises P1Error.
The accepted document, documented in DOCS.md:
{"import_w": 1234.0, "export_w": 0.0,
"phases": [{"import_w": 500, "export_w": 0}, ...], # optional
"timestamp": "2026-08-24T18:00:05+02:00"} # optional
ponytail: one strict schema rather than sniffing the half-dozen P1-bridge
dialects in the wild. The upgrade path is a `meter_mqtt_format` option
selecting a parser; a lenient parser is the wrong default at a safety
boundary, where guessing a key means guessing a kilowatt.
"""
try:
doc = json.loads(raw)
except (ValueError, TypeError) as err:
raise P1Error(f"payload is not JSON: {err}") from None
if not isinstance(doc, dict):
raise P1Error(f"payload is not a JSON object ({type(doc).__name__})")
ts = None
if doc.get("timestamp"):
try:
ts = datetime.fromisoformat(str(doc["timestamp"]))
except ValueError:
raise P1Error(f"unparseable timestamp {doc['timestamp']!r}") from None
if ts.tzinfo is None:
raise P1Error("timestamp has no UTC offset")
pi = pe = None
if "phases" in doc:
rows = doc["phases"]
if not isinstance(rows, list) or not all(isinstance(r, dict) for r in rows):
raise P1Error("'phases' must be a list of objects")
pi = [r.get("import_w") for r in rows]
pe = [r.get("export_w", 0.0) for r in rows]
return make_sample(SOURCE_MQTT, doc.get("import_w"), doc.get("export_w"),
phases=phases, phase_import_w=pi, phase_export_w=pe,
telegram_ts=ts, ingest_ts=now)
class MqttP1Source:
"""Subscribes to one topic and submits every message that parses.
Uses paho's own reconnect loop on its own thread, then hops back onto the
event loop with call_soon_threadsafe so the ingest state is only ever
mutated from one thread.
"""
def __init__(self, ingest: P1Ingest, topic: str, host, port=1883,
username=None, password=None):
self.ingest = ingest
self.topic = topic
self.host, self.port = host, int(port or 1883)
self.username, self.password = username, password
self.client = None
self.loop = None
async def run(self) -> None:
if mqtt is None or not self.host or not self.topic:
_LOG.error("P1 MQTT source not usable (broker=%s topic=%r) - no meter data",
self.host, self.topic)
return
self.loop = asyncio.get_running_loop()
try:
self.client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2,
client_id="goodwe_p1_ingest")
except AttributeError: # paho 1.x, which is what Alpine ships
self.client = mqtt.Client(client_id="goodwe_p1_ingest")
if self.username:
self.client.username_pw_set(self.username, self.password or "")
self.client.on_connect = lambda *_a, **_k: self.client.subscribe(self.topic, qos=0)
self.client.on_message = self._on_message
self.client.reconnect_delay_set(min_delay=1, max_delay=30)
self.client.connect_async(self.host, self.port, keepalive=60)
self.client.loop_start()
_LOG.info("P1 ingest: MQTT %s:%s topic %s", self.host, self.port, self.topic)
try:
while True:
await asyncio.sleep(3600)
finally:
self.client.loop_stop()
self.client.disconnect()
def _on_message(self, _client, _userdata, msg) -> None:
# Runs on paho's network thread.
if self.loop is None:
return
self.loop.call_soon_threadsafe(self._handle, msg.payload)
def _handle(self, payload) -> None:
try:
self.ingest.submit(parse_mqtt_payload(payload, self.ingest.phases))
except P1Error as err:
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
# --------------------------------------------------------------------------- #
# selection
# --------------------------------------------------------------------------- #
def is_enabled(opts: dict) -> bool:
"""Whether P1 ingestion is switched on at all.
⚠️ One definition, because three places depend on it and they MUST agree:
where the grid reading comes from, whether the ingest task is started, and
whether sensor.p1_sample_age_s is announced over MQTT discovery. An age
sensor announced with no ingester behind it is a watchdog input nobody is
feeding, and the ESP32 trips on it.
"""
return str(opts.get("meter_source", "off") or "off").strip() not in ("off", "")
def build_source(opts: dict, ingest: P1Ingest, session, broker: dict | None):
"""Return the transport named by `meter_source`, or None if disabled.
This is the whole of AC 1's "switchable": every consumer reads P1Ingest, so
changing transport is a config edit, never a code path.
"""
source = str(opts.get("meter_source", "off") or "off").strip()
if not is_enabled(opts):
return None
if source == SOURCE_HA:
return HaDsmrSource(session, ingest, {
"import": opts.get("p1_import_entity", ""),
"export": opts.get("p1_export_entity", ""),
"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_MQTT:
broker = broker or {}
return MqttP1Source(ingest, str(opts.get("meter_mqtt_topic", "")),
broker.get("host"), broker.get("port", 1883),
broker.get("username"), broker.get("password"))
if source:
_LOG.error("meter_source %r is not one of %s - P1 ingestion disabled",
source, ", ".join((SOURCE_HA, SOURCE_MQTT, SOURCE_HA_SIGNED)))
return None
+47 -1
View File
@@ -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
@@ -39,6 +39,34 @@ options:
batt_invert: false
setpoint_entity: ""
# --- 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;
# ha_signed reads ONE signed HA entity (+ import / - export), which is what a
# HomeWizard P1 publishes and what ha_dsmr cannot consume.
meter_source: "off"
meter_phases: 1
meter_max_age_s: 30
meter_mqtt_topic: ""
# 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: ""
p1_export_entity: ""
# Optional, in L1..L3 order. Required for the capacity-tariff peak on a
# 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
gain: 0.6
@@ -81,6 +109,24 @@ schema:
batt_invert: bool
setpoint_entity: str
meter_source: list(off|ha_dsmr|mqtt_p1|ha_signed)
# ⚠️ 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_mqtt_topic: 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)
slew_w: int(50,5000)
+175 -15
View File
@@ -24,6 +24,63 @@ def check(name, cond):
fails.append(name)
# ---------------------------------------------------------------------------
# COVERAGE AUDIT - measured, not executed. Read this before adding a mechanism.
#
# THE INVARIANT: every mechanism in compute() must be noticed by AT LEAST TWO
# checks when it is deleted. If you add a mechanism to compute(), re-run the
# audit and add it to the table. If a figure here drops, a check has started
# passing for a reason other than the one it names.
#
# THE TECHNIQUE, because there is no script to run: replace one mechanism in
# control.py with a no-op, run this file, count the failures, restore. That is
# the converse of the usual mutation - not "does a wrong value fail?" but "does
# anyone notice when the mechanism is GONE?". It is kept as a comment rather
# than as tooling on purpose: the only cheap way to automate it is to key on
# source lines, which goes stale silently, and a green audit that has quietly
# stopped testing anything is precisely the failure this ticket exists to fix.
# A comment cannot go stale-green, because it never claims to be running.
#
# Measured at 389d9ec. Numbers are the lead's independent reproduction.
#
# mechanism in compute() checks that fail when deleted
# ------------------------------------------ -----------------------------
# integrator freeze (AC 3) 2
# integrator clamp (AC 1) 6
# integrator bound follows max_w 4
# output clamp 3
# slew limit 4
# output freeze 2
# deadband 5
# quantisation 2
# saturation detector, `saturated_now = False` 11
# saturation duration (AC 2), fires instantly 2
# sat counter reset on a good cycle 6
# target_grid_w bias 3
# i_w=None seeding from prev_w 6
#
# The detector figure is for the `saturated_now = False` form specifically;
# disabling it further down as `frozen = False` is a weaker mutation and gives
# 10. Reproduce the same form or the number will not match.
#
# ⚠️ IT HAS FOUND A DEAD MECHANISM TWICE, BOTH THE SAME WAY: a clamp standing in
# for the mechanism under test. Deleting the integrator freeze once failed
# NOTHING, because the fixtures sat at max_w 2000 and the integrator bound
# truncated a wound value back to exactly 2000 - the assertion passed on the
# clamp. The output clamp was masked the same way by the integrator bound.
# Hence: A FIXTURE MUST SIT CLEAR OF EVERY RAIL IT IS NOT TESTING. Where a test
# names one mechanism, make that mechanism the binding one (see TCLAMP and TF).
#
# ⚠️ RUN MUTATIONS WITH `python -B` AND CLEAR app/__pycache__. CPython
# invalidates a .pyc on (source mtime in whole seconds, source size), so a
# same-second rewrite that also preserves the file size reuses stale bytecode
# and the suite reports on code you are no longer running. It under-reported one
# mutation here as 2 where the true figure is 6. The error is one-directional -
# stale bytecode can only under-report - so every figure above is a lower bound
# at worst, and the two zeros ever recorded were both confirmed by fixing them
# and watching the count rise, which a caching artefact cannot do.
# ---------------------------------------------------------------------------
print("control law")
# Deadband: inside meter noise, hold exactly - do not drift.
@@ -41,13 +98,22 @@ check("proportional step (gain 0.6)", d.target_w == 300)
d = compute(prev_w=0, grid_w=-500, actual_w=0, tuning=T)
check("export drives charging", d.target_w == -300)
# Clamp
d = compute(prev_w=1900, grid_w=1000, actual_w=1900, tuning=Tuning(max_w=2000, slew_w=5000))
# Clamp.
# ⚠️ integrator_max_w is lifted clear of max_w so that the OUTPUT clamp is the
# mechanism under test. Left at the default the integrator bound truncates
# first, these two assertions pass on that alone, and deleting the output clamp
# fails nothing - the same masking that hid the integrator freeze.
TCLAMP = Tuning(max_w=2000, slew_w=5000, integrator_max_w=5000)
d = compute(prev_w=1900, grid_w=1000, actual_w=1900, tuning=TCLAMP)
check("clamped to max_w", d.target_w == 2000)
# Slew: from 0 with a huge error, no more than slew_w in one cycle.
d = compute(prev_w=0, grid_w=5000, actual_w=0, tuning=Tuning(max_w=5000, slew_w=1000))
check("slew limits one cycle", d.target_w == 1000)
d = compute(prev_w=-1900, grid_w=-1000, actual_w=-1900, tuning=TCLAMP)
check("clamped to -max_w", d.target_w == -2000)
d = compute(prev_w=0, grid_w=-5000, actual_w=0, tuning=Tuning(max_w=5000, slew_w=1000))
check("slew limits one cycle, charging", d.target_w == -1000)
# Saturation needs DURATION: one diverging cycle must NOT freeze.
t = Tuning(saturation_w=500, saturation_cycles=3)
@@ -86,12 +152,12 @@ RUNAWAY_CYCLES = 150
HISTORICAL_W = 14768.0
def runaway(tuning):
def runaway(tuning, sign=1):
"""Inverter off: it reports 0 W forever, the error never clears."""
prev, i_w, sat = 0.0, 0.0, 0
worst_i, worst_cmd = 0.0, 0.0
for _ in range(RUNAWAY_CYCLES):
d = compute(prev_w=prev, grid_w=RUNAWAY_ERROR, actual_w=0.0,
d = compute(prev_w=prev, grid_w=sign * RUNAWAY_ERROR, actual_w=0.0,
tuning=tuning, sat_count=sat, i_w=i_w)
prev, i_w, sat = d.target_w, d.i_w, d.sat_count
worst_i = max(worst_i, abs(i_w))
@@ -113,6 +179,12 @@ check(f"runaway with the detector defeated: integrator still bounded ({wi:.0f} W
wi <= TD.max_w)
check("runaway with the detector defeated: command still <= max_w", wc <= TD.max_w)
# The mirror: the same runaway driving the other way. An export that never
# clears winds the integrator negative just as hard.
wi, wc = runaway(Tuning(max_w=2000), sign=-1)
check(f"runaway (export direction): integrator bounded at {wi:.0f} W", wi <= 2000)
check("runaway (export direction): emitted command <= max_w", wc <= 2000)
# The bound is a separate quantity, and the useful direction is BELOW max_w:
# there it binds first and caps unwind latency tighter than the rail does.
d = compute(prev_w=0, grid_w=6000, actual_w=0,
@@ -121,11 +193,22 @@ check("integrator bound binds independently of the output clamp",
d.i_w == 1000 and d.target_w == 1000)
# Freeze = may not wind further in the direction it is already pushing.
TF = Tuning(saturation_w=500, saturation_cycles=3)
# ⚠️ max_w is raised WELL above the fixtures on purpose. At the default 2000
# the integrator bound truncates a wound value back to exactly 2000 and
# satisfies these assertions on its own, so deleting the freeze outright
# failed nothing - the clamp was standing in for the mechanism under test.
# Any fixture here must sit clear of every rail, or it tests the rail.
TF = Tuning(saturation_w=500, saturation_cycles=3, max_w=5000)
f1 = compute(prev_w=2000, grid_w=800, actual_w=0, tuning=TF, sat_count=3, i_w=2000.0)
check("frozen: integration does not wind further", f1.i_w == 2000.0 and f1.frozen)
f2 = compute(prev_w=2000, grid_w=-800, actual_w=0, tuning=TF, sat_count=3, i_w=2000.0)
check("frozen: unwinding is still allowed", f2.i_w < 2000.0)
# ...and the same two on the charging side. Every freeze rule in this file has
# a mirror, because the one that did not is the defect that got through review.
f3 = compute(prev_w=-2000, grid_w=-800, actual_w=0, tuning=TF, sat_count=3, i_w=-2000.0)
check("frozen (charging): integration does not wind further", f3.i_w == -2000.0)
f4 = compute(prev_w=-2000, grid_w=800, actual_w=0, tuning=TF, sat_count=3, i_w=-2000.0)
check("frozen (charging): unwinding is still allowed", f4.i_w > -2000.0)
# ⚠️ REGRESSION, and the reason the first cut of SAFETY-04 was rejected. A
# freeze encoded as "only corrections that shrink |i_w|" is unsatisfiable for
@@ -145,6 +228,50 @@ check("frozen at i_w=0: the freeze then clears", not z2.frozen)
z3 = compute(prev_w=100, grid_w=-1000, actual_w=800, tuning=T, sat_count=3, i_w=100.0)
check("frozen at i_w=+100: a 1 kW export still moves the command",
z3.frozen and z3.target_w < 0)
z4 = compute(prev_w=-100, grid_w=1000, actual_w=-800, tuning=T, sat_count=3, i_w=-100.0)
check("frozen at i_w=-100: a 1 kW import still moves the command",
z4.frozen and z4.target_w > 0)
# ⚠️ EXACTLY ZERO, BOTH DIRECTIONS. This boundary has a history: the first cut
# deadlocked here under import, and the fix for it deadlocked here under export
# because `if i_w > 0 ... else ...` files 0.0 under rising-only. main.py resets
# i_w to exactly 0.0 on every stop and every reseed, so it is a normal state,
# not a corner.
zi = compute(prev_w=0, grid_w=2000, actual_w=600, tuning=T, sat_count=3, i_w=0.0)
check("frozen at i_w=0.0: an import push moves the integrator",
zi.frozen and zi.i_w > 0)
ze = compute(prev_w=0, grid_w=-2000, actual_w=-600, tuning=T, sat_count=3, i_w=0.0)
check("frozen at i_w=0.0: an export push moves the integrator",
ze.frozen and ze.i_w < 0)
# The COMMAND still holds at 0 W in that second case, and that is release/1.0's
# rule, not a leftover: at prev_w == 0 the output freeze forbids starting to
# charge while saturated, because commanding 0 while the inverter reports
# hundreds of watts means something else is driving the bus. Asserted so that
# nobody "fixes" it by accident - the integrator moving is what this ticket
# owns, the command rule belongs to the output freeze.
check("frozen at i_w=0.0: the output freeze still blocks a charge from 0 W",
ze.target_w == 0.0)
# Where prev_w is already charging the output freeze does NOT block, and there
# the difference reaches the wire: held at 0.0 the integrator abandons the
# charge mid-export.
zc = compute(prev_w=-2000, grid_w=-4000, actual_w=-600, tuning=T, sat_count=3, i_w=0.0)
check("frozen at i_w=0.0: a charge is not abandoned during heavy export",
zc.target_w == -2000.0)
# The general property, rather than another handful of points: while frozen the
# integrator may be held ONLY when the correction would push it further from
# zero on the side it already sits. Any other hold is a deadlock.
stuck = []
for i0 in [x * 25.0 for x in range(-80, 81)]:
for g in [x * 100.0 for x in range(-40, 41)]:
err = g - T.target_grid_w
if abs(err) < T.deadband_w:
continue
dd = compute(prev_w=0.0, grid_w=g, actual_w=1500.0, tuning=T, sat_count=3, i_w=i0)
if dd.i_w == i0 and not ((i0 > 0 and err > 0) or (i0 < 0 and err < 0)):
stuck.append((i0, g))
check(f"frozen integrator never deadlocks, over {161*81} states"
+ (f" (e.g. {stuck[0]})" if stuck else ""), not stuck)
# False-positive guard: a normal 2 kW load step must not trip the detector,
# because the plant needs several cycles to catch up on every one of them.
@@ -196,34 +323,67 @@ print("SAFETY-04: the i_w=None path is still release/1.0, exactly")
def legacy(prev, grid, actual, t, sat_count):
"""release/1.0's control law, transcribed. Do not 'improve' this."""
reason = "tracking"
sc = min(sat_count + 1, 10) if abs(prev - actual) > t.saturation_w else 0
frozen = sc >= t.saturation_cycles
error = grid - t.target_grid_w
want = prev if abs(error) < t.deadband_w else prev + t.gain * error
if abs(error) < t.deadband_w:
want, reason = prev, "deadband"
else:
want = prev + t.gain * error
target = max(-t.max_w, min(t.max_w, want))
target = max(prev - t.slew_w, min(prev + t.slew_w, target))
if target != want:
reason = "clamped"
slewed = max(prev - t.slew_w, min(prev + t.slew_w, target))
if slewed != target:
reason = "slew-limited"
target = slewed
if frozen:
target = min(target, prev) if prev > 0 else max(target, prev)
reason = "saturated-freeze"
step = max(1, int(t.step_w))
return float(round(target / step) * step), sc
return float(round(target / step) * step), sc, frozen, reason
# Exhaustive over the interesting corners, both freeze states, both signs, and
# either side of the deadband. This is what makes the claim in control.py's
# integrator comment a checked fact rather than an assertion.
# ⚠️ Compare EVERYTHING observable, not just the number. A previous version of
# this sweep compared (target_w, sat_count) only and passed 3024 cases while
# `reason` had silently lost a value - which is the kind of thing a sweep this
# broad exists to catch. `frozen` and `reason` are both in the tuple now.
#
# The one deliberate rename: what release/1.0 called "clamped" is now
# "i-clamped", because the truncation happens on the integrator before the
# command is derived from it. Aliased here rather than papered over - if any
# OTHER reason ever diverges, this check goes red.
ALIAS = {"i-clamped": "clamped"}
diffs = []
seen = set()
for tune in (Tuning(), Tuning(target_grid_w=-10.0), Tuning(max_w=5000, slew_w=5000)):
for prev in (-2000.0, -500.0, -100.0, 0.0, 100.0, 500.0, 2000.0):
for grid in (-6000.0, -1000.0, -500.0, -14.0, 0.0, 14.0, 500.0, 1000.0, 6000.0):
for actual in (-2000.0, 0.0, 600.0, 2000.0):
for sc in (0, 2, 3, 9):
d = compute(prev, grid, actual, tune, sc) # i_w defaults to None
lt, lsc = legacy(prev, grid, actual, tune, sc)
if (d.target_w, d.sat_count) != (lt, lsc):
diffs.append((prev, grid, actual, sc, d.target_w, lt))
check(f"i_w=None reproduces release/1.0 over {3*7*9*4*4} cases"
seen.add(d.reason)
got = (d.target_w, d.sat_count, d.frozen,
ALIAS.get(d.reason, d.reason))
if got != legacy(prev, grid, actual, tune, sc):
diffs.append((prev, grid, actual, sc, got,
legacy(prev, grid, actual, tune, sc)))
check(f"i_w=None reproduces release/1.0 over {3*7*9*4*4} cases, reason included"
+ (f" (first diff {diffs[0]})" if diffs else ""), not diffs)
# ...and the rename is not a quiet deletion: the signal SAFETY-03 alarms on has
# to actually occur in that sweep, or its hook is dead.
check("the integrator clamp reports itself as 'i-clamped'", "i-clamped" in seen)
# "clamped" stays reachable, but only where the integrator is deliberately
# allowed above the rail - then BOTH fire and the output clamp, which describes
# the value actually emitted, is the one reported.
dc = compute(prev_w=0, grid_w=6000, actual_w=0,
tuning=Tuning(max_w=2000, integrator_max_w=3000, slew_w=5000))
check("the output clamp still reports 'clamped' when it is the binding one",
dc.reason == "clamped" and dc.i_w == 3000 and dc.target_w == 2000)
print("capacity tariff")
check("no forecast means no cap", maintenance_charge_floor(2500, None, 3500) == 2500)
check("headroom caps the charge", maintenance_charge_floor(2500, 2000, 3500) == 1500)
+965
View File
@@ -0,0 +1,965 @@
"""Runnable check for P1 ingestion. `python3 test_p1.py`
No framework, no fixtures, no meter - it has to run on a tech's laptop and in CI
with nothing installed and nothing plugged in (§17: "usable without real
hardware").
Every assert here is a rule whose absence poisons something downstream: an
inverted sign inverts the control loop, a fabricated zero hides a dead meter, a
naive per-phase sum misbills the capacity tariff, and a plain mean of samples
computes the wrong quarter-hour figure whenever the telegram cadence changes.
"""
import asyncio
import sys
import time
from datetime import datetime, timedelta, timezone
import aiohttp # already required by app.p1, so this adds no new dependency
from app.p1 import (
P1Error, P1Ingest, HaDsmrSource, HaSignedSource, QuarterAverager,
SOURCE_HA, SOURCE_HA_SIGNED, SOURCE_MQTT,
build_source, is_enabled, make_sample, parse_mqtt_payload, split_signed,
)
fails = []
total = 0
TZ_BE_SUMMER = timezone(timedelta(hours=2))
TZ_BE_WINTER = timezone(timedelta(hours=1))
BASE = datetime(2026, 8, 24, 10, 0, 0, tzinfo=timezone.utc) # a quarter boundary
def check(name, cond):
global total
total += 1
if cond:
print(f" ok {name}")
else:
print(f" FAIL {name}")
fails.append(name)
def raises(name, fn):
global total
total += 1
try:
fn()
except P1Error:
print(f" ok {name}")
return
except Exception as err: # noqa: BLE001
print(f" FAIL {name} (raised {type(err).__name__}, wanted P1Error)")
fails.append(name)
return
print(f" FAIL {name} (no error raised)")
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,
ingest_ts=at, ingest_mono=at.timestamp())
# --------------------------------------------------------------------------- #
print("import/export -> signed net (the derivation the EMS owns)")
s = sample(1500.0, 0.0)
check("pure import is positive", s.net_w == 1500.0)
s = sample(0.0, 900.0)
check("pure export is negative", s.net_w == -900.0)
# The single test that catches an inverted control loop.
s = sample(120.0, 2000.0)
check("export-dominant telegram yields negative net", s.net_w == -1880.0)
s = sample(2000.0, 120.0)
check("import-dominant telegram yields positive net", s.net_w == 1880.0)
# Both registers non-zero at once is real: a three-phase house can import on one
# phase and export on another in the same telegram.
s = sample(400.0, 400.0)
check("both registers equal nets to exactly zero", s.net_w == 0.0)
s = sample(0.0, 0.0)
check("both registers zero is a valid balanced reading", s.net_w == 0.0
and s.import_w == 0.0 and s.export_w == 0.0)
check("the magnitudes survive the derivation",
sample(300.0, 50.0).import_w == 300.0 and sample(300.0, 50.0).export_w == 50.0)
# --------------------------------------------------------------------------- #
print("rejecting a telegram instead of believing it")
raises("negative 'unsigned' import is rejected", lambda: sample(-100.0, 0.0))
raises("negative 'unsigned' export is rejected", lambda: sample(0.0, -100.0))
raises("a non-numeric register is rejected", lambda: sample("n/a", 0.0))
raises("None is rejected, not read as zero", lambda: sample(None, 0.0))
raises("NaN is rejected", lambda: sample(float("nan"), 0.0))
raises("infinity is rejected", lambda: sample(float("inf"), 0.0))
# §20 open question 5: an HA sensor reporting 64954 for -582 W, i.e. an unsigned
# 16-bit register decoded without its sign. Must not average in as 65 kW.
raises("the 64954 signed-decode contamination is rejected",
lambda: sample(64954.0, 0.0))
raises("a naive timestamp is rejected",
lambda: make_sample(SOURCE_MQTT, 100, 0, phases=1,
telegram_ts=datetime(2026, 8, 24, 10, 0, 0)))
# --------------------------------------------------------------------------- #
print("single- and three-phase")
s = sample(800.0, 0.0, phases=1, pi=[800.0], pe=[0.0])
check("single phase accepts one phase", s.per_phase_w == (800.0,))
check("per-phase tuple, not list", isinstance(s.per_phase_w, tuple))
raises("three phases configured, one delivered -> rejected",
lambda: sample(800.0, 0.0, phases=3, pi=[800.0], pe=[0.0]))
raises("one phase configured, three delivered -> rejected",
lambda: sample(800.0, 0.0, phases=1, pi=[300.0, 300.0, 200.0],
pe=[0.0, 0.0, 0.0]))
s = sample(600.0, 0.0)
check("no phase data leaves per-phase None, not a fabricated tuple",
s.per_phase_w is None and s.per_phase_import_w is None)
# §17's three-phase unbalanced-load regression case. L1 imports hard, L2 exports,
# L3 idles: the connection nets to 600 W of offtake while 2100 W is drawn across
# the phases. This is the case a naive per-phase sum gets wrong.
s = sample(600.0, 0.0, phases=3, pi=[2000.0, 0.0, 100.0], pe=[0.0, 1500.0, 0.0])
check("unbalanced: per-phase net keeps the export phase negative",
s.per_phase_w == (2000.0, -1500.0, 100.0))
check("unbalanced: per-phase IMPORT clamps the exporting phase to zero",
s.per_phase_import_w == (2000.0, 0.0, 100.0))
check("unbalanced: per-phase import sums to 2100 W, the connection nets 600 W",
sum(s.per_phase_import_w) == 2100.0 and s.net_w == 600.0)
check("unbalanced: the naive sum is NOT the billed figure",
sum(s.per_phase_import_w) != s.net_w)
# --------------------------------------------------------------------------- #
print("rolling 15-minute average (time-weighted, clock-aligned, offtake only)")
# Irregular spacing, hand-computed:
# 0->3 s held at 1000 W -> 3000 Ws
# 3->13 s held at 0 W -> 0 Ws
# 13->20 s held at 2000 W -> 14000 Ws
# total 17000 Ws over 20 s -> 850 W
a = QuarterAverager(1)
a.add(sample(1000.0, at=BASE))
a.add(sample(0.0, at=BASE + timedelta(seconds=3)))
a.add(sample(2000.0, at=BASE + timedelta(seconds=13)))
a.add(sample(2000.0, at=BASE + timedelta(seconds=20)))
check("irregular spacing integrates to the hand-computed 850 W",
abs(a.offtake_avg_w - 850.0) < 1e-9)
check("elapsed-seconds-in-block is exposed for SAFETY-07", a.elapsed_s == 20.0)
check("partial accumulator is exposed for SAFETY-07", a.partial_ws == 17000.0)
# A plain mean over the four samples would be 1250 W. The whole point of the
# time weighting is that these two numbers differ.
check("a plain mean of those samples would have said 1250 W, not 850",
abs((1000 + 0 + 2000 + 2000) / 4 - 1250.0) < 1e-9 and a.offtake_avg_w != 1250.0)
# Cadence change mid-block: 1 s telegrams for a minute, then 10 s telegrams.
# 0->60 s held at 1000 W -> 60000 Ws
# 60->600 s held at 100 W -> 54000 Ws
# 114000 Ws over 600 s -> 190 W
a = QuarterAverager(1)
for i in range(0, 61):
a.add(sample(1000.0 if i < 60 else 100.0, at=BASE + timedelta(seconds=i)))
for i in range(70, 601, 10):
a.add(sample(100.0, at=BASE + timedelta(seconds=i)))
check("a cadence change does not bias the average (190 W)",
abs(a.offtake_avg_w - 190.0) < 1e-9)
check("elapsed tracks the whole 600 s despite the cadence change", a.elapsed_s == 600.0)
# The decimation trap: the fast minute contributes 60 of 114 samples but only
# 10 % of the time, so a per-sample mean lands near 574 W - three times high.
naive = (60 * 1000 + 54 * 100) / 114
check("a per-sample mean would have said ~574 W", 570 < naive < 578)
# Straddling the boundary: 890 s into a block, next telegram 20 s later. Ten
# seconds belong to each block and must be split, not attributed to one.
a = QuarterAverager(1)
a.add(sample(1000.0, at=BASE + timedelta(seconds=890)))
closed = a.add(sample(1000.0, at=BASE + timedelta(seconds=910)))
check("crossing a boundary closes exactly one block", len(closed) == 1)
check("the closed block keeps only its own 10 s (10000/900 W)",
abs(closed[0].offtake_avg_w - 10000.0 / 900.0) < 1e-9)
check("the closed block divides by the full 900 s, so a gap drags it down",
closed[0].offtake_avg_w < 1000.0)
check("the new block carries the other 10 s", a.elapsed_s == 10.0
and abs(a.offtake_avg_w - 1000.0) < 1e-9)
check("the closed block starts on a quarter boundary",
closed[0].start == BASE and closed[0].start.minute % 15 == 0)
# A whole block of pure export: the capacity tariff bills offtake, so this is
# 0 kW, never a negative peak.
a = QuarterAverager(1)
a.add(sample(0.0, 3000.0, at=BASE))
closed = a.add(sample(0.0, 3000.0, at=BASE + timedelta(seconds=900)))
check("a block of pure export averages to 0 W of offtake",
len(closed) == 1 and closed[0].offtake_avg_w == 0.0)
check("...while the signed sample itself stays negative",
sample(0.0, 3000.0).net_w == -3000.0)
# Clock alignment holds either side of a DST change, because every Belgian UTC
# offset is a whole number of hours and the block grid is 900 s of UTC.
for label, tz in (("summer (+02:00)", TZ_BE_SUMMER), ("winter (+01:00)", TZ_BE_WINTER)):
a = QuarterAverager(1)
odd = datetime(2026, 8, 24, 13, 7, 23, tzinfo=tz)
a.add(sample(500.0, at=odd))
local = a.block_start.astimezone(tz)
check(f"block boundary is local :00/:15/:30/:45 in {label}",
local.minute in (0, 15, 30, 45) and local.second == 0 and local.microsecond == 0)
# Three-phase averaging keeps the phases apart.
a = QuarterAverager(3)
a.add(sample(600.0, 0.0, at=BASE, phases=3, pi=[2000.0, 0.0, 100.0], pe=[0.0, 1500.0, 0.0]))
a.add(sample(600.0, 0.0, at=BASE + timedelta(seconds=100), phases=3,
pi=[2000.0, 0.0, 100.0], pe=[0.0, 1500.0, 0.0]))
check("per-phase offtake averages are held separately",
a.per_phase_offtake_avg_w == (2000.0, 0.0, 100.0))
check("the block's own average is the connection net, not the phase sum",
abs(a.offtake_avg_w - 600.0) < 1e-9)
# An out-of-order arrival must not subtract energy that really happened.
a = QuarterAverager(1)
a.add(sample(1000.0, at=BASE))
a.add(sample(1000.0, at=BASE + timedelta(seconds=10)))
before = a.partial_ws
a.add(sample(1000.0, at=BASE + timedelta(seconds=5)))
check("an out-of-order telegram is dropped, not integrated backwards",
a.partial_ws == before and a.elapsed_s == 10.0)
# ⚠️ The assertion above is NOT sufficient on its own, and that is the whole
# lesson: deleting the guard still passes it, because the negative interval is
# separately refused by the `covered > 0` test. What the guard actually prevents
# is the REWIND - without it the held timestamp moves back to +5 s and the next
# telegram re-integrates the 5..10 s window that was already counted. The damage
# only becomes visible one sample later, so the test has to go one sample later.
a.add(sample(1000.0, at=BASE + timedelta(seconds=20)))
check("...and the held timestamp is not rewound, so the next telegram "
"cannot double-count", a.elapsed_s == 20.0 and a.partial_ws == 20000.0)
# A duplicate telegram (identical timestamp) is the same rule.
a = QuarterAverager(1)
a.add(sample(1000.0, at=BASE))
a.add(sample(1000.0, at=BASE + timedelta(seconds=10)))
a.add(sample(4000.0, at=BASE + timedelta(seconds=10)))
a.add(sample(1000.0, at=BASE + timedelta(seconds=20)))
check("a duplicate timestamp neither re-integrates nor replaces the held value",
a.elapsed_s == 20.0 and a.partial_ws == 20000.0)
# A gap must not be filled with the last held value. The meter dies at 5 kW and
# returns ten minutes later; hold-forward would credit 5 kW x 600 s to the
# capacity-tariff accumulator - a fabricated peak, on a permanent record, from
# data nobody measured.
a = QuarterAverager(1, max_hold_s=30.0)
a.add(sample(5000.0, at=BASE))
a.add(sample(5000.0, at=BASE + timedelta(seconds=600)))
check("a 600 s gap is held for at most max_hold_s, not for the whole gap",
a.partial_ws == 5000.0 * 30.0)
check("the unobserved stretch does not count as elapsed time", a.elapsed_s == 30.0)
closed = a.add(sample(5000.0, at=BASE + timedelta(seconds=900)))
check("the outage drags the billed quarter down instead of inventing a peak",
len(closed) == 1 and abs(closed[0].offtake_avg_w - 300000.0 / 900.0) < 1e-9)
check("...nowhere near the 5000 W a hold-forward would have billed",
closed[0].offtake_avg_w < 400.0)
# The cap must not disturb a normally-spaced stream.
a = QuarterAverager(1, max_hold_s=30.0)
for i in range(0, 121, 5): # a healthy 5 s telegram cadence
a.add(sample(2000.0, at=BASE + timedelta(seconds=i)))
check("a healthy 5 s cadence is untouched by the hold cap",
a.elapsed_s == 120.0 and abs(a.offtake_avg_w - 2000.0) < 1e-9)
# --------------------------------------------------------------------------- #
print("ingest timestamp, age and staleness")
now = time.monotonic()
s = make_sample(SOURCE_HA, 1000, 0, phases=1, ingest_mono=now)
check("a sample carries a tz-aware ingest timestamp",
s.ingest_ts.tzinfo is not None)
check("age is ~0 immediately after ingest", s.age_s(now_mono=now) == 0.0)
check("age grows with elapsed time", s.age_s(now_mono=now + 12.5) == 12.5)
check("age never goes negative on a clock step",
s.age_s(now_mono=now - 100.0) == 0.0)
ing = P1Ingest(phases=1, max_age_s=30.0)
check("no sample yet is stale, not zero", ing.stale is True and ing.net_w is None)
check("no sample yet has no age at all", ing.age_s() is None)
ing.submit(make_sample(SOURCE_HA, 1234, 0, phases=1, ingest_mono=time.monotonic()))
check("a fresh sample is not stale", ing.stale is False)
check("a fresh sample exposes signed net power", ing.net_w == 1234.0)
check("a fresh sample's age is small", 0 <= ing.age_s() < 1.0)
ing.submit(make_sample(SOURCE_HA, 1234, 0, phases=1,
ingest_mono=time.monotonic() - 29.0))
check("29 s old with max_age_s 30 is still usable", ing.stale is False)
ing.submit(make_sample(SOURCE_HA, 1234, 0, phases=1,
ingest_mono=time.monotonic() - 31.0))
check("31 s old with max_age_s 30 is stale", ing.stale is True)
check("a stale sample reads as None, never as 0 W", ing.net_w is None)
check("...and its per-phase import is None too", ing.per_phase_import_w is None)
# The MQTT retained-message trap: replayed on reconnect with a fresh receive
# time but a ten-minute-old telegram time. Fresh ingest must not launder it.
old = datetime.now(timezone.utc) - timedelta(minutes=10)
ing = P1Ingest(phases=1, max_age_s=30.0)
ing.submit(make_sample(SOURCE_MQTT, 1000, 0, phases=1, telegram_ts=old,
ingest_mono=time.monotonic()))
check("a retained telegram is stale on arrival despite a fresh receive time",
ing.stale is True and ing.age_s() > 590)
# A rejected telegram must not refresh anything and must not become 0 W.
ing = P1Ingest(phases=1, max_age_s=30.0)
ing.submit(make_sample(SOURCE_HA, 1500, 0, phases=1,
ingest_mono=time.monotonic() - 10.0))
stamp = ing.last.ingest_mono
ing.reject("malformed telegram")
check("a rejected telegram counts as a parse error", ing.parse_errors == 1)
check("a rejected telegram leaves the last good value in place",
ing.last.net_w == 1500.0)
check("a rejected telegram does not refresh the timestamp",
ing.last.ingest_mono == stamp)
check("a rejected telegram does not resolve to 0 W", ing.net_w == 1500.0)
# Meter goes stale but stays connected (§17 failure injection): no new sample
# ever arrives, and the age keeps climbing past max_age_s on its own.
ing = P1Ingest(phases=1, max_age_s=30.0)
ing.submit(make_sample(SOURCE_HA, 800, 0, phases=1,
ingest_mono=time.monotonic() - 120.0))
check("connected-but-silent meter trips staleness with no new telegram",
ing.stale is True and ing.age_s() > 100)
# sensor.p1_sample_age_s: SAFETY-01's firmware subscribes to this, so it must
# always be a number and must climb while nothing arrives.
ing = P1Ingest(phases=1, max_age_s=30.0)
ing.started_mono = time.monotonic() - 45.0
check("the published age is a number before the first telegram ever arrives",
isinstance(ing.published_age_s, float) and ing.published_age_s > 44)
check("...and it is never None, unlike the raw age",
ing.age_s() is None and ing.published_age_s is not None)
ing.submit(make_sample(SOURCE_HA, 500, 0, phases=1, ingest_mono=time.monotonic()))
check("a telegram resets the published age", ing.published_age_s < 1.0)
# The false-trip this entity exists to remove: the meter keeps sending, the
# VALUE never changes, and the age must still reflect that it is being sent.
for _ in range(3):
ing.submit(make_sample(SOURCE_HA, 500, 0, phases=1, ingest_mono=time.monotonic()))
check("an unchanging meter value still reads as fresh while telegrams arrive",
ing.stale is False and ing.published_age_s < 1.0)
ing.submit(make_sample(SOURCE_HA, 500, 0, phases=1,
ingest_mono=time.monotonic() - 90.0))
check("the same unchanging value reads as stale once the telegrams stop",
ing.stale is True and ing.published_age_s > 89)
# --------------------------------------------------------------------------- #
print("MQTT transport: parsing a telegram off the topic")
good = '{"import_w": 1200.5, "export_w": 0}'
s = parse_mqtt_payload(good, 1)
check("a well-formed payload parses", s.net_w == 1200.5 and s.source == SOURCE_MQTT)
s = parse_mqtt_payload('{"import_w": 0, "export_w": 2500}', 1)
check("an export payload parses to a negative net", s.net_w == -2500.0)
s = parse_mqtt_payload(
'{"import_w": 600, "export_w": 0, "phases":'
' [{"import_w":2000,"export_w":0},{"import_w":0,"export_w":1500},'
' {"import_w":100,"export_w":0}]}', 3)
check("a three-phase payload parses per-phase",
s.per_phase_w == (2000.0, -1500.0, 100.0))
s = parse_mqtt_payload(
'{"import_w": 100, "export_w": 0, "timestamp": "2026-08-24T12:00:00+02:00"}', 1)
check("a telegram timestamp is kept when the payload has one",
s.telegram_ts == datetime(2026, 8, 24, 12, 0, 0, tzinfo=TZ_BE_SUMMER))
raises("a non-JSON payload is rejected", lambda: parse_mqtt_payload("not json", 1))
raises("a JSON array is rejected", lambda: parse_mqtt_payload("[1,2,3]", 1))
raises("a payload missing import_w is rejected",
lambda: parse_mqtt_payload('{"export_w": 0}', 1))
raises("a payload with a null register is rejected",
lambda: parse_mqtt_payload('{"import_w": null, "export_w": 0}', 1))
raises("a payload with a string register is rejected",
lambda: parse_mqtt_payload('{"import_w": "1200", "export_w": 0}', 1))
raises("a timestamp with no UTC offset is rejected",
lambda: parse_mqtt_payload(
'{"import_w":1,"export_w":0,"timestamp":"2026-08-24T12:00:00"}', 1))
raises("a phase count that disagrees with meter_phases is rejected",
lambda: parse_mqtt_payload(
'{"import_w":1,"export_w":0,"phases":[{"import_w":1,"export_w":0}]}', 3))
# --------------------------------------------------------------------------- #
print("HA DSMR transport: building a sample out of entity states")
ENT = {"import": "sensor.p1_import", "export": "sensor.p1_export",
"phase_import": [], "phase_export": []}
ing = P1Ingest(phases=1, max_age_s=30.0)
src = HaDsmrSource(None, ing, ENT, token="x")
src._absorb("sensor.p1_import", "1000")
check("one entity alone does not build a sample", src.build() is False
and ing.last is None)
src._absorb("sensor.p1_export", "0")
check("both entities present builds one sample", src.build() is True
and ing.net_w == 1000.0)
# ⚠️ The reason the debounce exists: HA emits one state_changed per entity, so
# mid-telegram the cache briefly holds a new import with the old export.
src._absorb("sensor.p1_import", "0")
src._absorb("sensor.p1_export", "2500")
src.build()
check("a coalesced telegram lands as one consistent sample", ing.net_w == -2500.0)
before = ing.last
src._absorb("sensor.p1_export", "unavailable")
check("an unavailable entity is a parse error", ing.parse_errors == 1)
check("an unavailable entity does not build a sample from a stale half",
src.build() is False)
check("an unavailable entity leaves the last good sample untouched",
ing.last is before and ing.net_w == -2500.0)
src._absorb("sensor.p1_export", "unknown")
check("an unknown entity is treated the same way", ing.parse_errors == 2)
src._absorb("sensor.p1_export", "banana")
check("a non-numeric entity state is a parse error, not 0 W",
ing.parse_errors == 3 and ing.net_w == -2500.0)
src._absorb("sensor.p1_export", "64954")
src._absorb("sensor.p1_import", "0")
check("the contaminated signed decode is refused at the transport too",
src.build() is False and ing.parse_errors == 4)
ing3 = P1Ingest(phases=3, max_age_s=30.0)
ENT3 = {"import": "sensor.p1_import", "export": "sensor.p1_export",
"phase_import": ["sensor.l1_i", "sensor.l2_i", "sensor.l3_i"],
"phase_export": ["sensor.l1_e", "sensor.l2_e", "sensor.l3_e"]}
src3 = HaDsmrSource(None, ing3, ENT3, token="x")
for eid, val in (("sensor.p1_import", "600"), ("sensor.p1_export", "0"),
("sensor.l1_i", "2000"), ("sensor.l2_i", "0")):
src3._absorb(eid, val)
check("an incomplete phase set waits instead of guessing", src3.build() is False)
for eid, val in (("sensor.l3_i", "100"), ("sensor.l1_e", "0"),
("sensor.l2_e", "1500"), ("sensor.l3_e", "0")):
src3._absorb(eid, val)
check("a complete three-phase set builds", src3.build() is True)
check("three-phase entities produce the unbalanced per-phase tuple",
ing3.last.per_phase_w == (2000.0, -1500.0, 100.0))
check("the sample's per-phase tuple length matches meter_phases",
len(ing3.last.per_phase_w) == ing3.phases == 3)
# --------------------------------------------------------------------------- #
print("HA DSMR transport: end to end against a fake Home Assistant")
# Everything above pokes at build()/_absorb() directly. This one drives the real
# thing over a real websocket - auth handshake, subscribe_events, get_states,
# per-entity events - because "the transport works" is otherwise an untested
# claim about a protocol nobody re-reads.
async def _e2e():
from aiohttp import web
import app.p1 as p1mod
done = asyncio.Event()
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"
assert sub["event_type"] == "state_changed"
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": "sensor.p1_import", "state": "1000"},
{"entity_id": "sensor.p1_export", "state": "0"},
{"entity_id": "sensor.something_else", "state": "hello"},
]})
# One telegram, two state_changed events - exactly how HA emits it.
for imp, exp in ((1500, 0), (0, 800)):
await asyncio.sleep(0.5)
for eid, val in (("sensor.p1_import", imp), ("sensor.p1_export", exp)):
await ws.send_json({"type": "event", "event": {"data": {
"entity_id": eid,
"new_state": {"entity_id": eid, "state": str(val)}}}})
await asyncio.sleep(0.5)
await ws.send_json({"type": "event", "event": {"data": {
"entity_id": "sensor.p1_export",
"new_state": {"entity_id": "sensor.p1_export", "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 = HaDsmrSource(sess, ing, dict(ENT), 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())
check("the websocket handshake and subscription complete", live.samples >= 1)
# ⚠️ TWO, not three. get_states primes the cache but must NOT build a sample:
# HA returns whatever it currently holds, which after a Core restart is a
# RestoreEntity value of unknown age, and stamping that with ingest_ts=now
# resets the age and reports a fresh meter that may have been dead for an hour.
# Only the two real state_changed telegrams become samples. Four state_changed
# events arrived (two per telegram); the debounce is what makes those two
# consistent samples rather than four half-updated ones.
check("connecting does not manufacture a sample from cached HA state",
live.samples == 2)
check("the final export-dominant telegram nets negative",
live.last.net_w == -800.0)
check("the sample was built over the wire, tagged with its transport",
live.last.source == SOURCE_HA)
check("an entity we did not subscribe to is never cached",
"sensor.something_else" not in wire.cache and len(wire.cache) == 1)
check("a mid-stream unavailable is a parse error, not a sample",
live.parse_errors == 1 and live.samples == 2)
check("the last good reading survives the unavailable", live.net_w == -800.0)
check("the averager integrated the live stream", live.averager.elapsed_s > 0.2)
# The reason get_states still matters: it is what lets the FIRST real telegram
# build a complete sample instead of waiting for every entity to change once.
check("the primed cache let the first telegram build immediately",
live.samples == 2 and live.last.import_w == 0.0)
# --------------------------------------------------------------------------- #
print("ha_signed: one signed entity -> the same two magnitudes")
# The meter actually fitted at this house is a HomeWizard P1 publishing ONE
# signed sensor. ha_dsmr cannot read it - it wants two unsigned registers and
# refuses a negative one, which is every exporting telegram.
check("a positive reading is import", split_signed(1500.0) == (1500.0, 0.0))
check("a negative reading is export", split_signed(-900.0) == (0.0, 900.0))
check("zero is a balanced reading, not a missing one",
split_signed(0.0) == (0.0, 0.0))
check("exactly one magnitude is ever non-zero",
all(a == 0.0 or b == 0.0 for a, b in
(split_signed(v) for v in (-5710.0, -1.0, 0.0, 1.0, 4384.0))))
# ⚠️ The split must not change the number. A control loop steered by a value
# that was rounded or clamped on the way in is steered by a different meter.
check("the split round-trips the signed value exactly",
all(make_sample(SOURCE_HA_SIGNED, *split_signed(v), phases=1).net_w == v
for v in (-11763.0, -5710.0, -0.5, 0.0, 0.5, 775.0, 4384.0)))
raises("a non-numeric signed reading is rejected", lambda: split_signed("n/a"))
raises("a signed None is rejected, not read as zero", lambda: split_signed(None))
raises("a signed NaN is rejected", lambda: split_signed(float("nan")))
raises("a signed infinity is rejected", lambda: split_signed(float("inf")))
# ⚠️ This one matters MORE on the signed path than on the unsigned one. On
# ha_dsmr the §20 contamination is also caught by "unsigned cannot be negative";
# here 64954 arrives as a perfectly well-formed positive signed reading and the
# plausibility ceiling is the only thing standing in front of it.
raises("the 64954 signed-decode contamination is still rejected",
lambda: split_signed(64954.0))
raises("...and its negative twin too", lambda: split_signed(-64954.0))
# --------------------------------------------------------------------------- #
print("ha_signed: the sign convention, against real captured readings")
# ⚠️ Not a datasheet claim. These are literal values out of
# sim/scenarios/ha-p1_meter_active_power-2026-08-{20,23}.json, HA recorder
# exports of sensor.p1_meter_active_power at this house, copied here rather than
# read from that repo so this file still runs on a laptop with nothing installed
# (§17). If the convention were inverted, the physics below would be absurd.
# 2026-08-23T11:46:52Z - the day's most negative reading, 13:46 local, full sun.
s = make_sample(SOURCE_HA_SIGNED, *split_signed(-5710.0), phases=1)
check("the midday PV peak (-5710 W) is EXPORT, not a 5.7 kW draw",
s.net_w == -5710.0 and s.export_w == 5710.0 and s.import_w == 0.0)
# 2026-08-19T22:00:00Z - midnight local, 20 Aug's file starts here. No sun.
s = make_sample(SOURCE_HA_SIGNED, *split_signed(775.0), phases=1)
check("the overnight base load (+775 W) is IMPORT",
s.net_w == 775.0 and s.import_w == 775.0 and s.export_w == 0.0)
# 2026-08-20T12:20:58Z - 14:20 local, the largest export in either capture.
s = make_sample(SOURCE_HA_SIGNED, *split_signed(-11763.0), phases=1)
check("the -11763 W midday extreme is export and survives the ceiling",
s.net_w == -11763.0 and s.export_w == 11763.0)
# 2026-08-23T10:18:28Z - the largest import in the healthy capture.
s = make_sample(SOURCE_HA_SIGNED, *split_signed(4384.0), phases=1)
check("the +4384 W peak is import", s.net_w == 4384.0 and s.import_w == 4384.0)
# The whole convention in one line: night draws, midday feeds back.
check("night is positive and midday is negative, which is the convention",
split_signed(775.0)[0] > 0 and split_signed(-5710.0)[1] > 0)
# --------------------------------------------------------------------------- #
print("ha_signed transport: building a sample out of one entity state")
NET = {"net": "sensor.p1_meter_active_power", "phase_net": []}
ing = P1Ingest(phases=1, max_age_s=30.0)
sig = HaSignedSource(None, ing, NET, token="x")
check("nothing cached yet builds nothing", built(sig) is False and ing.last is None)
sig._absorb("sensor.p1_meter_active_power", "1000")
check("one signed entity is a complete telegram on its own",
built(sig) is True and ing.net_w == 1000.0)
check("the sample is tagged with its own transport",
ing.last.source == SOURCE_HA_SIGNED)
sig._absorb("sensor.p1_meter_active_power", "-2500")
built(sig)
check("a negative state lands as a negative net", ing.net_w == -2500.0)
before = ing.last
sig._absorb("sensor.p1_meter_active_power", "unavailable")
check("an unavailable signed entity is a parse error", ing.parse_errors == 1)
check("an unavailable entity does not build a sample", built(sig) is False)
# ⚠️ The rule the whole ticket turns on: a missing reading is MISSING. Resolving
# it to 0 W would read as a perfectly balanced house and defeat the staleness
# trigger that FW-01's watchdog is built on.
check("an unavailable entity leaves the last good sample untouched, not 0 W",
ing.last is before and ing.net_w == -2500.0)
sig._absorb("sensor.p1_meter_active_power", "unknown")
check("an unknown signed entity is treated the same way", ing.parse_errors == 2)
sig._absorb("sensor.p1_meter_active_power", "banana")
check("a non-numeric signed state is a parse error, not 0 W",
ing.parse_errors == 3 and ing.net_w == -2500.0)
sig._absorb("sensor.p1_meter_active_power", "64954")
check("64954 is refused at the signed transport too",
built(sig) is False and ing.parse_errors == 4)
sig._absorb("sensor.not_ours", "123")
check("an unsubscribed entity is never cached by the signed transport",
"sensor.not_ours" not in sig.cache)
# A rejected reading must not make the age look fresh - the age is what the
# firmware watchdog reads, and a rejection is exactly when it must keep climbing.
ing = P1Ingest(phases=1, max_age_s=30.0)
sig = HaSignedSource(None, ing, dict(NET), token="x")
ing.submit(make_sample(SOURCE_HA_SIGNED, 1200, 0, phases=1,
ingest_mono=time.monotonic() - 20.0))
sig._absorb("sensor.p1_meter_active_power", "unavailable")
built(sig)
check("a rejected reading does not reset the published age",
ing.published_age_s > 19 and ing.net_w == 1200.0)
ing.submit(make_sample(SOURCE_HA_SIGNED, 1200, 0, phases=1,
ingest_mono=time.monotonic() - 40.0))
check("...and the age keeps climbing past max_age_s on its own",
ing.stale is True and ing.net_w is None)
# The three-phase reading the TEL-04 survey recorded at this house: L1 +2301 W,
# L2 +468 W, L3 -2582 W, netting +187 W. A signed per-phase set splits the same
# way, and the exporting phase must still clamp out of the billed figure.
ing3 = P1Ingest(phases=3, max_age_s=30.0)
NET3 = {"net": "sensor.p1_meter_active_power",
"phase_net": ["sensor.p1_l1", "sensor.p1_l2", "sensor.p1_l3"]}
sig3 = HaSignedSource(None, ing3, NET3, token="x")
for eid, val in (("sensor.p1_meter_active_power", "187"), ("sensor.p1_l1", "2301")):
sig3._absorb(eid, val)
check("an incomplete signed phase set waits instead of guessing", built(sig3) is False)
sig3._absorb("sensor.p1_l2", "468")
sig3._absorb("sensor.p1_l3", "-2582")
check("a complete signed three-phase set builds", built(sig3) is True)
check("signed per-phase entities keep the exporting phase negative",
ing3.last.per_phase_w == (2301.0, 468.0, -2582.0))
check("per-phase IMPORT clamps the exporting phase to zero",
ing3.last.per_phase_import_w == (2301.0, 468.0, 0.0))
check("the phase import sum is 2769 W while the connection nets 187 W",
sum(ing3.last.per_phase_import_w) == 2769.0 and ing3.last.net_w == 187.0)
check("the signed per-phase tuple length matches meter_phases",
len(ing3.last.per_phase_w) == ing3.phases == 3)
sig_bad = HaSignedSource(None, P1Ingest(phases=3, max_age_s=30.0),
{"net": "sensor.net", "phase_net": ["sensor.a", "sensor.b"]},
token="x")
for eid in ("sensor.net", "sensor.a", "sensor.b"):
sig_bad._absorb(eid, "100")
check("two phases delivered against meter_phases 3 is rejected, not padded",
built(sig_bad) is False and sig_bad.ingest.last is None
and sig_bad.ingest.parse_errors == 1)
# --------------------------------------------------------------------------- #
print("ha_signed transport: end to end against a fake Home Assistant")
# The transport is a subclass, so this is what proves the INHERITED machinery -
# auth, subscribe, the get_states priming rule, the reconnect-emits-nothing
# rule - still behaves when only _wanted() and build() were replaced.
async def _e2e_signed():
from aiohttp import web
import app.p1 as p1mod
done = asyncio.Event()
eid = "sensor.p1_meter_active_power"
async def fake_ha(request):
ws = web.WebSocketResponse()
await ws.prepare(request)
await ws.send_json({"type": "auth_required", "ha_version": "2026.8"})
auth = await ws.receive_json()
assert auth["type"] == "auth" and auth["access_token"] == "tok"
await ws.send_json({"type": "auth_ok"})
sub = await ws.receive_json()
assert sub["type"] == "subscribe_events"
await ws.send_json({"id": sub["id"], "type": "result", "success": True})
get = await ws.receive_json()
assert get["type"] == "get_states"
await ws.send_json({"id": get["id"], "type": "result", "success": True, "result": [
{"entity_id": eid, "state": "775.0"},
{"entity_id": "sensor.something_else", "state": "hello"},
]})
# Two real telegrams, both literal captured values: overnight import,
# then the midday export peak.
for val in ("775.0", "-5710.0"):
await asyncio.sleep(0.5)
await ws.send_json({"type": "event", "event": {"data": {
"entity_id": eid,
"new_state": {"entity_id": eid, "state": val}}}})
await asyncio.sleep(0.5)
await ws.send_json({"type": "event", "event": {"data": {
"entity_id": eid,
"new_state": {"entity_id": eid, "state": "unavailable"}}}})
await asyncio.sleep(0.5)
done.set()
return ws
srv = web.Application()
srv.router.add_get("/ws", fake_ha)
runner = web.AppRunner(srv)
await runner.setup()
site = web.TCPSite(runner, "127.0.0.1", 0)
await site.start()
port = site._server.sockets[0].getsockname()[1]
p1mod.WS_URL = f"http://127.0.0.1:{port}/ws"
ing = P1Ingest(phases=1, max_age_s=30.0)
async with aiohttp.ClientSession() as sess:
src = HaSignedSource(sess, ing, dict(NET), token="tok")
task = asyncio.get_running_loop().create_task(src.run())
try:
await asyncio.wait_for(done.wait(), 20)
await asyncio.sleep(0.5)
finally:
task.cancel()
try:
await task
except asyncio.CancelledError:
pass
await runner.cleanup()
return ing, src
live, wire = asyncio.run(_e2e_signed())
# ⚠️ TWO, not three - the same rule as the ha_dsmr e2e. get_states primes the
# cache but must never become a sample: after a Core restart it is a
# RestoreEntity value of unknown age, and stamping it with ingest_ts=now reports
# a fresh meter that may have been dead for an hour.
check("ha_signed does not manufacture a sample from cached HA state",
live.samples == 2)
check("the signed telegrams arrived over a real websocket",
live.last.source == SOURCE_HA_SIGNED)
check("the final export telegram nets negative, over the wire",
live.last.net_w == -5710.0 and live.last.export_w == 5710.0)
check("ha_signed subscribes to the one entity and caches nothing else",
wire.ids == {"sensor.p1_meter_active_power"}
and "sensor.something_else" not in wire.cache)
check("a mid-stream unavailable signed state is a parse error, not a sample",
live.parse_errors == 1 and live.samples == 2)
# ⚠️ And the cached half is DROPPED, so no later telegram can be assembled out
# of a value that stopped reporting.
check("an unavailable entity is evicted from the cache", wire.cache == {})
check("the last good reading survives the unavailable, and is not 0 W",
live.net_w == -5710.0)
check("the averager integrated the live signed stream", live.averager.elapsed_s > 0.2)
# ⚠️ The entity FW-01 waits on. It must be a number here exactly as it is on the
# other transports - the house P1 went 51.1 s and 36.2 s between state changes
# overnight, and without this the watchdog false-trips the battery to 0 W.
check("sensor.p1_sample_age_s is a live number on this transport too",
isinstance(live.published_age_s, float) and live.published_age_s >= 0.0)
# --------------------------------------------------------------------------- #
print("ha_signed: selection by config")
check("ha_signed is enabled", is_enabled({"meter_source": SOURCE_HA_SIGNED}) is True)
# ⚠️ `sel`, not `built` - that name is the build() wrapper defined at the top of
# this file, and rebinding it here silently disarms every check appended below
# this line. Caught in review: an added check went `TypeError: 'HaSignedSource'
# object is not callable` and aborted the suite, which is the exact failure the
# wrapper exists to prevent, reintroduced by a name collision.
sel = build_source({"meter_source": SOURCE_HA_SIGNED,
"p1_net_entity": "sensor.p1_meter_active_power"},
P1Ingest(), None, None)
check("meter_source ha_signed selects the signed transport",
isinstance(sel, HaSignedSource))
check("...wired to p1_net_entity, and subscribed to exactly that one entity",
sel.ids == {"sensor.p1_meter_active_power"})
# ⚠️ The three modes must not bleed into each other: ha_dsmr must keep ignoring
# p1_net_entity, or a half-configured install silently reads the wrong sensor.
plain = build_source({"meter_source": SOURCE_HA,
"p1_import_entity": "sensor.i", "p1_export_entity": "sensor.e",
"p1_net_entity": "sensor.p1_meter_active_power"},
P1Ingest(), None, None)
check("ha_dsmr still selects the unsigned transport and ignores p1_net_entity",
type(plain) is HaDsmrSource and plain.ids == {"sensor.i", "sensor.e"})
check("meter_source off still selects nothing",
build_source({"meter_source": "off"}, P1Ingest(), None, None) is None)
check("an unrecognised meter_source selects nothing rather than guessing",
build_source({"meter_source": "ha_signd"}, P1Ingest(), None, None) is None)
# ⚠️ Caught once at startup, not once per telegram. A source that is wired up
# wrong otherwise fails in the one way indistinguishable from a healthy source
# nobody has sent anything to yet: no samples, a climbing age, the watchdog
# holding the battery at 0 W, and nothing in the log saying why.
check("a blank p1_net_entity is refused rather than silently never receiving",
build_source({"meter_source": SOURCE_HA_SIGNED, "p1_net_entity": ""},
P1Ingest(), None, None) is None)
check("...and whitespace does not sneak past it",
build_source({"meter_source": SOURCE_HA_SIGNED, "p1_net_entity": " "},
P1Ingest(), None, None) is None)
check("a phase list that disagrees with meter_phases is refused at startup",
build_source({"meter_source": SOURCE_HA_SIGNED, "p1_net_entity": "sensor.n",
"p1_phase_net_entities": ["sensor.a", "sensor.b"]},
P1Ingest(phases=3), None, None) is None)
check("a phase list that agrees with meter_phases is accepted",
isinstance(build_source(
{"meter_source": SOURCE_HA_SIGNED, "p1_net_entity": "sensor.n",
"p1_phase_net_entities": ["sensor.a", "sensor.b", "sensor.c"]},
P1Ingest(phases=3), None, None), HaSignedSource))
check("no phase list at all is still fine - per-phase billing is optional",
isinstance(build_source(
{"meter_source": SOURCE_HA_SIGNED, "p1_net_entity": "sensor.n"},
P1Ingest(phases=3), None, None), HaSignedSource))
# --------------------------------------------------------------------------- #
print("the age sensor must not exist when P1 is off")
# ⚠️ This is a fleet-wide regression guard, not a nicety. The ESP32 watchdog
# does `id(p1_age_s).has_state() && id(p1_age_s).state >= max_age_s` and forces
# the layer-1 failsafe. published_age_s counts from P1Ingest.__init__, so if the
# age were published with meter_source off it would climb past 30 s on every
# existing install within half a minute and pin the inverter at 0 W forever.
from app.mqtt import SENSORS, MqttPublisher # noqa: E402
check("meter_source off is disabled", is_enabled({"meter_source": "off"}) is False)
check("a missing meter_source is disabled", is_enabled({}) is False)
check("an empty meter_source is disabled", is_enabled({"meter_source": ""}) is False)
check("ha_dsmr is enabled", is_enabled({"meter_source": SOURCE_HA}) is True)
check("mqtt_p1 is enabled", is_enabled({"meter_source": SOURCE_MQTT}) is True)
# The entity id SAFETY-01's firmware subscribes to, pinned by object_id.
row = [s for s in SENSORS if s[0] == "p1_age"]
check("the age sensor is declared exactly once", len(row) == 1)
check("its object_id pins entity_id to sensor.p1_sample_age_s",
row[0][1] == "p1_sample_age_s")
check("it is published in seconds", row[0][3] == "s")
class _RecordingClient:
def __init__(self):
self.sent = []
def publish(self, topic, payload=None, retain=False):
# Topic AND payload: object_id, the thing that actually pins the entity
# id, only appears in the discovery payload. Recording topics alone made
# the "is not announced" check pass for the wrong reason.
self.sent.append(f"{topic} {payload}")
def _announced(omit):
pub = MqttPublisher(None, 1883, omit=omit) # host None -> never connects
pub.client = _RecordingClient()
pub._announce()
return " ".join(pub.client.sent)
check("with P1 off the age sensor is never announced",
"p1_sample_age_s" not in _announced(("p1_age",)))
check("the other status entities are still announced with P1 off",
"goodwe_grid_power" in _announced(("p1_age",)))
check("with P1 on the age sensor IS announced",
"p1_sample_age_s" in _announced(()))
# And the publish dict itself, through the real Controller.
from app.main import Controller # noqa: E402
class _Store:
data = {}
def set(self, *a):
pass
def get_time(self, *a):
return None
class _Pub:
def __init__(self):
self.last = {}
def publish(self, values):
self.last = values
def close(self):
pass
pub_off = _Pub()
Controller({"meter_source": "off"}, None, _Store(), pub_off).publish()
check("with P1 off, p1_age is absent from the published payload",
"p1_age" not in pub_off.last)
check("...while the normal status keys are still published",
"setpoint" in pub_off.last and "grid" in pub_off.last)
pub_on = _Pub()
Controller({"meter_source": SOURCE_HA}, None, _Store(), pub_on).publish()
check("with P1 on, p1_age is published", "p1_age" in pub_on.last)
check("...as a number, so has_state() becomes true only once we feed it",
isinstance(pub_on.last["p1_age"], float))
# ⚠️ And on ha_signed identically - this is the whole reason TEL-04 exists. The
# age sensor is a hard prerequisite for the FW-01 flash, and it has to appear on
# the transport that can actually read the meter in this house.
pub_sig = _Pub()
Controller({"meter_source": SOURCE_HA_SIGNED}, None, _Store(), pub_sig).publish()
check("with ha_signed selected, p1_age is published too",
"p1_age" in pub_sig.last and isinstance(pub_sig.last["p1_age"], float))
print()
if fails:
print(f"{len(fails)} of {total} FAILED: {', '.join(fails)}")
sys.exit(1)
print(f"{total} checks")
print("all checks passed")