goodwe-addon/goodwe_controller/app/p1.py

"""P1 meter ingestion - the only authoritative measurement of real grid exchange.

Everything downstream trusts this module: the safety checks, the capacity-tariff
peak, the optimizer, the control loop's sign. So three things happen here and
nowhere else.

  1. The IMPORT/EXPORT DERIVATION. A Belgian P1 read over DSMR exposes two
     UNSIGNED registers - consumption and injection. Net power is
     `import_w - export_w`, positive = import, and that subtraction is done
     exactly once, here (spec §5.2: "the derivation is the EMS's job, not a
     template the user has to write"). A second copy of it somewhere else is a
     second chance to invert the control loop.

     Some P1 readers - the HomeWizard P1 among them - publish the OTHER shape:
     one SIGNED figure, positive = import, and no unsigned registers at all.
     `split_signed()` fans that back out into the same two magnitudes, so there
     is still exactly one internal representation and one sign convention. ⚠️ It
     lives here, next to the subtraction, for the same reason the subtraction
     does: the moment a user is asked to write two template sensors that split a
     signed value, the sign convention is back in unreviewed YAML underneath a
     safety input, which is precisely what §5.2 moved into the EMS.

  2. THE INGEST TIMESTAMP. Every accepted sample is stamped on arrival. A value
     with no age is a value that cannot be trusted (§5.2), and staleness is the
     failsafe trigger (§11.2).

  3. VALIDATION. This is untrusted external data at the edge of a safety chain.
     A malformed telegram must not become a plausible-looking number, and it
     must never resolve to 0 W - a fabricated zero is indistinguishable from a
     balanced house and defeats the very staleness trigger this module feeds.

⚠️ There is deliberately NO fallback to an inverter-side power figure. The
inverter's own AC power correlates 0.998 with battery power and 0.09 with the
real meter (§5.1) - regulating on it means regulating against your own output.
When the transport dies the correct behaviour is a gap: no sample, a growing
age, and the existing "inputs missing -> command 0 W" path in main.py.
"""

import asyncio
import json
import logging
import math
import os
import time
from dataclasses import dataclass
from datetime import datetime, timezone

import aiohttp

try:
    import paho.mqtt.client as mqtt
except ImportError:  # pragma: no cover - container always has it
    mqtt = None

_LOG = logging.getLogger("goodwe.p1")

SOURCE_HA = "ha_dsmr"
SOURCE_MQTT = "mqtt_p1"
SOURCE_HA_SIGNED = "ha_signed"
SOURCE_HOMEWIZARD = "homewizard_local"

QUARTER_S = 900

# ⚠️ Plausibility ceiling, not a clamp - anything above it is rejected as an
# anomaly rather than averaged in. Chosen to sit above the largest Belgian
# residential connection (3x63 A ~ 43 kW) and BELOW 65535: §20 open question 5
# records an HA sensor reporting 64954 for -582 W, i.e. an unsigned 16-bit
# register decoded without its sign. That corruption reads as a perfectly
# plausible 65 kW if you only bound it at "some big number".
PLAUSIBLE_MAX_W = 50_000.0


# --------------------------------------------------------------------------- #
# the sample
# --------------------------------------------------------------------------- #
class P1Error(ValueError):
    """A telegram that must be rejected rather than believed."""


@dataclass(frozen=True)
class P1Sample:
    """One telegram, validated, derived and stamped.

    Frozen on purpose: this object is handed to readers on other tasks (and,
    for the MQTT transport, produced on paho's network thread). Immutability is
    what makes "read the latest sample" safe without a lock.
    """

    ingest_ts: datetime            # tz-aware UTC, set at ingest
    ingest_mono: float             # time.monotonic() at ingest - see age_s()
    telegram_ts: datetime | None   # from the telegram, where the source has one
    source: str                    # one of the SOURCE_* constants above
    import_w: float                # unsigned magnitude, as the meter reports it
    export_w: float                # unsigned magnitude
    net_w: float                   # import_w - export_w  (+ import, - export)
    per_phase_w: tuple[float, ...] | None          # signed net, len == phases
    per_phase_import_w: tuple[float, ...] | None   # offtake only, for the tariff

    def age_s(self, now_mono: float | None = None,
              now_utc: datetime | None = None) -> float:
        """Seconds since this sample was ingested, never negative.

        ⚠️ Measured with time.monotonic(), not the wall clock. An NTP step on a
        Pi that just booted moves the wall clock by minutes; using it here would
        either fake a stale meter or, worse, hide a real one.

        Where the telegram carries its own timestamp we take the WORSE of the
        two ages. That is what stops an MQTT retained message - replayed on
        reconnect with a fresh receive time - from presenting a ten-minute-old
        reading as brand new.
        """
        now_mono = time.monotonic() if now_mono is None else now_mono
        age = max(0.0, now_mono - self.ingest_mono)
        if self.telegram_ts is not None:
            now_utc = datetime.now(timezone.utc) if now_utc is None else now_utc
            age = max(age, (now_utc - self.telegram_ts).total_seconds())
        return max(0.0, age)


def _watts(value, what: str) -> float:
    """Parse one power figure, or raise. Never returns a substituted default.

    ⚠️ Strings are refused even when float() would happily take them. A JSON
    telegram carrying "1200" where a number belongs is a payload from a source
    that is not the one we validated against, and the next surprise it has may
    not be a benign one. Transports that legitimately deal in text (HA entity
    states are always strings) convert before they get here, so this stays the
    strict edge for structured payloads.
    """
    if isinstance(value, (bool, str, bytes)) or value is None:
        raise P1Error(f"{what}: not a number ({value!r})")
    try:
        out = float(value)
    except (TypeError, ValueError):
        raise P1Error(f"{what}: not a number ({value!r})") from None
    if not math.isfinite(out):
        raise P1Error(f"{what}: not finite ({value!r})")
    if abs(out) > PLAUSIBLE_MAX_W:
        raise P1Error(f"{what}: {out:g} W is outside plausible meter range")
    return out


