3 Commits
Author SHA1 Message Date
glenn schrooyenandClaude Opus 5 0a617c5482 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 22:16:34 +02:00
glenn schrooyenandClaude Opus 5 2097b7aaf6 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 22:16:34 +02:00
glenn schrooyenandClaude Opus 5 e46175559b SAFETY-04 review fixes: the freeze deadlocked, the bound was too loose
S-1. The frozen branch admitted a correction only if it shrank |i_w|. That is
unsatisfiable for BOTH signs of error whenever |correction| > 2*|i_w|, i.e.
whenever the integrator is near zero, so the loop stopped moving and the freeze
could never clear - it clears when the inverter tracks, and not tracking is
what saturation means. Measured: 0 W held into a 2 kW import indefinitely,
where release/1.0 recovers on the next cycle. Re-encoded as the same asymmetric
rule the output freeze has always used: may not wind further in the direction
it is already pushing, may fall, cross zero or reverse. Same interpretation,
an encoding that cannot deadlock.

S-2. integrator_max_w defaulted to 1.5x max_w, which ADDED windup: in
release/1.0 the accumulator was the post-clamp command and could never pass the
rail. Default is now "follow max_w" (config 0 = unset). Measured on the 4000 W
load-drop sim, first cycle after the drop: 1000 W at the new default, 1800 W at
3000. DOCS row inverted - the useful direction is below max_w, and the 14 768 W
anecdote is a vendor controller, not evidence about this code.

S-3. The claim that i_w=None preserved release/1.0 exactly was false, because
the S-1 gate ran regardless of seeding. It is true again, and now asserted
rather than asserted-about: 3024-case exhaustive comparison against a
transcription of the old law, over both freeze states, both signs and either
side of the deadband. Added the carried-i_w convergence/overshoot sim that the
shipped configuration was missing.

