admin/goodwe-addon
1
0
Fork 0
Code Issues Pull Requests 4 Actions Packages Projects Releases Wiki Activity

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

Browse Source 
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
This commit is contained in:
glenn schrooyen
2026-08-24 22:16:34 +02:00
co-authored by Claude Opus 5
parent e46175559b
commit 2097b7aaf6
6 changed files with 1331 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files
goodwe_controller/test_p1.py
+504
View File
@@ -0,0 +1,504 @@
"""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, QuarterAverager, SOURCE_HA, SOURCE_MQTT,
make_sample, parse_mqtt_payload,
)
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 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)
# --------------------------------------------------------------------------- #
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)
# Six state_changed events arrived (two per telegram). The debounce is what
# makes that three consistent samples instead of six half-updated ones.
check("three telegrams produce three samples, not six", live.samples == 3)
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 == 3)
check("the last good reading survives the unavailable", live.net_w == -800.0)
check("the averager integrated the live stream", live.averager.elapsed_s > 0.5)
print()
if fails:
print(f"{len(fails)} of {total} FAILED: {', '.join(fails)}")
sys.exit(1)
print(f"{total} checks")
print("all checks passed")