def split_signed(net_w) -> tuple[float, float]:
    """One signed figure -> the (import, export) magnitudes the module speaks.

    The inverse of make_sample's subtraction, and the easy direction: no second
    register to disagree with, so there is nothing to mix a fresh reading with a
    stale one. `+` is import, `-` is export - verified in test_p1.py against real
    captured readings from this house's own meter, not against a datasheet.

    ⚠️ Exactly one of the two comes out non-zero. Splitting into `(max(v,0),
    max(-v,0))` rather than clamping keeps `import_w - export_w == v` exactly, so
    the signed value the meter published survives the round trip bit for bit -
    a control loop must not be steered by a number that changed on the way in.

    ⚠️ Validation is `_watts`, the same gate the unsigned path uses: NaN,
    infinity, non-numbers and the §20 open-question-5 unsigned-decode
    contamination (64954 for -582 W) are all refused here rather than believed.
    A signed source makes that check MORE important, not less - on this path
    64954 is not obviously wrong the way a negative "unsigned" register is.
    """
    v = _watts(net_w, "net")
    return (v, 0.0) if v >= 0 else (0.0, -v)


def make_sample(source: str, import_w, export_w, *, phases: int,
                phase_import_w=None, phase_export_w=None,
                telegram_ts: datetime | None = None,
                ingest_ts: datetime | None = None,
                ingest_mono: float | None = None) -> P1Sample:
    """Validate, derive net power, stamp. Raises P1Error on anything doubtful.

    `import_w`/`export_w` are the two unsigned Belgian registers. Per-phase
    figures are equally unsigned and equally split, so each phase gets the same
    derivation.
    """
    imp = _watts(import_w, "import")
    exp = _watts(export_w, "export")
    # ⚠️ Both registers are magnitudes. A negative one means the upstream
    # already applied a sign we are about to apply again - reject it rather
    # than silently double-signing the control loop.
    if imp < 0 or exp < 0:
        raise P1Error(f"unsigned registers cannot be negative (import={imp:g} export={exp:g})")

    per_phase = per_phase_import = None
    if phase_import_w is not None or phase_export_w is not None:
        pi = list(phase_import_w or [])
        pe = list(phase_export_w or [0.0] * len(pi))
        if len(pi) != phases or len(pe) != phases:
            raise P1Error(
                f"phase count mismatch: telegram has {len(pi)} import / {len(pe)} export "
                f"phases, meter_phases is {phases}")
        vals = [_watts(a, f"L{i + 1} import") - _watts(b, f"L{i + 1} export")
                for i, (a, b) in enumerate(zip(pi, pe))]
        per_phase = tuple(vals)
        per_phase_import = tuple(max(v, 0.0) for v in vals)

    if telegram_ts is not None and telegram_ts.tzinfo is None:
        raise P1Error("telegram timestamp has no timezone")

    return P1Sample(
        ingest_ts=ingest_ts or datetime.now(timezone.utc),
        ingest_mono=time.monotonic() if ingest_mono is None else ingest_mono,
        telegram_ts=telegram_ts,
        source=source,
        import_w=imp,
        export_w=exp,
        net_w=imp - exp,
        per_phase_w=per_phase,
        per_phase_import_w=per_phase_import,
    )


# --------------------------------------------------------------------------- #
# the 15-minute average
# --------------------------------------------------------------------------- #
@dataclass(frozen=True)
class QuarterBlock:
    start: datetime                             # UTC, aligned to :00/:15/:30/:45
    offtake_avg_w: float                        # billed figure: net offtake only
    per_phase_offtake_avg_w: tuple[float, ...] | None