S-4. Cycles are distinct meter values, not seconds: cycle() runs only when the
meter reading changes, so the window has no wall-clock bound. Comment and DOCS
corrected; the stall is detection latency, not a windup hazard, because the
same condition stalls the whole loop.

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

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Du77usMj8XNKNFZGmUiWDa
2026-08-24 22:13:55 +02:00
8 changed files with 421 additions and 80 deletions
+32 -3
View File
@@ -60,13 +60,30 @@ that subtraction into the add-on, where it is done once and tested, and replaces
|---|---|---| |---|---|---|
| `meter_source` | `off` | `off` keeps `meter_entity`. `ha_dsmr` subscribes to the DSMR integration over the HA WebSocket; `mqtt_p1` reads a topic | | `meter_source` | `off` | `off` keeps `meter_entity`. `ha_dsmr` subscribes to the DSMR integration over the HA WebSocket; `mqtt_p1` reads a topic |
| `meter_phases` | 1 | 1 or 3. Must match the telegram, or every telegram is rejected and logged | | `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: grid power reads as *missing*, and the existing failsafe commands 0 W | | `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 | | `meter_mqtt_topic` | | `mqtt_p1` only |
| `p1_import_entity` | | The **unsigned** consumption sensor. Do not point this at a signed template | | `p1_import_entity` | | The **unsigned** consumption sensor. Do not point this at a signed template |
| `p1_export_entity` | | The **unsigned** injection sensor | | `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_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_phase_export_entities` | `[]` | L1..L3, in order |
#### 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 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 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 hardly at all, so a controller that failed over to it would be regulating
@@ -102,6 +119,18 @@ emits nothing, which is indistinguishable — to anything watching the value —
a meter that has died. Watching the age instead separates the two: it climbs when 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. 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` only.** The age measures *arrival*, not change. On the
> `ha_dsmr` path that is exactly right: a frozen meter emits no `state_changed`,
> so nothing arrives and the age climbs. On the MQTT path a bridge that is stuck
> republishing its last telegram keeps arriving, so the age stays near zero and a
> frozen meter still looks fresh. Detecting *that* needs a change-detector rather
> than an arrival-detector, and it is not in this version. Prefer `ha_dsmr` where
> both are available.
### Control ### Control
| option | default | meaning | | option | default | meaning |
@@ -114,9 +143,9 @@ telegrams stop and resets when they arrive, whatever the reading says.
| `step_w` | 10 | Quantisation | | `step_w` | 10 | Quantisation |
| `saturation_w` | 500 | Divergence that counts as "the inverter is at a limit" | | `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. **Do not set to 1** |
| `integrator_max_w` | 3000 | Bound on the loop's accumulator, separate from `max_w`. Caps how much stale error can be waiting to unwind when the sign flips. **Keep it above `max_w`, and do not set it equal to `max_w`** | | `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 | | `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) | | `auto_start` | false | Start controlling on boot (only after commissioning) |
#### Why `target_grid_w` is not zero #### Why `target_grid_w` is not zero
+67 -40
View File
@@ -28,16 +28,22 @@ class Tuning:
step_w: int = 10 step_w: int = 10
saturation_w: float = 500.0 saturation_w: float = 500.0
saturation_cycles: int = 3 saturation_cycles: int = 3
# ⚠️ The integrator's OWN bound, and deliberately not max_w. A commercial # The integrator's own bound. None means "follow max_w", which is the
# controller on this same site clamped only its output and still reported # default and the recommended setting.
# 14 768 W: with the inverter switched off its integrator climbed ~130 W #
# every 4 s past 10 kW while the output sat on the 5 kW rail, so the moment # ⚠️ DO NOT RAISE THIS ABOVE max_w without a measurement to justify it.
# the error flipped there were minutes of accumulated wind to burn off # Every watt of integrator above the rail is a watt of wind that has to be
# before the command moved at all. Bounding the accumulator is what makes # burned off before the command can start moving the other way, i.e. extra
# recovery time finite; bounding the output only hides it. # cycles of discharge into an already-exporting meter after every
# Headroom above max_w is wanted (a legitimate large error must not be # saturation event. Measured on the closed-loop sim, 4000 W load dropped to
# truncated at the rail), headroom without limit is the bug. # 0: at integrator_max_w == max_w the command is 1000 W two cycles later; at
integrator_max_w: float = 3000.0 # 1.5x max_w it is 1800 W. The output clamp already bounds what reaches the
# wire, so headroom here buys nothing but unwind latency.
#
# It is a separate key because it has to be able to be SMALLER than max_w,
# which is the only direction that buys anything: it caps unwind latency
# below what the rail implies. Merging it into max_w would take that away.
integrator_max_w: float | None = None
# What the meter should rest at, in W. Negative = a slight export. # What the meter should rest at, in W. Negative = a slight export.
# ⚠️ The deadband is a one-way ratchet: any resting point inside it holds # ⚠️ The deadband is a one-way ratchet: any resting point inside it holds
# forever, and the meter's IMPORT register counts every positive one with # forever, and the meter's IMPORT register counts every positive one with
@@ -54,9 +60,9 @@ class Decision:
sat_count: int sat_count: int
frozen: bool frozen: bool
reason: str reason: str
# The integrator AFTER this cycle, pre-clamp-to-max_w. Carry it back in as # The integrator AFTER this cycle, before the output clamp, the slew limit
# `i_w` next cycle; that is what keeps it a separate quantity from the # and quantisation. Carry it back in as `i_w` next cycle; that is what keeps
# command, which is the whole point of the bound above. # it a separate quantity from the command.
i_w: float = 0.0 i_w: float = 0.0
@@ -89,14 +95,19 @@ def compute(
# lets slew be larger than saturation_w. # lets slew be larger than saturation_w.
# #
# The spec states this window twice and differently: "> 10 s" (§11.2) and # The spec states this window twice and differently: "> 10 s" (§11.2) and
# "3 samples" (§10.3). Cycles are authoritative here because this function # "3 samples" (§10.3). This counts CYCLES, and a cycle is not a unit of
# has no clock - it is driven one cycle per meter update by run_control(), # time: run_control() calls cycle() only when the meter value CHANGES
# which only calls cycle() when the meter value changes. At the ~5 s # (`if self.grid != last_grid`), so three cycles is three distinct meter
# HomeWizard P1 cadence the default 3 cycles is ~15 s, i.e. the stricter # readings and nothing more. At the reference P1's ~5 s update rate that is
# reading of the two. On a faster meter it is not, so saturation_cycles is # usually ~15 s, but there is no upper bound on it - a meter that repeats a
# configurable and must be raised to keep the window over 10 s. # value stalls the counter.
# ponytail: a seconds-based window would mean plumbing wall-clock or dt #
# into a pure function whose whole value is that it has neither. # That is a detection-latency limit, not a windup hazard: the same
# condition that stalls the counter stalls the whole loop, so nothing
# accumulates in the meantime either. If a wall-clock window is ever
# required, it belongs in Controller (which has a clock) and not here.
# ponytail: this function is worth keeping clockless; the ceiling is that
# saturation_cycles cannot express a guaranteed number of seconds.
saturated_now = abs(prev_w - actual_w) > tuning.saturation_w saturated_now = abs(prev_w - actual_w) > tuning.saturation_w
sat_count = min(sat_count + 1, 10) if saturated_now else 0 sat_count = min(sat_count + 1, 10) if saturated_now else 0
frozen = sat_count >= tuning.saturation_cycles frozen = sat_count >= tuning.saturation_cycles
@@ -112,32 +123,48 @@ def compute(
# --- the integrator ---------------------------------------------------- # --- the integrator ----------------------------------------------------
# This loop is in velocity form: the accumulator IS the commanded power, so # This loop is in velocity form: the accumulator IS the commanded power, so
# for years "the integrator" and "the output" were one variable and could # "the integrator" and "the output" were one variable and could not be
# not be bounded apart. `i_w` is that accumulator made explicit. A caller # bounded apart. `i_w` is that accumulator made explicit; main.py carries it
# that passes nothing gets the old behaviour exactly - seeded from the last # between cycles, which is what turns the two clamps into two limits.
# command every cycle - and main.py carries it instead, which is what turns #
# the two clamps below into two independent limits. # Passing i_w=None re-seeds it from the last command every cycle. With
# integrator_max_w following max_w that reduces this function to the exact
# velocity form it replaced, frozen branch included - asserted by an
# exhaustive comparison against a transcription of the old law in
# test_control.py, not by inspection. Break either the gate or the bound
# below and that test is what tells you the equivalence went with it.
if i_w is None: if i_w is None:
i_w = float(prev_w) i_w = float(prev_w)
limit = tuning.max_w if tuning.integrator_max_w is None else tuning.integrator_max_w
if abs(error) < tuning.deadband_w: if abs(error) < tuning.deadband_w:
reason = "deadband" reason = "deadband"
else: else:
step_i = tuning.gain * error moved = i_w + tuning.gain * error
# ⚠️ Freeze means "may not wind FURTHER", not "may not move". A strict # ⚠️ Freeze means "may not wind FURTHER in the direction it is already
# freeze would strand the command at whatever it had reached until the # pushing". It may fall, cross zero, or reverse outright.
# inverter started tracking again - and the inverter is not tracking, #
# that is what saturation means, so nothing would ever release it. The # It must NOT be encoded as "only corrections that shrink |i_w|": that
# unwind direction is the escape route and stays open; the same rule is # is unsatisfiable for BOTH signs of error whenever the correction is
# applied again to the output below. # larger than twice the integrator, i.e. every time the integrator is
if not frozen or abs(i_w + step_i) < abs(i_w): # near zero. The loop then sits at its last value forever, because what
i_w = i_w + step_i # clears the freeze is the inverter tracking again and not-tracking is
# the definition of saturation. Measured on that encoding: 0 W held
# indefinitely into a 2 kW import, where this form recovers next cycle.
#
# This is the same asymmetric rule the output freeze uses below, which
# has been in service on real hardware. It is applied here as well
# because the requirement is that the INTEGRATOR stop accumulating, not
# only the command.
if not frozen:
i_w = moved
else:
i_w = min(moved, i_w) if i_w > 0 else max(moved, i_w)
# ⚠️ Applied EVERY cycle, frozen or not, and before the output clamp: the # ⚠️ Applied EVERY cycle, frozen or not: the freeze is conditional, this
# freeze is conditional, this bound is not. Order matters only in that the # bound is not. It is what makes the worst-case unwind time finite and
# command below is derived from the already-bounded integrator, so no # knowable instead of a function of how long the error happened to stand.
# accumulated value can reach the wire even once. i_w = max(-limit, min(limit, i_w))
i_w = max(-tuning.integrator_max_w, min(tuning.integrator_max_w, i_w))
want = i_w want = i_w
# ⚠️ Maintenance shaping (charge-only, cheap-window floor) used to live # ⚠️ Maintenance shaping (charge-only, cheap-window floor) used to live
+21 -6
View File
@@ -39,7 +39,7 @@ from .control import Tuning, compute, maintenance_charge_floor, peak_at_risk
from .hass import HomeAssistant from .hass import HomeAssistant
from .maintenance import IDLE, MaintConfig, Maintenance from .maintenance import IDLE, MaintConfig, Maintenance
from .mqtt import MqttPublisher from .mqtt import MqttPublisher
from .p1 import P1Ingest, build_source from .p1 import P1Ingest, build_source, is_enabled
from . import web from . import web
OPTIONS_PATH = "/data/options.json" OPTIONS_PATH = "/data/options.json"
@@ -71,7 +71,11 @@ class Controller:
step_w=int(opts.get("step_w", 10)), step_w=int(opts.get("step_w", 10)),
saturation_w=float(opts.get("saturation_w", 500)), saturation_w=float(opts.get("saturation_w", 500)),
saturation_cycles=int(opts.get("saturation_cycles", 3)), saturation_cycles=int(opts.get("saturation_cycles", 3)),
integrator_max_w=float(opts.get("integrator_max_w", 3000)), # 0 / unset means "follow max_w", which is the recommended
# value. Read the note in control.py before raising it above
# max_w: every watt above the rail is unwind latency.
integrator_max_w=(float(opts["integrator_max_w"])
if opts.get("integrator_max_w") else None),
) )
self.maint = Maintenance( self.maint = Maintenance(
MaintConfig( MaintConfig(
@@ -92,7 +96,7 @@ class Controller:
# until it opts in. # until it opts in.
self.p1 = P1Ingest(phases=int(opts.get("meter_phases", 1)), self.p1 = P1Ingest(phases=int(opts.get("meter_phases", 1)),
max_age_s=float(opts.get("meter_max_age_s", 30))) max_age_s=float(opts.get("meter_max_age_s", 30)))
self.p1_enabled = str(opts.get("meter_source", "off")) not in ("off", "") self.p1_enabled = is_enabled(opts)
# live state # live state
self.auto = bool(store.data.get("auto", opts.get("auto_start", False))) self.auto = bool(store.data.get("auto", opts.get("auto_start", False)))
@@ -320,16 +324,26 @@ class Controller:
await asyncio.sleep(1) await asyncio.sleep(1)
def publish(self) -> None: def publish(self) -> None:
self.mqtt.publish({ values = {
"setpoint": self.target, "setpoint": self.target,
"grid": self.grid, "grid": self.grid,
"battery": self.batt, "battery": self.batt,
"soc": self.soc, "soc": self.soc,
"phase": self.maint.phase, "phase": self.maint.phase,
"status": "running" if self.auto else "stopped", "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. # Recomputed here, once a second, on purpose - see P1Ingest.
"p1_age": round(self.p1.published_age_s, 1), values["p1_age"] = round(self.p1.published_age_s, 1)
}) self.mqtt.publish(values)
async def shutdown(self) -> None: async def shutdown(self) -> None:
"""Deterministic wind-down. Do not skip this.""" """Deterministic wind-down. Do not skip this."""
@@ -497,6 +511,7 @@ async def amain() -> None:
broker.get("port", 1883) if broker else 1883, broker.get("port", 1883) if broker else 1883,
broker.get("username") if broker else None, broker.get("username") if broker else None,
broker.get("password") 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 except Exception as err: # noqa: BLE001
_LOG.warning("MQTT unavailable (%s) - continuing without status entities", err) _LOG.warning("MQTT unavailable (%s) - continuing without status entities", err)
+8 -1
View File
@@ -60,7 +60,12 @@ AVAILABILITY = f"{BASE}/availability"
class MqttPublisher: 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.enabled = mqtt is not None and bool(host)
self.client = None self.client = None
if not self.enabled: if not self.enabled:
@@ -94,6 +99,8 @@ class MqttPublisher:
def _announce(self) -> None: def _announce(self) -> None:
for key, object_id, name, unit, dev_class, state_class, icon in SENSORS: for key, object_id, name, unit, dev_class, state_class, icon in SENSORS:
if key in self.omit:
continue
cfg = { cfg = {
"name": name, "name": name,
"object_id": object_id, "object_id": object_id,
+45 -9
View File
@@ -213,8 +213,16 @@ class QuarterAverager:
halves credited to the two blocks, never attributed wholly to either. halves credited to the two blocks, never attributed wholly to either.
""" """
def __init__(self, phases: int = 1): def __init__(self, phases: int = 1, max_hold_s: float = 30.0):
self.phases = phases 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._block: int | None = None # epoch seconds of the block start
self._acc = 0.0 # W*s of offtake in the open block self._acc = 0.0 # W*s of offtake in the open block
self._pp_acc = [0.0] * phases self._pp_acc = [0.0] * phases
@@ -273,16 +281,22 @@ class QuarterAverager:
return closed return closed
cursor = self._last_t 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: while True:
end = self._block + QUARTER_S end = self._block + QUARTER_S
stop = min(t, end) stop = min(t, end)
dt = stop - cursor covered = max(0.0, min(stop, hold_end) - cursor)
if dt > 0: if covered > 0:
self._acc += max(self._last_net, 0.0) * dt self._acc += max(self._last_net, 0.0) * covered
if self._last_pp is not None: if self._last_pp is not None:
for i, v in enumerate(self._last_pp[: self.phases]): for i, v in enumerate(self._last_pp[: self.phases]):
self._pp_acc[i] += max(v, 0.0) * dt self._pp_acc[i] += max(v, 0.0) * covered
self._elapsed += dt self._elapsed += covered
cursor = stop cursor = stop
if stop < end: if stop < end:
break break
@@ -321,7 +335,9 @@ class P1Ingest:
def __init__(self, phases: int = 1, max_age_s: float = 30.0): def __init__(self, phases: int = 1, max_age_s: float = 30.0):
self.phases = phases self.phases = phases
self.max_age_s = float(max_age_s) self.max_age_s = float(max_age_s)
self.averager = QuarterAverager(phases) # 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.blocks: list[QuarterBlock] = []
self.samples = 0 self.samples = 0
self.parse_errors = 0 self.parse_errors = 0
@@ -477,9 +493,17 @@ class HaDsmrSource:
continue continue
payload = json.loads(msg.data) payload = json.loads(msg.data)
if payload.get("id") == 2 and payload.get("type") == "result": 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 []: for obj in payload.get("result") or []:
self._absorb(obj.get("entity_id"), obj.get("state")) self._absorb(obj.get("entity_id"), obj.get("state"))
self._schedule()
elif payload.get("type") == "event": elif payload.get("type") == "event":
data = (payload.get("event") or {}).get("data") or {} data = (payload.get("event") or {}).get("data") or {}
if data.get("entity_id") not in self.ids: if data.get("entity_id") not in self.ids:
@@ -647,6 +671,18 @@ class MqttP1Source:
# --------------------------------------------------------------------------- # # --------------------------------------------------------------------------- #
# selection # 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): def build_source(opts: dict, ingest: P1Ingest, session, broker: dict | None):
"""Return the transport named by `meter_source`, or None if disabled. """Return the transport named by `meter_source`, or None if disabled.
@@ -654,7 +690,7 @@ def build_source(opts: dict, ingest: P1Ingest, session, broker: dict | None):
changing transport is a config edit, never a code path. changing transport is a config edit, never a code path.
""" """
source = str(opts.get("meter_source", "off") or "off").strip() source = str(opts.get("meter_source", "off") or "off").strip()
if source in ("off", ""): if not is_enabled(opts):
return None return None
if source == SOURCE_HA: if source == SOURCE_HA:
return HaDsmrSource(session, ingest, { return HaDsmrSource(session, ingest, {
+2 -2
View File
@@ -65,7 +65,7 @@ options:
step_w: 10 step_w: 10
saturation_w: 500 saturation_w: 500
saturation_cycles: 3 saturation_cycles: 3
integrator_max_w: 3000 integrator_max_w: 0
heartbeat_s: 10 heartbeat_s: 10
stale_input_s: 15 stale_input_s: 15
auto_start: false auto_start: false
@@ -121,7 +121,7 @@ schema:
step_w: int(1,100) step_w: int(1,100)
saturation_w: int(100,2000) saturation_w: int(100,2000)
saturation_cycles: int(1,10) saturation_cycles: int(1,10)
integrator_max_w: int(100,15000) integrator_max_w: int(0,15000)
heartbeat_s: int(2,25) heartbeat_s: int(2,25)
stale_input_s: int(5,120) stale_input_s: int(5,120)
auto_start: bool auto_start: bool
+101 -14
View File
@@ -78,7 +78,7 @@ print("SAFETY-04: the integrator is bounded apart from the output")
# The historical runaway, with its real numbers. A commercial controller on # The historical runaway, with its real numbers. A commercial controller on
# this site, with the inverter switched OFF, wound ~130 W every 4 s past 10 kW # this site, with the inverter switched OFF, wound ~130 W every 4 s past 10 kW
# and reported 14 768 W while its output clamp sat at 5 kW. At gain 0.6 that # and reported 14 768 W while its output clamp sat at 5 kW. At gain 0.6 that
# rate is a standing error of 130/0.6 217 W that never resolves, because the # rate is a standing error of 130/0.6 = 217 W that never resolves, because the
# inverter is not there to resolve it. 150 cycles is past the ~113 it took to # inverter is not there to resolve it. 150 cycles is past the ~113 it took to
# reach 14 768 W at that rate. # reach 14 768 W at that rate.
RUNAWAY_ERROR = 130.0 / 0.6 RUNAWAY_ERROR = 130.0 / 0.6
@@ -99,34 +99,53 @@ def runaway(tuning):
return worst_i, worst_cmd return worst_i, worst_cmd
TR = Tuning(max_w=2000, integrator_max_w=3000) TR = Tuning(max_w=2000) # integrator_max_w unset => follows max_w
wi, wc = runaway(TR) wi, wc = runaway(TR)
check(f"runaway: integrator plateaus at {wi:.0f} W (<= 3000)", wi <= TR.integrator_max_w) check(f"runaway: integrator plateaus at {wi:.0f} W (<= 2000)", wi <= TR.max_w)
check(f"runaway: emitted command peaks at {wc:.0f} W (<= 2000)", wc <= TR.max_w) check(f"runaway: emitted command peaks at {wc:.0f} W (<= 2000)", wc <= TR.max_w)
check("runaway: nowhere near the historical 14 768 W", wc < HISTORICAL_W / 4) check("runaway: nowhere near the historical 14 768 W", wc < HISTORICAL_W / 4)
# ...and with the saturation detector deliberately defeated, so that only the # ...and with the saturation detector deliberately defeated, so that only the
# clamp is holding. This is the AC that says the two mechanisms are # clamp is holding. Kill one mechanism, the other still bounds it.
# independent: kill one, the other still bounds it. TD = Tuning(max_w=2000, saturation_w=1e9)
TD = Tuning(max_w=2000, integrator_max_w=3000, saturation_w=1e9)
wi, wc = runaway(TD) wi, wc = runaway(TD)
check(f"runaway with the detector defeated: integrator still <= 3000 ({wi:.0f} W)", check(f"runaway with the detector defeated: integrator still bounded ({wi:.0f} W)",
wi <= TD.integrator_max_w) wi <= TD.max_w)
check("runaway with the detector defeated: command still <= max_w", wc <= TD.max_w) check("runaway with the detector defeated: command still <= max_w", wc <= TD.max_w)
# The bound is not max_w. If someone "simplifies" them into one key this fails. # 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, d = compute(prev_w=0, grid_w=6000, actual_w=0,
tuning=Tuning(max_w=2000, integrator_max_w=3000, slew_w=5000)) tuning=Tuning(max_w=2000, integrator_max_w=1000, slew_w=5000))
check("integrator bound is separate from the output clamp", check("integrator bound binds independently of the output clamp",
d.i_w == 3000 and d.target_w == 2000) d.i_w == 1000 and d.target_w == 1000)
# Freeze = does not accumulate. Same input twice; the integrator must not move. # Freeze = may not wind further in the direction it is already pushing.
TF = Tuning(saturation_w=500, saturation_cycles=3) TF = Tuning(saturation_w=500, saturation_cycles=3)
f1 = compute(prev_w=2000, grid_w=800, actual_w=0, tuning=TF, sat_count=3, i_w=2000.0) 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 accumulate", f1.i_w == 2000.0 and f1.frozen) 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) 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) check("frozen: unwinding is still allowed", f2.i_w < 2000.0)
# ⚠️ REGRESSION, and the reason the first cut of SAFETY-04 was rejected. A
# freeze encoded as "only corrections that shrink |i_w|" is unsatisfiable for
# BOTH signs of error whenever |correction| > 2*|i_w|, so near zero the loop
# stops moving forever - the freeze cannot clear, because clearing it needs the
# inverter to track and not-tracking is what saturation means. Measured on that
# encoding: 0 W held into a 2 kW import for as long as the sim ran.
z = compute(prev_w=0, grid_w=2000, actual_w=600, tuning=T, sat_count=3, i_w=0.0)
check("frozen at i_w=0: a 2 kW import still moves the command",
z.frozen and z.target_w == 1000)
# ...and the next cycle the inverter is inside saturation_w of the command, so
# the freeze clears on its own. Deadlock would show up here as frozen=True.
z2 = compute(prev_w=1000, grid_w=1000, actual_w=600, tuning=T,
sat_count=z.sat_count, i_w=z.i_w)
check("frozen at i_w=0: the freeze then clears", not z2.frozen)
# Same stranding on the other side: a small positive integrator against export.
z3 = compute(prev_w=100, grid_w=-1000, actual_w=800, tuning=T, sat_count=3, i_w=100.0)
check("frozen at i_w=+100: a 1 kW export still moves the command",
z3.frozen and z3.target_w < 0)
# False-positive guard: a normal 2 kW load step must not trip the detector, # 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. # because the plant needs several cycles to catch up on every one of them.
prev, actual, sat, i_w, froze = 0.0, 0.0, 0, 0.0, False prev, actual, sat, i_w, froze = 0.0, 0.0, 0, 0.0, False
@@ -137,6 +156,74 @@ for _ in range(12):
froze = froze or d.frozen froze = froze or d.frozen
check("a normal 2 kW load step does not trip the saturation freeze", not froze) check("a normal 2 kW load step does not trip the saturation freeze", not froze)
# The convergence sim below runs WITHOUT a carried integrator. This is the same
# 2 kW step in the configuration that actually ships, where main.py carries it.
prev, actual, sat, i_w = 0.0, 0.0, 0, 0.0
carried = 0
for _ in range(12):
d = compute(prev, 2000.0 - actual, actual, T, sat, i_w)
prev, sat, i_w = d.target_w, d.sat_count, d.i_w
actual = actual + 0.94 * (prev - actual)
carried += 1
if abs(2000.0 - actual) < T.deadband_w:
break
check(f"carried integrator converges in {carried} cycles (<=6)", carried <= 6)
check("carried integrator does not overshoot the load", actual <= 2000.0 + T.deadband_w)
# ⚠️ REGRESSION: an integrator allowed to wind past the rail buys nothing (the
# output clamp already bounds the wire) and costs extra cycles of
# wrong-direction power after every saturation event. 4000 W load held to
# saturation, then dropped to 0; the figure is the command on the first cycle
# after the drop. This is what makes the DOCS advice checkable.
def unwind(t):
prev, actual, sat, i_w, load = 0.0, 0.0, 0, 0.0, 4000.0
for c in range(16):
if c == 15:
load = 0.0
d = compute(prev, load - actual, actual, t, sat, i_w)
prev, sat, i_w = d.target_w, d.sat_count, d.i_w
actual = actual + 0.94 * (prev - actual)
return prev
tight, loose = unwind(Tuning(max_w=2000)), unwind(Tuning(max_w=2000, integrator_max_w=3000))
check(f"after saturation ends the command is {tight:.0f} W (<= 1000)", tight <= 1000)
check(f"headroom above max_w makes that worse ({loose:.0f} W) - hence the default",
loose > tight)
print("SAFETY-04: the i_w=None path is still release/1.0, exactly")
def legacy(prev, grid, actual, t, sat_count):
"""release/1.0's control law, transcribed. Do not 'improve' this."""
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
target = max(-t.max_w, min(t.max_w, want))
target = max(prev - t.slew_w, min(prev + t.slew_w, target))
if frozen:
target = min(target, prev) if prev > 0 else max(target, prev)
step = max(1, int(t.step_w))
return float(round(target / step) * step), sc
# 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.
diffs = []
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"
+ (f" (first diff {diffs[0]})" if diffs else ""), not diffs)
print("capacity tariff") print("capacity tariff")
check("no forecast means no cap", maintenance_charge_floor(2500, None, 3500) == 2500) 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) check("headroom caps the charge", maintenance_charge_floor(2500, 2000, 3500) == 1500)
+145 -5
View File
@@ -229,6 +229,47 @@ before = a.partial_ws
a.add(sample(1000.0, at=BASE + timedelta(seconds=5))) a.add(sample(1000.0, at=BASE + timedelta(seconds=5)))
check("an out-of-order telegram is dropped, not integrated backwards", check("an out-of-order telegram is dropped, not integrated backwards",
a.partial_ws == before and a.elapsed_s == 10.0) 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") print("ingest timestamp, age and staleness")
@@ -482,9 +523,15 @@ async def _e2e():
live, wire = asyncio.run(_e2e()) live, wire = asyncio.run(_e2e())
check("the websocket handshake and subscription complete", live.samples >= 1) check("the websocket handshake and subscription complete", live.samples >= 1)
# Six state_changed events arrived (two per telegram). The debounce is what # ⚠️ TWO, not three. get_states primes the cache but must NOT build a sample:
# makes that three consistent samples instead of six half-updated ones. # HA returns whatever it currently holds, which after a Core restart is a
check("three telegrams produce three samples, not six", live.samples == 3) # 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", check("the final export-dominant telegram nets negative",
live.last.net_w == -800.0) live.last.net_w == -800.0)
check("the sample was built over the wire, tagged with its transport", check("the sample was built over the wire, tagged with its transport",
@@ -492,9 +539,102 @@ check("the sample was built over the wire, tagged with its transport",
check("an entity we did not subscribe to is never cached", check("an entity we did not subscribe to is never cached",
"sensor.something_else" not in wire.cache and len(wire.cache) == 1) "sensor.something_else" not in wire.cache and len(wire.cache) == 1)
check("a mid-stream unavailable is a parse error, not a sample", check("a mid-stream unavailable is a parse error, not a sample",
live.parse_errors == 1 and live.samples == 3) live.parse_errors == 1 and live.samples == 2)
check("the last good reading survives the unavailable", live.net_w == -800.0) 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.5) 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("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.p1 import is_enabled # noqa: E402
from app.mqtt import SENSORS, MqttPublisher # noqa: E402
check("meter_source off is disabled", is_enabled({"meter_source": "off"}) is False)
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))
print() print()
if fails: if fails: