"""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 json
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, HomeWizardLocalSource,
    QuarterAverager,
    SOURCE_HA, SOURCE_HA_SIGNED, SOURCE_HOMEWIZARD, SOURCE_MQTT,
    build_source, is_enabled, make_sample, parse_homewizard, parse_mqtt_payload,
    split_signed,
)

fails = []
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("homewizard_local: parsing the meter's own /api/v1/data document")
# The same JSON `sim/hwsim.py` serves and the same JSON the real meter at
# 192.168.2.250 serves. One SIGNED connection figure plus three signed legs, so
# it goes through the same split_signed as ha_signed - the sign convention is
# not re-derived on this transport, it is reused.


def hw_doc(w, l1=None, l2=None, l3=None, imp_kwh=1234.567, exp_kwh=890.123):
    return {"wifi_ssid": "sim", "smr_version": 50, "meter_model": "SIM-P1",
            "total_power_import_kwh": imp_kwh, "total_power_export_kwh": exp_kwh,
            "active_power_w": w,
            "active_power_l1_w": l1, "active_power_l2_w": l2,
            "active_power_l3_w": l3, "total_gas_m3": 0.0}


s = parse_homewizard(hw_doc(775.0, l1=775.0), 1)
check("the overnight base load (+775 W) is import on this transport too",
      s.net_w == 775.0 and s.import_w == 775.0 and s.export_w == 0.0)
check("the sample is tagged homewizard_local", s.source == SOURCE_HOMEWIZARD)
s = parse_homewizard(hw_doc(-5710.0, l1=-5710.0), 1)
check("the midday PV peak (-5710 W) is export, not a 5.7 kW draw",
      s.net_w == -5710.0 and s.export_w == 5710.0)
check("a single-phase document still yields its one leg", s.per_phase_w == (-5710.0,))

# The TEL-04 three-phase survey reading, served the HomeWizard way.
s = parse_homewizard(hw_doc(187.0, 2301.0, 468.0, -2582.0), 3)
check("signed legs keep the exporting phase negative",
      s.per_phase_w == (2301.0, 468.0, -2582.0))
check("per-phase import clamps the exporting leg out of the billed figure",
      s.per_phase_import_w == (2301.0, 468.0, 0.0))
check("the legs sum to 2769 W while the connection nets 187 W",
      sum(s.per_phase_import_w) == 2769.0 and s.net_w == 187.0)

# ⚠️ A single-phase HomeWizard serves `active_power_l2_w: null`. Unlike the HA
# transports there is no "wait for the rest" case - every field came out of ONE
# response, so a missing leg cannot be a STALE leg. Dropping the whole sample
# over it would turn a healthy meter into a climbing age, which is the exact
# false trip this transport exists to remove.
def parsed(doc, phases=1):
    """parse_homewizard(), with an escaping exception turned into a visible value.

    Same reason as `built()` above: if the "all legs or none" guard goes missing,
    parsing raises, and without this the suite aborts on a traceback instead of
    reddening the NAMED check that says which rule died.
    """
    try:
        return parse_homewizard(doc, phases)
    except Exception as err:  # noqa: BLE001 - a raise here is itself the failure
        print(f"       parse_homewizard raised {type(err).__name__}: {err}")
        return err


s = parsed(hw_doc(500.0, l1=500.0), 3)
check("a meter serving fewer legs than meter_phases still gives a reading",
      getattr(s, "net_w", None) == 500.0 and getattr(s, "per_phase_w", "?") is None)
check("...and does not fabricate a partial phase tuple",
      getattr(s, "per_phase_import_w", "?") is None)

raises("a response that is not a JSON object is rejected",
       lambda: parse_homewizard([1, 2, 3], 1))
raises("a document with no active_power_w is rejected, not read as 0 W",
       lambda: parse_homewizard({"total_gas_m3": 0.0}, 1))
raises("a null active_power_w is rejected", lambda: parse_homewizard(hw_doc(None), 1))
raises("a string active_power_w is rejected", lambda: parse_homewizard(hw_doc("775"), 1))
raises("a NaN active_power_w is rejected",
       lambda: parse_homewizard(hw_doc(float("nan")), 1))
raises("the 64954 signed-decode contamination is refused here too",
       lambda: parse_homewizard(hw_doc(64954.0), 1))
raises("a non-numeric leg is rejected rather than quietly dropped",
       lambda: parse_homewizard(hw_doc(600.0, "2000", 0.0, 100.0), 3))

# --------------------------------------------------------------------------- #
print("homewizard_local: every HTTP response is an arrival")
# THE WHOLE TICKET. sensor.p1_sample_age_s has to mean "time since the meter
# REPORTED", not "time since the value CHANGED". Home Assistant cannot express
# the first: a repeated reading emits no state_changed, advances last_reported
# on neither serialiser, and state_reported cannot be subscribed to at all
# ("Event filter is required"). Measured on the ENV-01 rig over 70 s of a frozen
# meter and again against the live house over ten repeated readings. Our own
# capture of this house's meter goes 42.2 s and 97.0 s between changes, both past
# the default meter_max_age_s of 30 - i.e. the age sensor would command 0 W on a
# perfectly healthy meter. Polling the meter itself removes that: the response is
# the arrival, and the number in it is not consulted.


class _FakeMeter:
    """hwsim in eight lines: serves whatever document it is told to, or a fault."""

    def __init__(self, doc=None):
        self.doc = doc
        self.status = 200
        self.body = None          # set to raw text to serve something unparseable
        self.delay = 0.0          # set to seconds to imitate a meter that hangs
        self.hits = 0

    async def handle(self, request):
        from aiohttp import web
        self.hits += 1
        if self.delay:
            await asyncio.sleep(self.delay)
        # ⚠️ web.Response(text=...) defaults to text/plain. That is not just the
        # "unparseable body" path: fed VALID json it serves a good document under
        # the wrong mimetype, which is the only way to reach the content_type
        # guard in poll_once - json_response can never produce it.
        if self.body is not None:
            return web.Response(text=self.body, status=self.status)
        return web.json_response(self.doc, status=self.status)


async def _hw_rig(fn):
    """Run `fn(make_source, meter)` against a real HTTP server on localhost."""
    from aiohttp import web
    meter = _FakeMeter(hw_doc(350.0, l1=350.0))
    app = web.Application()
    app.router.add_get("/api/v1/data", meter.handle)
    runner = web.AppRunner(app)
    await runner.setup()
    site = web.TCPSite(runner, "127.0.0.1", 0)
    await site.start()
    port = site._server.sockets[0].getsockname()[1]
    async with aiohttp.ClientSession() as sess:
        try:
            return await fn(sess, port, meter)
        finally:
            await runner.cleanup()


async def _arrivals(sess, port, meter):
    ing = P1Ingest(phases=1, max_age_s=30.0)
    src = HomeWizardLocalSource(sess, ing, "127.0.0.1", port=port, poll_s=1.0)
    out = {}
    out["first"] = await src.poll_once()
    out["stamp1"] = ing.last.ingest_mono
    # Four more polls of the IDENTICAL document - the meter has not moved a watt.
    for _ in range(4):
        await asyncio.sleep(0.05)
        await src.poll_once()
    out["ing"], out["src"], out["meter"] = ing, src, meter
    return out


r = asyncio.run(_hw_rig(_arrivals))
ing, src = r["ing"], r["src"]
check("a poll of the meter's own API builds a sample",
      r["first"] is True and ing.net_w == 350.0)
# ⚠️ THE acceptance criterion. Five identical readings, five arrivals. On
# ha_signed this whole sequence produces exactly ONE state_changed and then
# silence, and the age climbs to 30 s on a meter that is answering perfectly.
check("five identical readings are five arrivals, not one",
      ing.samples == 5 and r["meter"].hits == 5 and src.polls == 5)
check("an unchanged value still stamps a NEW arrival time",
      ing.last.ingest_mono > r["stamp1"])
check("...so the age resets on a response carrying an unchanged number",
      ing.published_age_s < 1.0 and ing.stale is False)
check("no arrival was ever a parse error", ing.parse_errors == 0)

# --------------------------------------------------------------------------- #
print("homewizard_local: frozen vs steady, the pair HA cannot separate")
# ⚠️ Read this before changing anything here. A frozen meter (hwsim --fault
# freeze) answers 200 OK forever with a stale document. It is ARRIVING. So the
# age - an arrival detector, correctly - reads fresh on both, and that is not a
# defect in the age, it is the definition of the signal. What the local API adds
# that Home Assistant never had is the ENERGY REGISTERS: a meter under real load
# advances total_power_import_kwh (1 Wh resolution, ~10 s at 350 W) even when the
# power figure repeats, and a frozen one does not. That is the discriminator, and
# it is exposed as `unchanged_s` - deliberately NOT folded into the age, because
# this controller regulates grid power toward ~0 W and at a converged -10 W the
# export register needs six minutes to move. Thresholding unchanged_s at 30 s
# would rebuild the false-trip limit cycle at the exact operating point we aim
# for. See DOCS.md and HomeWizardLocalSource.unchanged_s.


async def _freeze_vs_steady(sess, port, meter):
    ing_f = P1Ingest(phases=1, max_age_s=30.0)
    frozen = HomeWizardLocalSource(sess, ing_f, "127.0.0.1", port=port, poll_s=1.0)
    meter.doc = hw_doc(350.0, l1=350.0)          # --fault freeze: never moves
    for i in range(4):
        if i:
            await asyncio.sleep(0.1)             # sleep BEFORE, so unchanged_s
        await frozen.poll_once()                 # is read the instant a poll lands

    ing_s = P1Ingest(phases=1, max_age_s=30.0)
    steady = HomeWizardLocalSource(sess, ing_s, "127.0.0.1", port=port, poll_s=1.0)
    kwh = 1234.567
    for i in range(4):
        if i:
            await asyncio.sleep(0.1)
        # A steady 350 W house: the power figure repeats, the register climbs.
        kwh += 0.001
        meter.doc = hw_doc(350.0, l1=350.0, imp_kwh=round(kwh, 3))
        await steady.poll_once()
    return (ing_f, frozen), (ing_s, steady)


(ing_f, frozen), (ing_s, steady) = asyncio.run(_hw_rig(_freeze_vs_steady))
check("a frozen meter keeps arriving, so both read the same power",
      ing_f.net_w == 350.0 and ing_s.net_w == 350.0)
# ⚠️ Recorded as a rule, not a shortcoming: the age is an ARRIVAL detector and a
# frozen meter genuinely is arriving. Anyone tempted to make the age catch freeze
# is about to reintroduce the false trip on a steady house.
check("the age cannot separate them, and is fresh on both",
      ing_f.published_age_s < 1.0 and ing_s.published_age_s < 1.0)
check("the frozen meter's measurement stands still", frozen.unchanged_s > 0.25)
check("...while the steady meter's energy register keeps advancing",
      steady.unchanged_s < 0.05)
check("so the two ARE separable on the local API, which HA could not do",
      frozen.unchanged_s > steady.unchanged_s * 3)
check("unchanged_s is None before the first response ever lands",
      HomeWizardLocalSource(None, P1Ingest(), "h").unchanged_s is None)

# --------------------------------------------------------------------------- #
print("homewizard_local: a failed poll is a missing reading, never 0 W")


async def _failures(sess, port, meter):
    ing = P1Ingest(phases=1, max_age_s=30.0)
    src = HomeWizardLocalSource(sess, ing, "127.0.0.1", port=port, poll_s=1.0)
    await src.poll_once()
    # Age the good sample by hand so a reset would be unmistakable.
    ing.submit(make_sample(SOURCE_HOMEWIZARD, 350.0, 0.0, phases=1,
                           ingest_mono=time.monotonic() - 90.0))
    out = {}
    meter.status = 500
    out["http500"] = await src.poll_once()
    out["conn_after_500"] = src.connected
    meter.status, meter.body = 200, "<html>gateway</html>"
    out["notjson"] = await src.poll_once()
    meter.body = None
    meter.doc = {"total_gas_m3": 0.0}            # 200 OK, no power in it
    out["nopower"] = await src.poll_once()
    out["ing"], out["src"] = ing, src

    # And a meter that is not listening at all: `--fault down`.
    dead = HomeWizardLocalSource(sess, P1Ingest(phases=1, max_age_s=30.0),
                                 "127.0.0.1", port=1, poll_s=1.0)
    out["down"] = await dead.poll_once()
    out["dead"] = dead
    return out


r = asyncio.run(_hw_rig(_failures))
ing, src = r["ing"], r["src"]
check("an HTTP 500 is not a reading", r["http500"] is False)
# A source that was connected and then failed must SAY it is disconnected -
# otherwise the diagnostic reads healthy while the age climbs, which is exactly
# the "connected: True, parse_errors: 0, samples: 0" state the rig recorded.
check("a poll that fails clears the connected flag", r["conn_after_500"] is False)
check("a 200 OK carrying something that is not JSON is not a reading",
      r["notjson"] is False)
check("a 200 OK with no active_power_w in it is not a reading", r["nopower"] is False)
check("a meter that refuses the connection is not a reading",
      r["down"] is False and r["dead"].connected is False)
check("every failure was counted as a parse error", ing.parse_errors == 3)
check("...and the transport records the transport-level ones separately",
      src.poll_errors == 2 and src.polls == 4)
# ⚠️ The rule the safety chain rests on. A failed poll must not manufacture a
# balanced house, and must not reset the clock the watchdog reads.
check("a failed poll does not reset the age", ing.published_age_s > 89)
check("a failed poll leaves the last good value in place, and it is not 0 W",
      ing.last.net_w == 350.0 and ing.stale is True and ing.net_w is None)
check("no failed poll ever became a sample", ing.samples == 2)
check("a 200 OK marks the transport connected even when its body is refused",
      src.connected is True)

# --------------------------------------------------------------------------- #
print("homewizard_local: a wrong header, a hung meter, a submit that throws")
# Three failure shapes that all end the same way if they are mishandled - no
# sample, a climbing age, the battery at 0 W - and each of which would send the
# operator hunting the wrong device.


async def _header_and_timeout(sess, port, meter):
    out = {}
    # Valid JSON under text/plain: exactly what content_type=None is for.
    meter.body = json.dumps(hw_doc(350.0, l1=350.0))
    ing = P1Ingest(phases=1, max_age_s=30.0)
    src = HomeWizardLocalSource(sess, ing, "127.0.0.1", port=port, poll_s=1.0)
    out["mimetype"] = await src.poll_once()
    out["ing"] = ing
    meter.body = None

    # A meter that takes the connection and then does not answer.
    meter.delay = 0.6
    ing_t = P1Ingest(phases=1, max_age_s=30.0)
    slow = HomeWizardLocalSource(sess, ing_t, "127.0.0.1", port=port,
                                 poll_s=1.0, timeout_s=0.05)
    out["timeout"] = await slow.poll_once()
    out["timeout_err"] = ing_t.last_error
    meter.delay = 0.0
    return out


r = asyncio.run(_hw_rig(_header_and_timeout))
# ⚠️ A real firmware answering text/plain must not read as a dead meter. Drop
# content_type=None from poll_once and this goes red with "unexpected mimetype".
check("valid JSON under the wrong Content-Type is still a reading",
      r["mimetype"] is True and r["ing"].samples == 1 and r["ing"].net_w == 350.0)
# ⚠️ str(asyncio.TimeoutError()) is the EMPTY STRING. Without the class-name
# fallback the status page reads "last error:" and then nothing, on a hung
# meter, at the moment someone is reading that line to find out why the battery
# went to 0 W.
check("a timed-out poll names the fault instead of logging an empty reason",
      r["timeout"] is False and "TimeoutError" in (r["timeout_err"] or ""))


async def _submit_rejects(sess, port, meter):
    """submit() raising P1Error must be a rejection, not an escaping exception."""
    class _P1Boom(P1Ingest):
        def submit(self, sample):
            raise P1Error("register went backwards")

    ing = _P1Boom(phases=1, max_age_s=30.0)
    src = HomeWizardLocalSource(sess, ing, "127.0.0.1", port=port, poll_s=1.0)
    try:
        ok = await src.poll_once()
    except Exception as err:      # noqa: BLE001 - an escape IS the failure
        ok = err
    return ok, ing.parse_errors


ok, errs = asyncio.run(_hw_rig(_submit_rejects))
check("a submit that rejects the sample is handled, not left to escape",
      ok is False and errs == 1)


async def _submit_explodes(sess, port, meter):
    """And an UNEXPECTED raise must not kill the poll task for good."""
    class _Boom(P1Ingest):
        def submit(self, sample):
            raise RuntimeError("kaboom")

    src = HomeWizardLocalSource(sess, _Boom(phases=1, max_age_s=30.0),
                                "127.0.0.1", port=port, poll_s=0.05)
    task = asyncio.create_task(src.run())
    await asyncio.sleep(0.3)
    alive = not task.done()
    task.cancel()
    # ⚠️ BaseException, and the same reason built() exists: if run() loses its
    # guard the task is already dead HOLDING the RuntimeError, and awaiting it
    # re-raises - aborting the whole suite with a traceback instead of reddening
    # the check that names the rule. The death is what `alive` records; catching
    # it here is bookkeeping, not leniency.
    try:
        await task
    except BaseException:  # noqa: BLE001
        pass
    return alive, src.polls


# ⚠️ The silent-death case. A dead poll task fails SAFE - the age climbs and the
# controller commands 0 W - but it looks exactly like a dead meter, so the
# operator spends the outage power-cycling hardware that was never at fault.
alive, polls = asyncio.run(_hw_rig(_submit_explodes))
check("a raise inside a poll does not permanently kill the polling task",
      alive is True and polls > 1)

# --------------------------------------------------------------------------- #
print("homewizard_local: selection by config")

check("homewizard_local is enabled", is_enabled({"meter_source": SOURCE_HOMEWIZARD}) is True)
sel = build_source({"meter_source": SOURCE_HOMEWIZARD, "p1_host": "192.168.2.250"},
                   P1Ingest(), None, None)
check("meter_source homewizard_local selects the polling transport",
      isinstance(sel, HomeWizardLocalSource))
# ⚠️ getattr, not attribute access, for the same reason `built()` exists: a
# startup guard that goes wrong returns None here, and `None.url` would abort the
# suite with a traceback instead of reddening the check that names the rule.
check("...pointed at the meter's own local API on the default port",
      getattr(sel, "url", None) == "http://192.168.2.250:80/api/v1/data")
check("the default cadence is the meter's own ~5 s update rate",
      getattr(sel, "poll_s", None) == 5.0)
# ⚠️ The request must not outlive the poll interval: a backlog of queued requests
# behind a slow meter would land several arrivals for one measurement.
check("the request timeout is held under the poll interval",
      getattr(sel, "timeout_s", 99) < getattr(sel, "poll_s", 0))

sel = build_source({"meter_source": SOURCE_HOMEWIZARD, "p1_host": "10.0.0.9:8080"},
                   P1Ingest(), None, None)
check("an explicit host:port is honoured, which is how the sim is reached",
      getattr(sel, "url", None) == "http://10.0.0.9:8080/api/v1/data")
sel = build_source({"meter_source": SOURCE_HOMEWIZARD,
                    "p1_host": "http://192.168.2.250/"}, P1Ingest(), None, None)
check("a pasted browser URL still resolves to the right host",
      getattr(sel, "url", None) == "http://192.168.2.250:80/api/v1/data")

# ⚠️ Caught once at startup, not once per poll - a source wired up wrong
# otherwise fails in the one way indistinguishable from a healthy one nobody has
# polled yet: no samples, a climbing age, the watchdog at 0 W, nothing in the log.
check("a blank p1_host is refused rather than silently never polling",
      build_source({"meter_source": SOURCE_HOMEWIZARD, "p1_host": ""},
                   P1Ingest(), None, None) is None)
check("...and whitespace does not sneak past it",
      build_source({"meter_source": SOURCE_HOMEWIZARD, "p1_host": "   "},
                   P1Ingest(), None, None) is None)
check("an unparseable port is refused rather than guessed",
      build_source({"meter_source": SOURCE_HOMEWIZARD, "p1_host": "meter:eighty"},
                   P1Ingest(), None, None) is None)
# ⚠️ The age can never be fresher than the poll interval, so a cadence at or past
# max_age_s means every reading is stale before its successor arrives: the
# controller would sit permanently on missing inputs while the meter is fine.
check("a poll cadence at meter_max_age_s is refused",
      build_source({"meter_source": SOURCE_HOMEWIZARD, "p1_host": "m",
                    "meter_poll_s": 30}, P1Ingest(max_age_s=30), None, None) is None)
check("...and one past it too",
      build_source({"meter_source": SOURCE_HOMEWIZARD, "p1_host": "m",
                    "meter_poll_s": 45}, P1Ingest(max_age_s=30), None, None) is None)
check("a cadence with headroom is accepted",
      isinstance(build_source({"meter_source": SOURCE_HOMEWIZARD, "p1_host": "m",
                               "meter_poll_s": 5}, P1Ingest(max_age_s=30),
                              None, None), HomeWizardLocalSource))
check("a zero cadence is refused rather than spinning",
      build_source({"meter_source": SOURCE_HOMEWIZARD, "p1_host": "m",
                    "meter_poll_s": 0}, P1Ingest(max_age_s=30), None, None) is None)
check("a non-numeric cadence is refused",
      build_source({"meter_source": SOURCE_HOMEWIZARD, "p1_host": "m",
                    "meter_poll_s": "fast"}, P1Ingest(max_age_s=30), None, None) is None)
# The four modes must not bleed into each other.
check("ha_signed ignores p1_host and still selects the entity transport",
      type(build_source({"meter_source": SOURCE_HA_SIGNED, "p1_net_entity": "sensor.n",
                         "p1_host": "192.168.2.250"}, P1Ingest(), None, None))
      is HaSignedSource)
check("homewizard_local ignores p1_net_entity and needs its own host",
      build_source({"meter_source": SOURCE_HOMEWIZARD,
                    "p1_net_entity": "sensor.n"}, P1Ingest(), None, None) is None)

# --------------------------------------------------------------------------- #
print("the age sensor must not exist when P1 is off")
# ⚠️ This is a fleet-wide regression guard, not a nicety. The ESP32 watchdog
# 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)
check("homewizard_local is enabled here too",
      is_enabled({"meter_source": SOURCE_HOMEWIZARD}) is True)

# The entity id SAFETY-01's firmware subscribes to, pinned by object_id.
row = [s for s in SENSORS if s[0] == "p1_age"]
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))

# ⚠️ And on homewizard_local, where it is the one age FW-01 may actually
# threshold - every other transport's age measures when the VALUE changed.
pub_hw = _Pub()
Controller({"meter_source": SOURCE_HOMEWIZARD}, None, _Store(), pub_hw).publish()
check("with homewizard_local selected, p1_age is published",
      "p1_age" in pub_hw.last and isinstance(pub_hw.last["p1_age"], float))

# --------------------------------------------------------------------------- #
print("unchanged_s has somewhere an operator can read it")
# ⚠️ The counter is deliberately NOT thresholded and NOT folded into the age -
# at the converged -10 W this controller aims for, a 1 Wh register needs six
# minutes to move, so any limit false-trips at the target operating point. That
# refusal only holds up if a HUMAN can interpret the number instead, and DOCS.md
# tells them "the transport tracks it as unchanged_s". Before this, nothing ever
# read the transport object back: connected, polls, poll_errors and unchanged_s
# were all write-only, and the documented signal existed nowhere an operator
# could see it.


def p1_rows(ctl):
    """The P1 status line(s), or the exception that stopped checks() making one.

    Same reason as built(): a status line that raises would abort the suite with
    a traceback instead of reddening the check that names the rule.
    """
    try:
        return [c["text"] for c in ctl.checks() if c["text"].startswith("P1 meter")]
    except Exception as err:  # noqa: BLE001 - a raise here is itself the failure
        print(f"       checks() raised {type(err).__name__}: {err}")
        return err


def _fed(source):
    ctl = Controller({"meter_source": source}, None, _Store(), _Pub())
    ctl.p1.submit(make_sample(source, 350.0, 0.0, phases=1,
                              ingest_mono=time.monotonic()))
    return ctl


hw_ctl = _fed(SOURCE_HOMEWIZARD)
hw_src = HomeWizardLocalSource(None, hw_ctl.p1, "127.0.0.1")
hw_src._note(hw_doc(350.0, l1=350.0))
hw_ctl.p1_source = hw_src            # what amain() does once the transport exists
rows = p1_rows(hw_ctl)
check("the healthy P1 status line reports the transport's unchanged_s",
      isinstance(rows, list) and len(rows) == 1 and "unchanged" in rows[0])

# ⚠️ getattr, not attribute access: p1_source is None until amain() builds one,
# and ha_dsmr/ha_signed/mqtt_p1 have no such counter at all. Reaching for it
# directly would turn the whole status page into a 500 on every other transport.
ha_rows = p1_rows(_fed(SOURCE_HA))
check("...and the line is unharmed on a transport that has no such counter",
      isinstance(ha_rows, list) and len(ha_rows) == 1 and "unchanged" not in ha_rows[0])

print()
if fails:
    print(f"{len(fails)} of {total} FAILED: {', '.join(fails)}")
    sys.exit(1)
print(f"{total} checks")
print("all checks passed")