class QuarterAverager:
    """Time-weighted average of net offtake over clock-aligned 15-min blocks.

    Samples arrive irregularly (~1-10 s), so a plain mean over samples would
    weight a burst of fast telegrams the same as a slow one and produce a figure
    that is not the billed quantity. Each sample's value is therefore HELD until
    the next arrives and integrated over that interval: sum(value * dt) / dt.

    ⚠️ Only OFFTAKE is accumulated (§9.1) - the capacity tariff bills the highest
    quarter-hour average offtake, and a quarter of pure export averages to 0 kW,
    not to a negative one. The signed series stays available for control; this
    accumulator is for the meter's bill.

    ⚠️ Blocks are found by flooring epoch seconds to 900. That IS clock-aligned
    and DST-proof for Belgium, because every offset in that tz is a whole number
    of hours, so a 900 s grid in UTC lands on :00/:15/:30/:45 local before and
    after a transition - no tz database, no DST special case.
    ponytail: the ceiling is a timezone with a sub-hour offset (India +05:30,
    Nepal, Chatham). Those need real tz-aware boundary maths; upgrade path is to
    compute the boundary with zoneinfo instead of the modulo, everything else
    here is unchanged.

    A sample that straddles a boundary is split at the boundary and its two
    halves credited to the two blocks, never attributed wholly to either.
    """

    def __init__(self, phases: int = 1, max_hold_s: float = 30.0):
        self.phases = phases
        # ⚠️ How long one sample may be held forward before the series is
        # treated as a gap rather than a plateau. Without this the meter can die
        # while importing 5 kW, come back ten minutes later, and the hold-forward
        # credits 5 kW x 600 s to the capacity-tariff accumulator - a fabricated
        # peak, on a permanent record, from data that was never measured. Set
        # from meter_max_age_s: the point past which the reading is not trusted
        # for control is the point past which it must not be billed either.
        self.max_hold_s = float(max_hold_s)
        self._block: int | None = None       # epoch seconds of the block start
        self._acc = 0.0                      # W*s of offtake in the open block
        self._pp_acc = [0.0] * phases
        self._elapsed = 0.0                  # seconds integrated in the open block
        self._last_t: float | None = None    # epoch seconds of the held sample
        self._last_net = 0.0
        self._last_pp: tuple[float, ...] | None = None

    # -- reading ------------------------------------------------------------
    @property
    def block_start(self) -> datetime | None:
        if self._block is None:
            return None
        return datetime.fromtimestamp(self._block, timezone.utc)

    @property
    def elapsed_s(self) -> float:
        """Seconds already integrated into the open block.

        Exposed alongside the partial accumulator because SAFETY-07 projects the
        end-of-quarter average and cannot do that from a finished average.
        """
        return self._elapsed

    @property
    def partial_ws(self) -> float:
        """Offtake watt-seconds accumulated in the open block so far."""
        return self._acc

    @property
    def offtake_avg_w(self) -> float:
        """Average offtake over the part of the open block seen so far."""
        return self._acc / self._elapsed if self._elapsed > 0 else 0.0

    @property
    def per_phase_offtake_avg_w(self) -> tuple[float, ...] | None:
        if self._last_pp is None or self._elapsed <= 0:
            return None
        return tuple(a / self._elapsed for a in self._pp_acc)

    # -- writing ------------------------------------------------------------
    def add(self, sample: P1Sample) -> list[QuarterBlock]:
        """Integrate up to this sample, then hold its value. Returns any blocks
        that closed in the process (usually none, occasionally one)."""
        t = sample.ingest_ts.timestamp()
        closed: list[QuarterBlock] = []

        if self._last_t is None:
            self._block = int(t // QUARTER_S) * QUARTER_S
            self._last_t, self._last_net = t, sample.net_w
            self._last_pp = sample.per_phase_w
            return closed
        if t <= self._last_t:
            # Out-of-order or duplicate arrival: integrating a negative dt would
            # subtract energy that really happened. Drop it, keep the held value.
            return closed

        cursor = self._last_t
        # Beyond this instant the held value stops being evidence of anything.
        # The stretch from here to `t` is walked so the block boundaries are
        # still crossed correctly, but nothing is accumulated and `_elapsed`
        # does not grow - which is what makes a closed block, always divided by
        # the full 900 s, actually get dragged down by the missing coverage.
        hold_end = self._last_t + self.max_hold_s
        while True:
            end = self._block + QUARTER_S
            stop = min(t, end)
            covered = max(0.0, min(stop, hold_end) - cursor)
            if covered > 0:
                self._acc += max(self._last_net, 0.0) * covered
                if self._last_pp is not None:
                    for i, v in enumerate(self._last_pp[: self.phases]):
                        self._pp_acc[i] += max(v, 0.0) * covered
                self._elapsed += covered
            cursor = stop
            if stop < end:
                break
            closed.append(QuarterBlock(
                start=datetime.fromtimestamp(self._block, timezone.utc),
                # A closed block is always divided by the full 900 s, never by
                # the seconds we happened to observe - a gap in coverage must
                # drag the billed average down, not be averaged away.
                offtake_avg_w=self._acc / QUARTER_S,
                per_phase_offtake_avg_w=(
                    tuple(a / QUARTER_S for a in self._pp_acc)
                    if self._last_pp is not None else None),
            ))
            self._block = end
            self._acc = 0.0
            self._pp_acc = [0.0] * self.phases
            self._elapsed = 0.0

        self._last_t, self._last_net = t, sample.net_w
        self._last_pp = sample.per_phase_w
        return closed


# --------------------------------------------------------------------------- #
# what the rest of the add-on talks to
# --------------------------------------------------------------------------- #
class P1Ingest:
    """Holds the latest sample and the rolling quarter-hour average.

    ⚠️ Staleness is DERIVED from the stored sample, not carried as a separate
    flag. That is what makes "flag before the value is visible" free: there is
    one immutable object and a single attribute rebind to publish it, so a
    reader can never see a fresh value with a stale flag or the reverse.
    """

    def __init__(self, phases: int = 1, max_age_s: float = 30.0):
        self.phases = phases
        self.max_age_s = float(max_age_s)
        # The same threshold governs control and billing: a reading too old to
        # steer by is too old to bill by. See QuarterAverager.max_hold_s.
        self.averager = QuarterAverager(phases, max_hold_s=self.max_age_s)
        self.blocks: list[QuarterBlock] = []
        self.samples = 0
        self.parse_errors = 0
        self.last_error: str | None = None
        self.started_mono = time.monotonic()
        self._last: P1Sample | None = None

    @property
    def last(self) -> P1Sample | None:
        return self._last

    def submit(self, sample: P1Sample) -> None:
        self._last = sample
        self.samples += 1
        for block in self.averager.add(sample):
            self.blocks.append(block)
            del self.blocks[:-96]   # a day of quarters; STATE-01 owns real retention

    def reject(self, err: Exception | str) -> None:
        """A malformed telegram or an unavailable entity.

        ⚠️ The last good sample and ITS timestamp are left untouched. The reading
        does not become 0 W and it does not become fresh - the age keeps growing,
        which is precisely the signal a rejected telegram should produce.
        """
        self.parse_errors += 1
        self.last_error = str(err)
        _LOG.warning("P1 telegram rejected: %s", err)

    # -- what consumers read -------------------------------------------------
    def age_s(self) -> float | None:
        """Age of the newest accepted sample, or None if there has never been one."""
        return None if self._last is None else self._last.age_s()

    @property
    def published_age_s(self) -> float:
        """The figure behind `sensor.p1_sample_age_s`.

        Seconds since the newest accepted telegram, or since this ingester
        started when none has ever arrived.

        ⚠️ Always a number and never `unknown`, because SAFETY-01's firmware
        watchdog subscribes to it: an entity that simply stops existing is
        indistinguishable, from the firmware's side, from a meter that is fine.
        And ⚠️ it is recomputed against the clock on every publish rather than
        stamped once per telegram, so a meter that freezes at a constant reading
        still produces a visibly climbing age. That is the whole point of this
        entity - HA pushes state changes, so a genuinely constant P1 value emits
        nothing at all, and a watchdog watching the value would sit there
        believing the last update was recent.
        """
        age = self.age_s()
        return max(0.0, time.monotonic() - self.started_mono) if age is None else age

    @property
    def stale(self) -> bool:
        """True when there is no sample, or the newest one is past max_age_s."""
        age = self.age_s()
        return age is None or age > self.max_age_s

    @property
    def net_w(self) -> float | None:
        """Signed net grid power, or None when stale. Never a substituted zero."""
        return None if self.stale else self._last.net_w

    @property
    def per_phase_import_w(self) -> tuple[float, ...] | None:
        if self.stale or self._last is None:
            return None
        return self._last.per_phase_import_w


# --------------------------------------------------------------------------- #
# transport 1: Home Assistant WebSocket (the DSMR integration's entities)
# --------------------------------------------------------------------------- #
WS_URL = "ws://supervisor/core/websocket"
BAD_STATES = ("unknown", "unavailable", "none", "")


class HaDsmrSource:
    """Subscribes to state_changed for the configured DSMR entities.

    ⚠️ WebSocket, not REST polling. REST returns states, but polling at the
    30 s planning tick decimates a 5 s telegram stream and the quarter-hour
    average would then be computed from a sixth of the data (§5.3). "Consume
    every telegram" means event-driven.

    ⚠️ One telegram updates several entities, and HA emits one state_changed per
    entity. Building a sample on each event would mix a new import reading with
    a stale export one for a few milliseconds every 5 s. A short debounce
    coalesces the burst back into the single telegram it came from.
    """

    DEBOUNCE_S = 0.35

    def __init__(self, session: aiohttp.ClientSession, ingest: P1Ingest,
                 entities: dict, token: str | None = None):
        self.session = session
        self.ingest = ingest
        self.entities = entities          # {"import": id, "export": id, "phase_import": [...], ...}
        self.token = token or os.environ.get("SUPERVISOR_TOKEN", "")
        self.ids = self._wanted()
        self.cache: dict[str, float] = {}
        self.connected = False
        self._pending: asyncio.Task | None = None

    def _wanted(self) -> set[str]:
        out = set()
        for key in ("import", "export"):
            if self.entities.get(key):
                out.add(self.entities[key])
        for key in ("phase_import", "phase_export"):
            out.update(e for e in self.entities.get(key) or [] if e)
        return out

    async def run(self) -> None:
        """Long-lived task: connect, subscribe, reconnect with backoff, forever.

        ⚠️ A reconnect emits nothing. A gap must stay a gap - a synthetic sample
        on reconnect would reset the age and hide the outage from the very
        watchdog that exists to catch it.
        """
        backoff = 1.0
        while True:
            try:
                await self._session_once()
                backoff = 1.0
            except asyncio.CancelledError:
                raise
            except Exception as err:  # noqa: BLE001 - any transport fault retries
                _LOG.warning("P1 HA websocket: %s - reconnecting in %.0fs", err, backoff)
            finally:
                self.connected = False
            await asyncio.sleep(backoff)
            backoff = min(backoff * 2, 30.0)

    async def _session_once(self) -> None:
        async with self.session.ws_connect(WS_URL, heartbeat=30) as ws:
            hello = await ws.receive_json()
            if hello.get("type") == "auth_required":
                await ws.send_json({"type": "auth", "access_token": self.token})
                reply = await ws.receive_json()
                if reply.get("type") != "auth_ok":
                    raise RuntimeError(f"auth rejected: {reply.get('message', reply)}")
            await ws.send_json({"id": 1, "type": "subscribe_events",
                                "event_type": "state_changed"})
            await ws.send_json({"id": 2, "type": "get_states"})
            self.connected = True
            _LOG.info("P1 ingest: subscribed to %s", ", ".join(sorted(self.ids)))

            async for msg in ws:
                if msg.type is not aiohttp.WSMsgType.TEXT:
                    continue
                payload = json.loads(msg.data)
                if payload.get("id") == 2 and payload.get("type") == "result":
                    # ⚠️ Prime the cache, but do NOT build a sample from it.
                    # get_states returns whatever HA currently holds, which
                    # after a Core restart is a RestoreEntity value of unknown
                    # age. Stamping that with ingest_ts=now resets the age to
                    # zero and reports a fresh meter that may have been dead for
                    # an hour - a synthetic sample hiding the outage from the
                    # watchdog that exists to catch it. The cache is what lets
                    # the FIRST real state_changed build a complete sample; the
                    # age stays honest until one arrives.
                    for obj in payload.get("result") or []:
                        self._absorb(obj.get("entity_id"), obj.get("state"))
                elif payload.get("type") == "event":
                    data = (payload.get("event") or {}).get("data") or {}
                    if data.get("entity_id") not in self.ids:
                        continue
                    new = data.get("new_state") or {}
                    self._absorb(data.get("entity_id"), new.get("state"))
                    self._schedule()
            raise RuntimeError("websocket closed")

    def _absorb(self, entity_id: str | None, state) -> None:
        if not entity_id or entity_id not in self.ids:
            return
        raw = str(state).strip().lower()
        if raw in BAD_STATES:
            # ⚠️ An `unavailable` DSMR entity is a missing reading, not 0 W.
            # Forget the cached value so no sample can be built from a mixture
            # of a live register and one that stopped reporting.
            self.cache.pop(entity_id, None)
            self.ingest.reject(f"{entity_id} is {raw}")
            return
        try:
            self.cache[entity_id] = float(raw)
        except ValueError:
            self.cache.pop(entity_id, None)
            self.ingest.reject(f"{entity_id} is not numeric: {raw!r}")

    def _schedule(self) -> None:
        if self._pending and not self._pending.done():
            return
        self._pending = asyncio.get_running_loop().create_task(self._after_debounce())

    async def _after_debounce(self) -> None:
        await asyncio.sleep(self.DEBOUNCE_S)
        self.build()

    def build(self) -> bool:
        """Assemble one sample from the cache. Returns True if one was accepted."""
        imp_id, exp_id = self.entities.get("import"), self.entities.get("export")
        if imp_id not in self.cache or exp_id not in self.cache:
            return False
        pi = [self.cache.get(e) for e in self.entities.get("phase_import") or []]
        pe = [self.cache.get(e) for e in self.entities.get("phase_export") or []]
        if pi and (None in pi or (pe and None in pe)):
            return False           # incomplete phase set: wait, do not guess
        try:
            self.ingest.submit(make_sample(
                SOURCE_HA, self.cache[imp_id], self.cache[exp_id],
                phases=self.ingest.phases,
                phase_import_w=pi or None,
                phase_export_w=pe or None,
                # ⚠️ No telegram_ts: HA's last_changed is when the STATE changed,
                # which for a constant reading is minutes ago even though the
                # telegram is current. Using it as a telegram time would fake
                # staleness on a genuinely steady meter.
            ))
            return True
        except P1Error as err:
            self.ingest.reject(err)
            return False


# --------------------------------------------------------------------------- #
# transport 2: MQTT
# --------------------------------------------------------------------------- #
def parse_mqtt_payload(raw: bytes | str, phases: int, *,
                       now: datetime | None = None) -> P1Sample:
    """One JSON telegram from the configured topic. Raises P1Error.

    The accepted document, documented in DOCS.md:

        {"import_w": 1234.0, "export_w": 0.0,
         "phases": [{"import_w": 500, "export_w": 0}, ...],   # optional
         "timestamp": "2026-08-24T18:00:05+02:00"}            # optional

    ponytail: one strict schema rather than sniffing the half-dozen P1-bridge
    dialects in the wild. The upgrade path is a `meter_mqtt_format` option
    selecting a parser; a lenient parser is the wrong default at a safety
    boundary, where guessing a key means guessing a kilowatt.
    """
    try:
        doc = json.loads(raw)
    except (ValueError, TypeError) as err:
        raise P1Error(f"payload is not JSON: {err}") from None
    if not isinstance(doc, dict):
        raise P1Error(f"payload is not a JSON object ({type(doc).__name__})")

    ts = None
    if doc.get("timestamp"):
        try:
            ts = datetime.fromisoformat(str(doc["timestamp"]))
        except ValueError:
            raise P1Error(f"unparseable timestamp {doc['timestamp']!r}") from None
        if ts.tzinfo is None:
            raise P1Error("timestamp has no UTC offset")

    pi = pe = None
    if "phases" in doc:
        rows = doc["phases"]
        if not isinstance(rows, list) or not all(isinstance(r, dict) for r in rows):
            raise P1Error("'phases' must be a list of objects")
        pi = [r.get("import_w") for r in rows]
        pe = [r.get("export_w", 0.0) for r in rows]

    return make_sample(SOURCE_MQTT, doc.get("import_w"), doc.get("export_w"),
                       phases=phases, phase_import_w=pi, phase_export_w=pe,
                       telegram_ts=ts, ingest_ts=now)


class MqttP1Source:
    """Subscribes to one topic and submits every message that parses.

    Uses paho's own reconnect loop on its own thread, then hops back onto the
    event loop with call_soon_threadsafe so the ingest state is only ever
    mutated from one thread.
    """

    def __init__(self, ingest: P1Ingest, topic: str, host, port=1883,
                 username=None, password=None):
        self.ingest = ingest
        self.topic = topic
        self.host, self.port = host, int(port or 1883)
        self.username, self.password = username, password
        self.client = None
        self.loop = None

    async def run(self) -> None:
        if mqtt is None or not self.host or not self.topic:
            _LOG.error("P1 MQTT source not usable (broker=%s topic=%r) - no meter data",
                       self.host, self.topic)
            return
        self.loop = asyncio.get_running_loop()
        try:
            self.client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2,
                                      client_id="goodwe_p1_ingest")
        except AttributeError:      # paho 1.x, which is what Alpine ships
            self.client = mqtt.Client(client_id="goodwe_p1_ingest")
        if self.username:
            self.client.username_pw_set(self.username, self.password or "")
        self.client.on_connect = lambda *_a, **_k: self.client.subscribe(self.topic, qos=0)
        self.client.on_message = self._on_message
        self.client.reconnect_delay_set(min_delay=1, max_delay=30)
        self.client.connect_async(self.host, self.port, keepalive=60)
        self.client.loop_start()
        _LOG.info("P1 ingest: MQTT %s:%s topic %s", self.host, self.port, self.topic)
        try:
            while True:
                await asyncio.sleep(3600)
        finally:
            self.client.loop_stop()
            self.client.disconnect()

    def _on_message(self, _client, _userdata, msg) -> None:
        # Runs on paho's network thread.
        if self.loop is None:
            return
        self.loop.call_soon_threadsafe(self._handle, msg.payload)

    def _handle(self, payload) -> None:
        try:
            self.ingest.submit(parse_mqtt_payload(payload, self.ingest.phases))
        except P1Error as err:
            self.ingest.reject(err)


