Compare commits
12
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
 glenn schrooyenandClaude Opus 5
|
ad9c5772a4 | ||
|
glenn schrooyenandClaude Opus 5
|
1b343da8e3 | ||
|
glenn schrooyenandClaude Opus 5
|
e663e10245 | ||
|
glenn schrooyenandClaude Opus 5
|
632be44f6c | ||
|
glenn schrooyenandClaude Opus 5
|
8b51a51e20 | ||
|
glenn schrooyenandClaude Opus 5
|
c24bc0a011 | ||
|
glenn schrooyenandClaude Opus 5
|
80402c978f | ||
|
glenn schrooyenandClaude Opus 5
|
6c980e87b0 | ||
|
glenn schrooyenandClaude Opus 5
|
505a847d85 | ||
|
glenn schrooyenandClaude Opus 5
|
f498d5fa54 | ||
|
glenn schrooyenandClaude Opus 5
|
4bd659c499 | ||
|
glenn schrooyenandClaude Opus 5
|
f47f1f0129 |
@@ -0,0 +1,10 @@
|
||||
# This add-on is deployed to a Linux container. core.autocrlf=true on the
|
||||
# authoring box gave the checkout CRLF, so a plain copy shipped CRLF files -
|
||||
# run.sh with CRLF is a "bad interpreter" failure, and any hash-based drift
|
||||
# check between repo and deployment fails for a reason that has nothing to do
|
||||
# with the code. Deploy with:
|
||||
# git -c core.autocrlf=false archive release/1.0 goodwe_controller | tar -x
|
||||
# which is how 0.3.0 went out, byte-identical to the blobs.
|
||||
* text=auto eol=lf
|
||||
*.png binary
|
||||
*.gz binary
|
||||
@@ -1,5 +1,75 @@
|
||||
# Changelog
|
||||
|
||||
## Unreleased
|
||||
|
||||
**TEL-04.** A third `meter_source`, `ha_signed`, reading **one signed** Home
|
||||
Assistant entity: positive = import, negative = export. That is the shape a
|
||||
HomeWizard P1 publishes (`sensor.p1_meter_active_power`), and it is the meter
|
||||
actually fitted here - which neither TEL-01 transport can read, because
|
||||
`ha_dsmr` needs two unsigned registers and refuses a negative one, i.e. every
|
||||
exporting telegram. Set `p1_net_entity`, and `p1_phase_net_entities` for the
|
||||
per-phase capacity-tariff figures on a three-phase connection.
|
||||
|
||||
Everything TEL-01 established is inherited rather than re-implemented - the
|
||||
new transport is a subclass of the `ha_dsmr` one overriding only which
|
||||
entities it wants and how they become a sample. So ingest timestamping,
|
||||
`meter_max_age_s`, `sensor.p1_sample_age_s` recomputed against the clock and
|
||||
republished once a second, the plausibility bounds, and `unavailable` /
|
||||
`unknown` treated as a *missing reading and never 0 W* all behave identically
|
||||
across the three sources.
|
||||
|
||||
Still defaults to `off`; an existing install is unaffected until it opts in.
|
||||
|
||||
⚠️ **`sensor.p1_sample_age_s` is published on `ha_signed`, but must not yet be
|
||||
thresholded by the ESP32 stale-input watchdog.** On the HA WebSocket paths the
|
||||
age is stamped from `state_changed`, so it measures time since the value
|
||||
*changed*, not since the meter *reported* - and Home Assistant exposes no
|
||||
arrival signal for a repeated reading (no `state_changed`, no `last_reported`
|
||||
movement on either serialiser, and `state_reported` is not subscribable over
|
||||
the WebSocket). Measured on the ENV-01 rig against the real HomeWizard
|
||||
integration. `ha_dsmr` mostly escapes it because a telegram moves several
|
||||
entities at once; `ha_signed` has one, so a healthy meter under a flat load is
|
||||
indistinguishable from a dead one. Our own capture has the house meter going
|
||||
42.2 s and 97.0 s between changes. Raising `meter_max_age_s` does not fix that,
|
||||
it only chooses which error you get; the fix is an arrival stamp from the meter
|
||||
itself and is a separate ticket. Full detail in DOCS.md.
|
||||
|
||||
## 0.3.0
|
||||
|
||||
**SAFETY-04.** The control law's integrator is now an explicit accumulator,
|
||||
bounded independently of the output clamp instead of inheriting whatever
|
||||
headroom the clamp happened to leave. It also freezes while the inverter is
|
||||
not tracking, rather than continuing to wind up against a command nothing is
|
||||
acting on. `integrator_max_w` (default `0`) governs the bound; `0` means
|
||||
"follow `max_w`", which is the existing behaviour.
|
||||
|
||||
Behaviour is unchanged at the defaults - a 4,928-case equivalence sweep
|
||||
against the previous control law confirms it decides identically at
|
||||
`integrator_max_w: 0`.
|
||||
|
||||
**TEL-01.** P1 meter ingestion, so a Belgian P1's two unsigned registers
|
||||
(consumption, injection) no longer need a hand-written signed template
|
||||
sensor: the subtraction moves into the add-on, done once and tested. Two
|
||||
transports, chosen with the new `meter_source` option: `ha_dsmr` subscribes
|
||||
to the DSMR integration over the HA WebSocket, `mqtt_p1` reads a topic.
|
||||
Defaults to `off`, which keeps the existing `meter_entity` path untouched -
|
||||
nothing changes for an install that does not opt in.
|
||||
|
||||
Enabling it publishes `sensor.p1_sample_age_s`: seconds since the newest
|
||||
accepted telegram, recomputed against the clock and republished roughly once
|
||||
a second rather than only when a telegram lands. That is deliberate - Home
|
||||
Assistant only pushes a state on change, so a meter sitting at a genuinely
|
||||
constant reading would otherwise look identical to a dead one. Watching the
|
||||
age instead means a frozen meter shows a climbing age, not a flat line. The
|
||||
firmware watchdog subscribes to this exact entity id.
|
||||
|
||||
Known limits, both already in DOCS.md: on `mqtt_p1`, a bridge stuck
|
||||
republishing its last telegram still "arrives", so the age cannot detect
|
||||
that particular failure - prefer `ha_dsmr` where both are available. And a
|
||||
dead P1 meter takes 45 s to reach 0 W commanded (30 s for `meter_max_age_s`
|
||||
to call the reading stale, then 15 s of `stale_input_s` on top), which is
|
||||
`meter_max_age_s` and `stale_input_s` stacking, not either one alone.
|
||||
|
||||
## 0.2.1
|
||||
|
||||
`target_grid_w` (default -10 W): what the meter should rest at. The deadband
|
||||
|
||||
+145
-1
@@ -48,6 +48,150 @@ Use the ESP32's readings rather than the inverter's cloud or dongle sensors:
|
||||
those serve cached values, and a stale reading here ends the maintenance charge
|
||||
phase having charged nothing.
|
||||
|
||||
### P1 meter ingestion
|
||||
|
||||
`meter_entity` above expects one signed sensor, which usually means a template
|
||||
someone wrote by hand. Setting `meter_source` moves the whole derivation into
|
||||
the add-on, where it is done once and tested, and replaces `meter_entity`
|
||||
entirely.
|
||||
|
||||
Which mode you want depends on what your P1 reader publishes, and there are two
|
||||
shapes in the wild:
|
||||
|
||||
- **Two unsigned registers**, consumption and injection, which is what a Belgian
|
||||
P1 read over DSMR gives you → `ha_dsmr`, or `mqtt_p1` for a bridge. The add-on
|
||||
subtracts them.
|
||||
- **One signed figure**, positive = import and negative = export, which is what
|
||||
a HomeWizard P1 gives you (`sensor.p1_meter_active_power`) → `ha_signed`. The
|
||||
add-on splits it. `ha_dsmr` **cannot** read this: it wants two registers and
|
||||
rejects a negative one outright, which is every exporting telegram.
|
||||
|
||||
Either way, do not build the missing shape out of template sensors. The point of
|
||||
`meter_source` is that the sign convention is derived in one tested place rather
|
||||
than in YAML nobody reviews underneath a safety input.
|
||||
|
||||
> ⚠️ **`ha_dsmr` and `mqtt_p1` have never processed a telegram from real
|
||||
> hardware.** No meter in this installation uses either one. Both were written
|
||||
> to the assumption in `specs.md` §5.2 that a Belgian P1 exposes two unsigned
|
||||
> registers, and the meter actually fitted here does not — it is the HomeWizard
|
||||
> P1 that `ha_signed` reads. They are covered by the unit checks in `test_p1.py`
|
||||
> and by an end-to-end test against a fake Home Assistant, and nothing more.
|
||||
>
|
||||
> This is recorded because the realistic way it bites is someone debugging a
|
||||
> meter problem months from now treating those two paths as proven and looking
|
||||
> for the fault elsewhere. If you are the first person to point one at a real
|
||||
> meter, expect to find something, and please update this note when you do.
|
||||
|
||||
| option | default | meaning |
|
||||
|---|---|---|
|
||||
| `meter_source` | `off` | `off` keeps `meter_entity`. `ha_dsmr` subscribes to the DSMR integration over the HA WebSocket; `mqtt_p1` reads a topic; `ha_signed` subscribes to one signed entity over the HA WebSocket |
|
||||
| `meter_phases` | 1 | 1 or 3. Must match the telegram, or every telegram is rejected and logged |
|
||||
| `meter_max_age_s` | 30 | Beyond this the reading is stale and grid power reads as *missing*. On its own it does **not** command 0 W — see the timing note below. It is also the longest a reading is held forward into the 15-minute average |
|
||||
| `meter_mqtt_topic` | | `mqtt_p1` only |
|
||||
| `p1_import_entity` | | The **unsigned** consumption sensor. Do not point this at a signed template |
|
||||
| `p1_export_entity` | | The **unsigned** injection sensor |
|
||||
| `p1_phase_import_entities` | `[]` | L1..L3, in order. Needed for the capacity-tariff peak on a three-phase connection |
|
||||
| `p1_phase_export_entities` | `[]` | L1..L3, in order |
|
||||
| `p1_net_entity` | | `ha_signed` only. The **signed** net-power sensor: `+` import, `-` export |
|
||||
| `p1_phase_net_entities` | `[]` | `ha_signed` only. L1..L3, in order, each signed the same way. Needed for the capacity-tariff peak on a three-phase connection. **The list length must equal `meter_phases`** |
|
||||
|
||||
Both per-phase lists are checked against `meter_phases` **once at startup**: a
|
||||
list of the wrong length disables P1 ingestion with an error in the log, rather
|
||||
than letting every telegram fail its phase-count check one at a time. Leaving
|
||||
the list empty is fine and is not an error — you simply get no per-phase
|
||||
figures, and therefore no capacity-tariff peak. On a three-phase connection
|
||||
that is a much bigger omission than it looks: on a surveyed reading here the
|
||||
phases carried 2769 W of import while the connection netted 187 W, so the
|
||||
billed quantity is understated roughly fifteenfold if the phases are missing.
|
||||
|
||||
#### How long a dead meter takes to reach 0 W
|
||||
|
||||
`meter_max_age_s` and `stale_input_s` **stack**. They are two different clocks
|
||||
and neither one is the whole answer:
|
||||
|
||||
| step | option | default |
|
||||
|---|---|---|
|
||||
| telegrams stop, P1 sample goes stale, grid power starts reading *missing* | `meter_max_age_s` | 30 s |
|
||||
| inputs have been missing long enough for the loop to command 0 W | `stale_input_s` | 15 s |
|
||||
| **total, meter death → 0 W commanded by this add-on** | | **45 s** |
|
||||
|
||||
So in P1 mode `stale_input_s` is *not* "how long inputs may be missing before
|
||||
commanding 0 W" measured from the meter dying — it is measured from the moment
|
||||
the P1 sample already went stale. Size the pair together: the ESP32's own
|
||||
watchdog commands 0 W after ~30 s of silence from this add-on regardless, and
|
||||
that layer is unaffected by either option.
|
||||
|
||||
There is **no fallback to an inverter-side power figure**, deliberately. The
|
||||
inverter's own AC power tracks its battery almost perfectly and the real meter
|
||||
hardly at all, so a controller that failed over to it would be regulating
|
||||
against its own output while looking healthy.
|
||||
|
||||
The `mqtt_p1` payload is one JSON object per telegram, and the schema is strict —
|
||||
a key it does not recognise is a telegram from something other than what was
|
||||
tested, and guessing a key here means guessing a kilowatt:
|
||||
|
||||
```json
|
||||
{"import_w": 1234.0,
|
||||
"export_w": 0.0,
|
||||
"phases": [{"import_w": 500, "export_w": 0},
|
||||
{"import_w": 400, "export_w": 0},
|
||||
{"import_w": 334, "export_w": 0}],
|
||||
"timestamp": "2026-08-24T18:00:05+02:00"}
|
||||
```
|
||||
|
||||
`phases` and `timestamp` are optional; `timestamp` must carry a UTC offset. Where
|
||||
it is present it is used for the age, which is what stops a retained message
|
||||
replayed on reconnect from presenting a ten-minute-old reading as current.
|
||||
|
||||
#### `sensor.p1_sample_age_s`
|
||||
|
||||
Published over MQTT discovery whenever a broker is available: **seconds since the
|
||||
newest accepted telegram**, refreshed every second rather than only when a
|
||||
telegram lands. The ESP32's stale-input watchdog subscribes to this exact entity
|
||||
id, so do not rename it.
|
||||
|
||||
The reason it is recomputed against the clock is that Home Assistant only pushes
|
||||
a state when the state *changes*. A meter sitting at a genuinely constant reading
|
||||
emits nothing, which is indistinguishable — to anything watching the value — from
|
||||
a meter that has died. Watching the age instead separates the two: it climbs when
|
||||
telegrams stop and resets when they arrive, whatever the reading says.
|
||||
|
||||
The entity is only created when `meter_source` is not `off`. With P1 ingestion
|
||||
disabled there is nothing feeding it, and an age sensor climbing with no ingester
|
||||
behind it would trip the firmware watchdog on a system that is working fine.
|
||||
|
||||
> **Known limit, `mqtt_p1`.** The age measures *arrival*. On the MQTT path a
|
||||
> bridge that is stuck republishing its last telegram keeps arriving, so the age
|
||||
> stays near zero and a frozen meter still looks fresh. Detecting *that* needs a
|
||||
> change-detector rather than an arrival-detector, and it is not in this version.
|
||||
|
||||
> ⚠️ **Known limit, `ha_signed` — do not drive a watchdog off this age yet.**
|
||||
> On the HA WebSocket paths the age is stamped when a `state_changed` arrives,
|
||||
> which means it measures *time since the value last changed*, not time since
|
||||
> the meter last reported. Home Assistant offers nothing better: a repeated
|
||||
> reading produces no `state_changed`, does **not** advance `last_reported` on
|
||||
> either the REST or the WebSocket 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 and no
|
||||
> timestamp movement anywhere.
|
||||
>
|
||||
> `ha_dsmr` mostly escapes this because a DSMR telegram updates several entities
|
||||
> and something in the set almost always moves. **`ha_signed` has exactly one
|
||||
> entity, so a healthy meter under a flat load is indistinguishable from a dead
|
||||
> one.** This is not hypothetical: in our own captures
|
||||
> (`sim/scenarios/ha-p1_meter_active_power-2026-08-20.json`) the real house meter
|
||||
> went **42.2 s and 97.0 s** between changes, and 23 Aug peaks at 29.1 s — all
|
||||
> past the default `meter_max_age_s` of 30.
|
||||
>
|
||||
> So `sensor.p1_sample_age_s` on `ha_signed` is safe to *read*, and it is
|
||||
> correct whenever the value is moving, but it must not yet be thresholded by
|
||||
> the ESP32 stale-input watchdog: a quiet house would trip the battery to 0 W.
|
||||
> Raising `meter_max_age_s` is **not** the fix — the two conditions produce an
|
||||
> identical signal, so a bigger number only chooses which of the two errors you
|
||||
> get. The real fix is an arrival stamp the meter itself provides, i.e. reading
|
||||
> the HomeWizard local API directly rather than through an HA entity.
|
||||
|
||||
### Control
|
||||
|
||||
| option | default | meaning |
|
||||
@@ -62,7 +206,7 @@ phase having charged nothing.
|
||||
| `saturation_cycles` | 3 | How many consecutive cycles before freezing. A cycle is one *changed* meter reading, not a fixed period - see the note below. **Do not set to 1** |
|
||||
| `integrator_max_w` | 0 | Bound on the loop's accumulator, and 0 means "same as `max_w`". Caps how much stale error can be waiting to unwind when the sign flips. **Do not raise it above `max_w`** - the output clamp already bounds what is commanded, so the only thing extra headroom buys is more cycles of wrong-direction power after every saturation event. Lowering it below `max_w` is the useful direction |
|
||||
| `heartbeat_s` | 10 | Refresh interval; must stay well under the firmware watchdog |
|
||||
| `stale_input_s` | 15 | How long inputs may be missing before commanding 0 W |
|
||||
| `stale_input_s` | 15 | How long inputs may be missing before commanding 0 W. In P1 mode this clock starts only *after* `meter_max_age_s` has already expired — the two stack, see "How long a dead meter takes to reach 0 W" |
|
||||
| `auto_start` | false | Start controlling on boot (only after commissioning) |
|
||||
|
||||
#### Saturation is counted in cycles, not seconds
|
||||
|
||||
@@ -39,6 +39,7 @@ from .control import Tuning, compute, maintenance_charge_floor, peak_at_risk
|
||||
from .hass import HomeAssistant
|
||||
from .maintenance import IDLE, MaintConfig, Maintenance
|
||||
from .mqtt import MqttPublisher
|
||||
from .p1 import P1Ingest, build_source, is_enabled
|
||||
from . import web
|
||||
|
||||
OPTIONS_PATH = "/data/options.json"
|
||||
@@ -90,6 +91,13 @@ class Controller:
|
||||
store,
|
||||
)
|
||||
|
||||
# P1 ingestion (TEL-01). `meter_source: off` keeps the original
|
||||
# single-entity meter_entity path, so an existing install is unchanged
|
||||
# until it opts in.
|
||||
self.p1 = P1Ingest(phases=int(opts.get("meter_phases", 1)),
|
||||
max_age_s=float(opts.get("meter_max_age_s", 30)))
|
||||
self.p1_enabled = is_enabled(opts)
|
||||
|
||||
# live state
|
||||
self.auto = bool(store.data.get("auto", opts.get("auto_start", False)))
|
||||
self.target = 0.0
|
||||
@@ -121,6 +129,16 @@ class Controller:
|
||||
# -- io ------------------------------------------------------------------
|
||||
async def read_inputs(self) -> None:
|
||||
o = self.o
|
||||
if self.p1_enabled:
|
||||
# ⚠️ P1 is the only authoritative measurement of what the utility
|
||||
# sees (§5.1). When it is stale this is None, which falls into the
|
||||
# existing "inputs missing -> command 0 W" path below. There is
|
||||
# deliberately NO fallback to an inverter-side figure: the
|
||||
# inverter's own AC power correlates 0.998 with battery power and
|
||||
# 0.09 with the real meter, so a controller that failed over to it
|
||||
# would be regulating against its own output.
|
||||
self.grid = self.p1.net_w
|
||||
else:
|
||||
self.grid = await self.hass.number(o.get("meter_entity", ""),
|
||||
bool(o.get("meter_invert")))
|
||||
self.soc = await self.hass.number(o.get("soc_entity", ""))
|
||||
@@ -306,14 +324,26 @@ class Controller:
|
||||
await asyncio.sleep(1)
|
||||
|
||||
def publish(self) -> None:
|
||||
self.mqtt.publish({
|
||||
values = {
|
||||
"setpoint": self.target,
|
||||
"grid": self.grid,
|
||||
"battery": self.batt,
|
||||
"soc": self.soc,
|
||||
"phase": self.maint.phase,
|
||||
"status": "running" if self.auto else "stopped",
|
||||
})
|
||||
}
|
||||
# ⚠️ ONLY when P1 ingestion is actually running. The ESP32's stale-input
|
||||
# watchdog subscribes to sensor.p1_sample_age_s and forces the layer-1
|
||||
# failsafe once it reaches max_age_s. With meter_source off there is no
|
||||
# ingester feeding it, so published_age_s would be time-since-startup
|
||||
# climbing without bound - i.e. every existing install would cross the
|
||||
# threshold within 30 s and pin its inverter at 0 W forever. Publishing
|
||||
# nothing leaves the entity non-existent, which is the status quo and
|
||||
# what has_state() in the firmware is checking for.
|
||||
if self.p1_enabled:
|
||||
# Recomputed here, once a second, on purpose - see P1Ingest.
|
||||
values["p1_age"] = round(self.p1.published_age_s, 1)
|
||||
self.mqtt.publish(values)
|
||||
|
||||
async def shutdown(self) -> None:
|
||||
"""Deterministic wind-down. Do not skip this."""
|
||||
@@ -326,11 +356,27 @@ class Controller:
|
||||
def checks(self) -> list:
|
||||
o = self.o
|
||||
out = []
|
||||
for label, value, entity in (
|
||||
("grid power", self.grid, o.get("meter_entity")),
|
||||
("battery SoC", self.soc, o.get("soc_entity")),
|
||||
("battery power", self.batt, o.get("batt_entity")),
|
||||
):
|
||||
if self.p1_enabled:
|
||||
age = self.p1.published_age_s
|
||||
if self.p1.stale:
|
||||
out.append({"ok": False, "warn": False,
|
||||
"text": f"P1 meter ({o.get('meter_source')}): no reading for "
|
||||
f"{age:.0f} s (limit {self.p1.max_age_s:.0f} s)"
|
||||
+ (f" - last error: {self.p1.last_error}"
|
||||
if self.p1.last_error else "")})
|
||||
else:
|
||||
out.append({"ok": True, "warn": False,
|
||||
"text": f"P1 meter ({o.get('meter_source')}): {self.p1.net_w:g} W, "
|
||||
f"{age:.0f} s old, {self.p1.samples} telegrams, "
|
||||
f"{self.p1.parse_errors} rejected"})
|
||||
rows = [("battery SoC", self.soc, o.get("soc_entity")),
|
||||
("battery power", self.batt, o.get("batt_entity"))]
|
||||
if not self.p1_enabled:
|
||||
# In P1 mode the check above replaces this one; leaving both in
|
||||
# would report "no entity configured" for a meter_entity that is
|
||||
# correctly unused, i.e. a permanent false NOT READY.
|
||||
rows.insert(0, ("grid power", self.grid, o.get("meter_entity")))
|
||||
for label, value, entity in rows:
|
||||
if not entity:
|
||||
out.append({"ok": False, "warn": False, "text": f"{label}: no entity configured"})
|
||||
elif value is None:
|
||||
@@ -457,6 +503,7 @@ async def amain() -> None:
|
||||
# observability, and the battery does not care. Caught broadly and on
|
||||
# purpose: this crashed the add-on once already (paho 1.x vs 2.x) and
|
||||
# took the control loop down with it.
|
||||
broker = None
|
||||
try:
|
||||
broker = await hass.mqtt_service()
|
||||
pub = MqttPublisher(
|
||||
@@ -464,6 +511,7 @@ async def amain() -> None:
|
||||
broker.get("port", 1883) if broker else 1883,
|
||||
broker.get("username") if broker else None,
|
||||
broker.get("password") if broker else None,
|
||||
omit=() if is_enabled(opts) else ("p1_age",),
|
||||
)
|
||||
except Exception as err: # noqa: BLE001
|
||||
_LOG.warning("MQTT unavailable (%s) - continuing without status entities", err)
|
||||
@@ -484,11 +532,22 @@ async def amain() -> None:
|
||||
with contextlib.suppress(NotImplementedError):
|
||||
loop.add_signal_handler(sig, stop.set)
|
||||
|
||||
task = asyncio.create_task(controller.run_control())
|
||||
tasks = [asyncio.create_task(controller.run_control())]
|
||||
|
||||
# P1 ingestion runs as its own long-lived task. ⚠️ It must not be driven
|
||||
# off the control loop: telegrams arrive every ~5 s and the loop would
|
||||
# decimate them, so the 15-minute average - the capacity-tariff billing
|
||||
# unit - would be computed from a fraction of the data.
|
||||
p1_source = build_source(opts, controller.p1, session, broker)
|
||||
if p1_source is not None:
|
||||
tasks.append(asyncio.create_task(p1_source.run()))
|
||||
|
||||
await stop.wait()
|
||||
|
||||
await controller.shutdown()
|
||||
for task in tasks:
|
||||
task.cancel()
|
||||
for task in tasks:
|
||||
with contextlib.suppress(asyncio.CancelledError):
|
||||
await task
|
||||
await runner.cleanup()
|
||||
|
||||
@@ -45,6 +45,14 @@ SENSORS = [
|
||||
("soc", "goodwe_battery_soc", "Battery SoC", "%", "battery", "measurement", None),
|
||||
("phase", "goodwe_maintenance_phase", "Maintenance phase", None, None, None, "mdi:battery-sync"),
|
||||
("status", "goodwe_controller_status", "Controller status", None, None, None, "mdi:heart-pulse"),
|
||||
# ⚠️ This one deliberately breaks the goodwe_ prefix above: the entity id
|
||||
# must be exactly `sensor.p1_sample_age_s`, because SAFETY-01's firmware
|
||||
# watchdog subscribes to that literal id and the ENV-01 simulation rig
|
||||
# asserts on it. Renaming it silently disarms a safety layer. It is seconds
|
||||
# since the newest accepted P1 telegram, republished every second so that a
|
||||
# meter frozen at a constant value still shows a climbing age - which is the
|
||||
# false-trip that this entity exists to remove.
|
||||
("p1_age", "p1_sample_age_s", "P1 sample age", "s", "duration", "measurement", None),
|
||||
]
|
||||
|
||||
BASE = "goodwe_ctl"
|
||||
@@ -52,7 +60,12 @@ AVAILABILITY = f"{BASE}/availability"
|
||||
|
||||
|
||||
class MqttPublisher:
|
||||
def __init__(self, host, port, username=None, password=None):
|
||||
def __init__(self, host, port, username=None, password=None, omit=()):
|
||||
# `omit` drops sensor keys from discovery entirely. ⚠️ Announcing a
|
||||
# sensor that nothing will ever publish to is not harmless here:
|
||||
# p1_sample_age_s is a watchdog input, and an entity that exists but is
|
||||
# never fed is a worse signal than one that does not exist at all.
|
||||
self.omit = set(omit)
|
||||
self.enabled = mqtt is not None and bool(host)
|
||||
self.client = None
|
||||
if not self.enabled:
|
||||
@@ -86,6 +99,8 @@ class MqttPublisher:
|
||||
|
||||
def _announce(self) -> None:
|
||||
for key, object_id, name, unit, dev_class, state_class, icon in SENSORS:
|
||||
if key in self.omit:
|
||||
continue
|
||||
cfg = {
|
||||
"name": name,
|
||||
"object_id": object_id,
|
||||
|
||||
@@ -0,0 +1,862 @@
|
||||
"""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"
|
||||
|
||||
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 | SOURCE_HA_SIGNED
|
||||
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
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# 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_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)))
|
||||
return None
|
||||
@@ -1,5 +1,5 @@
|
||||
name: GoodWe RS485 Controller
|
||||
version: "0.2.1"
|
||||
version: "0.3.0"
|
||||
slug: goodwe_controller
|
||||
description: >-
|
||||
Drives a GoodWe ES/BP battery inverter over RS485 by emulating its smart
|
||||
@@ -39,6 +39,34 @@ options:
|
||||
batt_invert: false
|
||||
setpoint_entity: ""
|
||||
|
||||
# --- P1 meter ingestion (specs §5.2 / §14 `meter:`) -------------------------
|
||||
# `off` keeps the original single meter_entity path above, so an existing
|
||||
# install is untouched until it opts in. ha_dsmr subscribes to the DSMR
|
||||
# integration's entities over the HA WebSocket; mqtt_p1 reads the topic below;
|
||||
# ha_signed reads ONE signed HA entity (+ import / - export), which is what a
|
||||
# HomeWizard P1 publishes and what ha_dsmr cannot consume.
|
||||
meter_source: "off"
|
||||
meter_phases: 1
|
||||
meter_max_age_s: 30
|
||||
meter_mqtt_topic: ""
|
||||
# The two UNSIGNED Belgian registers. The EMS derives net power from them
|
||||
# (import - export); do NOT point these at a signed template sensor.
|
||||
p1_import_entity: ""
|
||||
p1_export_entity: ""
|
||||
# Optional, in L1..L3 order. Required for the capacity-tariff peak on a
|
||||
# three-phase connection; the list length must equal meter_phases.
|
||||
p1_phase_import_entities: []
|
||||
p1_phase_export_entities: []
|
||||
# ha_signed only. ONE signed net-power sensor: positive = import from the
|
||||
# grid, negative = export to it. Do NOT split it into two template sensors -
|
||||
# the split is done in the add-on (p1.split_signed) precisely so the sign
|
||||
# convention is tested rather than living in unreviewed YAML.
|
||||
p1_net_entity: ""
|
||||
# Optional, in L1..L3 order, each one signed the same way. Same role as
|
||||
# p1_phase_import_entities: the capacity-tariff peak on a three-phase
|
||||
# connection. The list length must equal meter_phases.
|
||||
p1_phase_net_entities: []
|
||||
|
||||
# --- control ---------------------------------------------------------------
|
||||
max_w: 2000
|
||||
gain: 0.6
|
||||
@@ -81,6 +109,24 @@ schema:
|
||||
batt_invert: bool
|
||||
setpoint_entity: str
|
||||
|
||||
meter_source: list(off|ha_dsmr|mqtt_p1|ha_signed)
|
||||
# ⚠️ 2 is accepted by this range but is not a real Belgian connection. A
|
||||
# telegram whose phase count disagrees is rejected at ingest and logged, so a
|
||||
# mis-set 2 shows up immediately as "0 telegrams accepted" rather than as a
|
||||
# quietly wrong number.
|
||||
meter_phases: int(1,3)
|
||||
meter_max_age_s: int(5,300)
|
||||
meter_mqtt_topic: str?
|
||||
p1_import_entity: str?
|
||||
p1_export_entity: str?
|
||||
p1_phase_import_entities:
|
||||
- str
|
||||
p1_phase_export_entities:
|
||||
- str
|
||||
p1_net_entity: str?
|
||||
p1_phase_net_entities:
|
||||
- str
|
||||
|
||||
max_w: int(100,5000)
|
||||
gain: float(0.05,1.0)
|
||||
slew_w: int(50,5000)
|
||||
|
||||
@@ -0,0 +1,965 @@
|
||||
"""Runnable check for P1 ingestion. `python3 test_p1.py`
|
||||
|
||||
No framework, no fixtures, no meter - it has to run on a tech's laptop and in CI
|
||||
with nothing installed and nothing plugged in (§17: "usable without real
|
||||
hardware").
|
||||
|
||||
Every assert here is a rule whose absence poisons something downstream: an
|
||||
inverted sign inverts the control loop, a fabricated zero hides a dead meter, a
|
||||
naive per-phase sum misbills the capacity tariff, and a plain mean of samples
|
||||
computes the wrong quarter-hour figure whenever the telegram cadence changes.
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
import sys
|
||||
import time
|
||||
from datetime import datetime, timedelta, timezone
|
||||
|
||||
import aiohttp # already required by app.p1, so this adds no new dependency
|
||||
|
||||
from app.p1 import (
|
||||
P1Error, P1Ingest, HaDsmrSource, HaSignedSource, QuarterAverager,
|
||||
SOURCE_HA, SOURCE_HA_SIGNED, SOURCE_MQTT,
|
||||
build_source, is_enabled, make_sample, 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("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)
|
||||
|
||||
# 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))
|
||||
|
||||
print()
|
||||
if fails:
|
||||
print(f"{len(fails)} of {total} FAILED: {', '.join(fails)}")
|
||||
sys.exit(1)
|
||||
print(f"{total} checks")
|
||||
print("all checks passed")
|
||||
Reference in New Issue
Block a user