glenn schrooyenandClaude Opus 5
4bd659c499
T-1, and it was a fleet-wide trip to zero. publish() emitted p1_age unconditionally, and published_age_s counts from P1Ingest.__init__ when no sample has ever arrived. With meter_source defaulting to off, every existing install would have published sensor.p1_sample_age_s climbing without bound; the ESP32 does `has_state() && state >= max_age_s` and forces the layer-1 failsafe, so each of them would have pinned its inverter at 0 W within 30 s. Exactly the opposite of the zero-regression the off default was for. The key is now omitted from the payload AND from MQTT discovery when P1 is off, so the entity does not exist at all - which is the status quo, and what has_state() is testing for. The predicate is one function, is_enabled(), because the grid reading, the task start and the discovery announcement have to agree or this comes back. T-2, connect no longer manufactures a sample. get_states returns whatever HA currently holds, which after a Core restart is a RestoreEntity value of unknown age; stamping it with ingest_ts=now reset the age and reported a fresh meter that could have been dead for an hour. run()'s own docstring already said a reconnect must emit nothing - the code disagreed with it, and a test asserted the violation. The cache is still primed, so the first real state_changed builds a complete sample; the age just stays honest until one arrives. T-3, gaps are no longer filled with the last held value. The averager held a sample forward across any interval, so a meter dying at 5 kW and returning ten minutes later credited 5 kW x 600 s to the capacity-tariff accumulator - a fabricated peak on a permanent record. The hold is capped at max_age_s: past that the stretch is walked so block boundaries still land correctly, but nothing accumulates and elapsed does not grow, which is what finally makes the comment about a gap dragging the billed average down true. Same threshold for control and billing: a reading too old to steer by is too old to bill by. T-4, the out-of-order/duplicate guard is covered. It was untested, and the reason is worth recording: the obvious assertion passes without the guard, because the negative interval is separately refused by the covered > 0 test. What the guard prevents is the timestamp REWIND, which only shows up one sample later as a re-integrated window. The test now goes one sample later. T-6, DOCS was wrong about latency. meter_max_age_s and stale_input_s stack, so meter death to 0 W is 45 s and not 30. Documented as a table with both clocks. Also documented the T-5 asymmetry rather than papering over it: the age measures arrival, not change, so a stuck MQTT bridge republishing its last telegram still looks fresh. Correct on ha_dsmr, not detectable on mqtt_p1 without a change-detector. Written up as a known limit. Writing the T-1 test caught a second defect in the test itself: it recorded only MQTT topics, and object_id lives in the payload, so "the age sensor is not announced" had been passing for the wrong reason. test_p1.py: 99 -> 122 checks. 14 mutations run, all 14 red, files restored byte-identical - including one per fix above. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Du77usMj8XNKNFZGmUiWDa
711 lines
32 KiB
Python
711 lines
32 KiB
Python
"""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 meter exposes two UNSIGNED
|
|
registers - consumption and injection - never one signed figure. 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.
|
|
|
|
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"
|
|
|
|
QUARTER_S = 900
|
|
|
|
# ⚠️ Plausibility ceiling, not a clamp - anything above it is rejected as an
|
|
# anomaly rather than averaged in. Chosen to sit above the largest Belgian
|
|
# residential connection (3x63 A ~ 43 kW) and BELOW 65535: §20 open question 5
|
|
# records an HA sensor reporting 64954 for -582 W, i.e. an unsigned 16-bit
|
|
# register decoded without its sign. That corruption reads as a perfectly
|
|
# plausible 65 kW if you only bound it at "some big number".
|
|
PLAUSIBLE_MAX_W = 50_000.0
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# the sample
|
|
# --------------------------------------------------------------------------- #
|
|
class P1Error(ValueError):
|
|
"""A telegram that must be rejected rather than believed."""
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class P1Sample:
|
|
"""One telegram, validated, derived and stamped.
|
|
|
|
Frozen on purpose: this object is handed to readers on other tasks (and,
|
|
for the MQTT transport, produced on paho's network thread). Immutability is
|
|
what makes "read the latest sample" safe without a lock.
|
|
"""
|
|
|
|
ingest_ts: datetime # tz-aware UTC, set at ingest
|
|
ingest_mono: float # time.monotonic() at ingest - see age_s()
|
|
telegram_ts: datetime | None # from the telegram, where the source has one
|
|
source: str # SOURCE_HA | SOURCE_MQTT
|
|
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 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)
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# 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_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 %s or %s - P1 ingestion disabled",
|
|
source, SOURCE_HA, SOURCE_MQTT)
|
|
return None
|