# --------------------------------------------------------------------------- #
# transport 3: Home Assistant WebSocket, one signed entity
# --------------------------------------------------------------------------- #
class HaSignedSource(HaDsmrSource):
    """The same websocket, subscribed to ONE signed power entity.

    For readers that publish net power as a single signed figure - a HomeWizard
    P1's `sensor.p1_meter_active_power`, positive = import - rather than the two
    unsigned DSMR registers. This is the meter actually installed at the house,
    and `ha_dsmr` cannot read it: it needs two registers and refuses a negative
    one outright, which is every exporting telegram.

    ⚠️ A subclass, not a copy. The connect / auth / subscribe / reconnect /
    `_absorb` machinery above is transport, not shape, and it has already been
    debugged once - notably "prime the cache from get_states but never build a
    sample out of it" and "a reconnect emits nothing". Only `_wanted` (which
    entity ids) and `build` (how they become a sample) differ, so only those two
    are overridden. Everything TEL-01 established therefore applies unchanged:
    ingest timestamping, meter_max_age_s, the clock-recomputed age sensor, and
    `unavailable` treated as a missing reading rather than 0 W.

    ⚠️ THE AGE ON THIS TRANSPORT MEASURES TIME SINCE THE VALUE CHANGED, not time
    since the meter reported, and on one entity those are very different things.
    Home Assistant offers no arrival signal for a repeated reading: it emits no
    `state_changed`, it does not advance `last_reported` on either serialiser,
    and `state_reported` cannot be subscribed to over the websocket at all
    ("Event filter is required for event state_reported"). All three measured on
    the ENV-01 rig against the real HomeWizard integration with the meter frozen
    - 0 state_changed in 70 s, no timestamp movement anywhere.

    `ha_dsmr` mostly escapes it because a DSMR telegram moves several entities at
    once. This transport has ONE, so a healthy meter under a flat load looks
    exactly like a dead one - and our own capture has the real meter going 42.2 s
    and 97.0 s between changes, both past the default max_age_s of 30. Hence
    DOCS.md: sensor.p1_sample_age_s is correct while the value moves and must not
    yet be thresholded by the firmware watchdog on this transport. Raising
    meter_max_age_s does not fix it, it only chooses which of the two errors you
    get. The fix is an arrival stamp from the meter itself - reading the
    HomeWizard local API rather than an HA entity - which is a separate ticket.

    ⚠️ The debounce is inherited but does nothing useful here, and that is fine:
    one telegram is one entity, so there is no burst of per-entity events to
    coalesce and no window in which a new reading sits beside a stale one. It
    costs one scheduled sleep per telegram at ~0.2 Hz. Left in place rather than
    special-cased, because a second code path through build() is a second place
    for the sign to go wrong.
    """

    def _wanted(self) -> set[str]:
        out = set()
        if self.entities.get("net"):
            out.add(self.entities["net"])
        out.update(e for e in self.entities.get("phase_net") or [] if e)
        return out

    def build(self) -> bool:
        """Assemble one sample from the cache. Returns True if one was accepted."""
        net_id = self.entities.get("net")
        if net_id not in self.cache:
            return False
        pn = [self.cache.get(e) for e in self.entities.get("phase_net") or []]
        if pn and None in pn:
            return False           # incomplete phase set: wait, do not guess
        try:
            imp, exp = split_signed(self.cache[net_id])
            pi = pe = None
            if pn:
                # ponytail: this split is arithmetically redundant today -
                # make_sample subtracts the two lists again and does not
                # sign-check per-phase figures, so handing it the signed values
                # with a zero export list produces the identical tuple. Verified:
                # mutating it that way leaves all 174 checks green, i.e. no test
                # can tell the difference, and it is recorded here rather than
                # left as a silent equivalent mutant for the next reviewer to
                # rediscover. Kept because `phase_import_w` means a MAGNITUDE:
                # a negative in it is the double-signing that make_sample refuses
                # outright for the connection-level registers, and the day that
                # check is extended per-phase the shortcut breaks the meter, not
                # the test.
                pairs = [split_signed(v) for v in pn]
                pi = [a for a, _ in pairs]
                pe = [b for _, b in pairs]
            self.ingest.submit(make_sample(
                SOURCE_HA_SIGNED, imp, exp,
                phases=self.ingest.phases,
                phase_import_w=pi, phase_export_w=pe,
                # ⚠️ No telegram_ts, for the same reason as ha_dsmr: HA's
                # last_changed is when the VALUE changed, which on a steady meter
                # is minutes ago while the telegram is current. sensor.
                # p1_sample_age_s is what covers a genuinely frozen meter.
            ))
            return True
        except P1Error as err:
            self.ingest.reject(err)
            return False


# --------------------------------------------------------------------------- #
# transport 4: the HomeWizard P1's own local API
# --------------------------------------------------------------------------- #
HOMEWIZARD_PATH = "/api/v1/data"
HOMEWIZARD_PORT = 80


def parse_homewizard(doc, phases: int, *, now: datetime | None = None,
                     ingest_mono: float | None = None) -> P1Sample:
    """One `GET /api/v1/data` document -> a sample. Raises P1Error.

    The meter's own JSON, of which we read four keys:

        {"active_power_w": -5710.0,          # SIGNED, + import  - export
         "active_power_l1_w": ..., "active_power_l2_w": ..., "..._l3_w": ...,
         "total_power_import_kwh": 1234.567, "total_power_export_kwh": 890.123}

    Same shape as `ha_signed` - one signed connection figure plus signed
    per-phase figures - so it goes through the same `split_signed` and the same
    `make_sample`. Nothing about the sign convention is re-derived here.

    ⚠️ No telegram timestamp, because the API carries none, and that is correct
    rather than a gap: the document was produced by the meter in the moment it
    answered, so the ingest stamp IS the measurement time. This is the whole
    reason the transport exists - see HomeWizardLocalSource.

    ⚠️ Per-phase figures are taken only when the meter serves ALL `phases` of
    them. A single-phase HomeWizard returns `active_power_l2_w: null`, and a
    partial set must not become a fabricated tuple. Unlike the HA transports
    there is no "wait for the rest" case to worry about: every field here came
    out of ONE response, so a missing phase cannot be a stale phase, and the
    connection-level reading is still a genuine, complete measurement. Dropping
    the whole sample over an absent per-phase field would turn a healthy meter
    into a climbing age, which is the exact false-trip this ticket removes.
    """
    if not isinstance(doc, dict):
        raise P1Error(f"meter response is not a JSON object ({type(doc).__name__})")

    imp, exp = split_signed(doc.get("active_power_w"))

    pi = pe = None
    legs = [doc.get(f"active_power_l{i + 1}_w") for i in range(phases)]
    if all(v is not None for v in legs):
        pairs = [split_signed(v) for v in legs]
        pi = [a for a, _ in pairs]
        pe = [b for _, b in pairs]

    return make_sample(SOURCE_HOMEWIZARD, imp, exp, phases=phases,
                       phase_import_w=pi, phase_export_w=pe,
                       ingest_ts=now, ingest_mono=ingest_mono)


def measurement_fingerprint(doc) -> tuple:
    """The part of a response that a genuinely new measurement has to move.

    Power plus both energy registers. A meter under any real load advances a
    kWh counter (1 Wh resolution: ~10 s at 350 W), so this changes even when the
    power figure happens to repeat. Used ONLY for `unchanged_s` - see there for
    why it must not be allowed anywhere near the age.
    """
    if not isinstance(doc, dict):
        return ()
    return (doc.get("active_power_w"),
            doc.get("total_power_import_kwh"),
            doc.get("total_power_export_kwh"))


class HomeWizardLocalSource:
    """Polls the meter's own local HTTP API instead of a Home Assistant entity.

    ⚠️ THIS IS THE ONLY TRANSPORT WHOSE AGE MEASURES ARRIVAL. That is the entire
    reason it exists, and it is not an optimisation - it is the difference
    between an age sensor a firmware watchdog may threshold and one it may not.

    Every HTTP response is a genuine arrival: the meter answered, now, with its
    current reading. Whether the NUMBER moved is irrelevant, so a healthy meter
    under a flat load resets the age exactly like a busy one. Home Assistant
    cannot provide this at all - it emits `state_changed` only on a change, does
    not advance `last_reported` on either serialiser for a repeat, and refuses a
    `state_reported` subscription outright ("Event filter is required"). Measured
    on the ENV-01 rig over 70 s of a frozen meter, and independently against the
    live house over ten repeated readings. So on `ha_signed` the age means "time
    since the value changed", and our own capture of this house's meter has
    42.2 s and 97.0 s between changes - both past the default `meter_max_age_s`
    of 30, i.e. a false trip to 0 W on a perfectly healthy meter.

    Everything else is TEL-01's pipeline unchanged: ingest timestamping,
    `meter_max_age_s`, the clock-recomputed age, the plausibility bounds, and a
    failed read treated as a MISSING reading rather than 0 W.

    ⚠️ A failed or timed-out poll submits nothing. It is a rejection, so the
    last good sample and its stamp stay exactly where they were and the age goes
    on climbing - which is precisely the signal a dead meter should produce.

    ⚠️ POLL CADENCE vs `meter_max_age_s`. The age can only be as fresh as the
    poll interval, so `meter_poll_s` must sit well under `meter_max_age_s` or the
    reading flaps in and out of staleness on a healthy meter. The real meter
    updates every ~5.0 s (NOTES.md:640 measures 4.97 s), so polling faster than
    that buys nothing but re-reads. Default 5 s against a 30 s max age; the
    startup check in build_source refuses `meter_poll_s >= meter_max_age_s` and
    warns past half of it.

    ponytail: a plain `asyncio.sleep` loop rather than a scheduler, and no
    backoff. A poll that fails costs one rejection and is retried on the next
    tick; a meter that is down stays down and the age reports it. The ceiling is
    a meter so slow that polls overlap - the request timeout is held under the
    poll interval so they cannot.
    """

    def __init__(self, session, ingest: P1Ingest, host: str, *,
                 port: int = HOMEWIZARD_PORT, poll_s: float = 5.0,
                 timeout_s: float | None = None):
        self.session = session
        self.ingest = ingest
        self.host, self.port = host, int(port)
        self.poll_s = float(poll_s)
        # ⚠️ Held under the poll interval on purpose: a timeout longer than the
        # cadence queues polls behind a hung meter, and a backlog of requests
        # all landing at once would stamp several arrivals for one measurement.
        self.timeout_s = float(timeout_s) if timeout_s else max(1.0, self.poll_s * 0.8)
        self.url = f"http://{self.host}:{self.port}{HOMEWIZARD_PATH}"
        self.connected = False
        self.polls = 0
        self.poll_errors = 0
        self._fingerprint: tuple | None = None
        self._changed_mono: float | None = None

    @property
    def unchanged_s(self) -> float | None:
        """Seconds since the meter last served a DIFFERENT measurement.

        ⚠️ Diagnostic only, and deliberately NOT folded into
        `sensor.p1_sample_age_s`. A frozen meter (SAFETY-01's six-minute
        runaway) answers 200 OK with a stale document forever, so no arrival
        detector can see it - only the document standing still can, and this is
        that signal. It is exposed rather than acted on because it is NOT safe
        as a 30 s watchdog input: this controller regulates grid power toward
        ~0 W, and at a converged -10 W the export register takes six minutes to
        advance by its 1 Wh resolution while the power figure legitimately
        repeats. Thresholding that would rebuild the very limit cycle the
        arrival stamp just removed, at the exact operating point we aim for.
        Whoever wires a freeze detector must handle the low-power case first.
        """
        if self._changed_mono is None:
            return None
        return max(0.0, time.monotonic() - self._changed_mono)

    async def run(self) -> None:
        """Long-lived task: poll, submit, sleep, forever."""
        _LOG.info("P1 ingest: polling %s every %.1fs", self.url, self.poll_s)
        while True:
            await self.poll_once()
            await asyncio.sleep(self.poll_s)

    async def poll_once(self) -> bool:
        """One GET. Returns True if a sample was accepted."""
        self.polls += 1
        try:
            async with self.session.get(
                    self.url,
                    timeout=aiohttp.ClientTimeout(total=self.timeout_s)) as resp:
                if resp.status != 200:
                    raise P1Error(f"meter returned HTTP {resp.status}")
                # content_type=None: the meter's own firmware is the authority on
                # what it serves, and refusing a reading over a Content-Type
                # header would be a fabricated outage.
                # ⚠️ Known EQUIVALENT MUTANT: dropping this argument leaves all
                # 236 checks green, because aiohttp's own json_response - which
                # the fake meter in test_p1.py uses - always sets
                # application/json, so no test can serve valid JSON under a
                # wrong header. Recorded here rather than left for the next
                # reviewer to rediscover. It is kept because a real HomeWizard
                # firmware that ever answers text/plain would otherwise read as
                # a dead meter and hold the battery at 0 W.
                doc = await resp.json(content_type=None)
        except asyncio.CancelledError:
            raise
        except Exception as err:  # noqa: BLE001 - any read failure is a gap
            self.connected = False
            self.poll_errors += 1
            self.ingest.reject(f"meter poll {self.url}: {err}")
            return False

        self.connected = True
        try:
            sample = parse_homewizard(doc, self.ingest.phases)
        except P1Error as err:
            self.ingest.reject(err)
            return False
        # ⚠️ Submitted unconditionally, INCLUDING a document identical to the
        # last one. The response is the arrival; the number is not.
        self.ingest.submit(sample)
        self._note(doc)
        return True

    def _note(self, doc) -> None:
        fp = measurement_fingerprint(doc)
        if self._changed_mono is None or fp != self._fingerprint:
            self._fingerprint = fp
            self._changed_mono = time.monotonic()


# --------------------------------------------------------------------------- #
# selection
# --------------------------------------------------------------------------- #
def is_enabled(opts: dict) -> bool:
    """Whether P1 ingestion is switched on at all.

    ⚠️ One definition, because three places depend on it and they MUST agree:
    where the grid reading comes from, whether the ingest task is started, and
    whether sensor.p1_sample_age_s is announced over MQTT discovery. An age
    sensor announced with no ingester behind it is a watchdog input nobody is
    feeding, and the ESP32 trips on it.
    """
    return str(opts.get("meter_source", "off") or "off").strip() not in ("off", "")


def build_source(opts: dict, ingest: P1Ingest, session, broker: dict | None):
    """Return the transport named by `meter_source`, or None if disabled.

    This is the whole of AC 1's "switchable": every consumer reads P1Ingest, so
    changing transport is a config edit, never a code path.
    """
    source = str(opts.get("meter_source", "off") or "off").strip()
    if not is_enabled(opts):
        return None
    if source == SOURCE_HA:
        return HaDsmrSource(session, ingest, {
            "import": opts.get("p1_import_entity", ""),
            "export": opts.get("p1_export_entity", ""),
            "phase_import": opts.get("p1_phase_import_entities") or [],
            "phase_export": opts.get("p1_phase_export_entities") or [],
        })
    if source == SOURCE_HA_SIGNED:
        net = str(opts.get("p1_net_entity", "") or "").strip()
        phase_net = [str(e).strip() for e in
                     (opts.get("p1_phase_net_entities") or []) if str(e).strip()]
        # ⚠️ Both of these are checked ONCE here rather than per telegram. A
        # misconfigured source otherwise fails silently in the only way that
        # looks exactly like a healthy one that has not been sent anything yet:
        # no samples, a climbing age, and the firmware watchdog holding the
        # battery at 0 W with nothing in the log saying why.
        if not net:
            _LOG.error("meter_source %s needs p1_net_entity - P1 ingestion "
                       "disabled (sensor.p1_sample_age_s would otherwise be "
                       "announced with nothing feeding it)", SOURCE_HA_SIGNED)
            return None
        if phase_net and len(phase_net) != ingest.phases:
            _LOG.error("p1_phase_net_entities has %d entities but meter_phases "
                       "is %d - P1 ingestion disabled. Every telegram would be "
                       "rejected on the phase-count check.",
                       len(phase_net), ingest.phases)
            return None
        return HaSignedSource(session, ingest,
                              {"net": net, "phase_net": phase_net})
    if source == SOURCE_HOMEWIZARD:
        host = str(opts.get("p1_host", "") or "").strip()
        # A pasted browser URL is the obvious way to fill this in wrong.
        host = host.removeprefix("http://").removeprefix("https://").rstrip("/")
        host, _, port_s = host.partition(":")
        # ⚠️ Checked ONCE here, like the ha_signed entity ids and for the same
        # reason: a misconfigured source otherwise fails in the one way that
        # looks exactly like a healthy one nobody has sent anything to yet - no
        # samples, a climbing age, the watchdog holding the battery at 0 W, and
        # nothing in the log saying why.
        if not host:
            _LOG.error("meter_source %s needs p1_host (the meter's own address, "
                       "e.g. 192.168.2.250) - P1 ingestion disabled",
                       SOURCE_HOMEWIZARD)
            return None
        try:
            port = int(port_s) if port_s else HOMEWIZARD_PORT
        except ValueError:
            _LOG.error("p1_host %r has an unparseable port - P1 ingestion disabled",
                       opts.get("p1_host"))
            return None
        try:
            # ⚠️ Not `or 5`: that turns an explicit 0 into the default, and a
            # cadence of 0 is a misconfiguration that must be reported, not
            # quietly corrected into something that looks like it was asked for.
            raw = opts.get("meter_poll_s", 5)
            poll_s = float(5 if raw is None or raw == "" else raw)
        except (TypeError, ValueError):
            _LOG.error("meter_poll_s %r is not a number - P1 ingestion disabled",
                       opts.get("meter_poll_s"))
            return None
        if poll_s <= 0:
            _LOG.error("meter_poll_s must be positive - P1 ingestion disabled")
            return None
        # ⚠️ The age can never be fresher than the poll interval. At or past
        # max_age_s every reading is stale before its successor arrives, so the
        # controller would sit permanently on missing inputs while the meter is
        # perfectly healthy - refuse it rather than ship that.
        if poll_s >= ingest.max_age_s:
            _LOG.error("meter_poll_s %.1f is not under meter_max_age_s %.1f - every "
                       "reading would go stale before the next poll. P1 ingestion "
                       "disabled.", poll_s, ingest.max_age_s)
            return None
        if poll_s > ingest.max_age_s / 2:
            _LOG.warning("meter_poll_s %.1f leaves no room under meter_max_age_s "
                         "%.1f: one missed poll makes the reading stale. The meter "
                         "updates every ~5 s; 5 s against 30 s is the tested pair.",
                         poll_s, ingest.max_age_s)
        return HomeWizardLocalSource(session, ingest, host, port=port, poll_s=poll_s)
    if source == SOURCE_MQTT:
        broker = broker or {}
        return MqttP1Source(ingest, str(opts.get("meter_mqtt_topic", "")),
                            broker.get("host"), broker.get("port", 1883),
                            broker.get("username"), broker.get("password"))
    if source:
        _LOG.error("meter_source %r is not one of %s - P1 ingestion disabled",
                   source, ", ".join((SOURCE_HA, SOURCE_MQTT, SOURCE_HA_SIGNED,
                                      SOURCE_HOMEWIZARD)))
    return None