From 99ffa00c9927e219d311e2796fd4601619d2a36a Mon Sep 17 00:00:00 2001 From: tomquist <528585+tomquist@users.noreply.github.com> Date: Sun, 27 Sep 2026 22:21:57 +0000 Subject: [PATCH 1/7] Let passive readers reuse the control loop's meter reading The Marstek MQTT responder and cloud reporting only report the grid reading, yet each of their polls read a polled meter (Shelly, HTTP, Modbus, ...) again on top of the batteries' own polls. A new innermost LastReadingPowermeter remembers every real read of the source, and the raw read those two use serves it while it is at most 5 s (plus the throttle interval) old, reading the source only when no battery has. Push meters already answer from their cache and pass straight through. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01Vv6VhaqsWHbc7sTdpnfNLw --- CHANGELOG.md | 2 + CONTRIBUTING.md | 2 +- src/astrameter/config/addon_test.py | 12 +- src/astrameter/config/config_loader.py | 11 ++ src/astrameter/main.py | 6 +- src/astrameter/main_test.py | 61 +++++++- src/astrameter/meter_pool.py | 13 +- src/astrameter/powermeter/base.py | 10 +- .../powermeter/wrappers/__init__.py | 3 + src/astrameter/powermeter/wrappers/health.py | 6 +- .../powermeter/wrappers/health_test.py | 12 ++ .../powermeter/wrappers/last_reading.py | 58 ++++++++ .../powermeter/wrappers/last_reading_test.py | 131 ++++++++++++++++++ 13 files changed, 306 insertions(+), 21 deletions(-) create mode 100644 src/astrameter/powermeter/wrappers/last_reading.py create mode 100644 src/astrameter/powermeter/wrappers/last_reading_test.py diff --git a/CHANGELOG.md b/CHANGELOG.md index da5d916a..311748ca 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,8 @@ ## Next +- **Fixed** the Marstek app and cloud reporting sending extra requests to a polled power meter, such as a Shelly or an HTTP source, on top of the batteries' own polls; they now reuse the batteries' latest reading. + - **Fixed** a battery's Home Assistant entities staying unavailable until a restart since 2.3.1, after a network hiccup briefly dropped the battery and the MQTT connection at the same time ([#718](https://github.com/tomquist/astrameter/issues/718), [#719](https://github.com/tomquist/astrameter/pull/719)). - **Fixed** a battery sometimes staying stuck below its share since 2.3.1 while the other carried the rest, letting their charge levels drift apart; a battery that only paused briefly is now back to sharing within a minute ([#704](https://github.com/tomquist/astrameter/issues/704), [#707](https://github.com/tomquist/astrameter/pull/707)). diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index fa5ab076..8e94ec09 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -73,7 +73,7 @@ When you fix a bug in: `split_balancer_knobs` in `balancer.py` is Python-only too — it sorts the flat tuning knobs a scenario file or a `--set` override names into a `BalancerConfig`, and the firmware takes its configuration from codegen instead. -Fixes to `src/astrameter/powermeter/wrappers/{transform,throttling}.py` have **no** C++ counterpart — those wrappers are delegated to ESPHome's standard `sensor: filters:` (`offset:`, `multiply:`, `throttle:`) on the upstream sensor. `src/astrameter/powermeter/wrappers/health.py` (the outermost `HealthTrackingPowermeter` feeding the MQTT Insights Online sensor) likewise has **no** C++ counterpart — it tracks Python powermeter reads, which the firmware doesn't have. +Fixes to `src/astrameter/powermeter/wrappers/{transform,throttling}.py` have **no** C++ counterpart — those wrappers are delegated to ESPHome's standard `sensor: filters:` (`offset:`, `multiply:`, `throttle:`) on the upstream sensor. `src/astrameter/powermeter/wrappers/health.py` (the outermost `HealthTrackingPowermeter` feeding the MQTT Insights Online sensor) likewise has **no** C++ counterpart — it tracks Python powermeter reads, which the firmware doesn't have. Nor does `src/astrameter/powermeter/wrappers/last_reading.py` (the innermost `LastReadingPowermeter`, which lets the Marstek responder and cloud reporting reuse the control loop's read of a polled meter): the firmware's passive readers already read its sensor cache (`CT002Component::latest_grid_power()`). The `power_sensor_lX` unit handling (unit→W auto-conversion, non-power-unit rejection, and the kW-suspicion warning — issue #572) is **ESPHome-only**: it lives in `ct002/__init__.py` (`FINAL_VALIDATE_SCHEMA` + `set_power_unit_scale` codegen) and the sensor-callback cache in `ct002.cpp`, with no counterpart in `src/astrameter/ct002/` — the Python stack receives watts from its powermeter layer, where the equivalent unit handling is implemented per-source (currently `powermeter/homeassistant.py`, which reads the entity's `unit_of_measurement` attribute). Keep the two accepted-unit tables (`POWER_UNIT_SCALES` in `ct002/__init__.py`, `POWER_UNIT_SCALE` in `src/astrameter/power_units.py`) in sync — the Python side keeps its copy in that leaf module because the dashboard's sensor picker reads it too, and drift between the two is what hid working sensors from the form. diff --git a/src/astrameter/config/addon_test.py b/src/astrameter/config/addon_test.py index 89d572a4..167fc4d7 100644 --- a/src/astrameter/config/addon_test.py +++ b/src/astrameter/config/addon_test.py @@ -12,6 +12,7 @@ from astrameter.powermeter import HomeAssistant, ThrottledPowermeter from astrameter.powermeter.wrappers.base import PowermeterWrapper from astrameter.powermeter.wrappers.health import HealthTrackingPowermeter +from astrameter.powermeter.wrappers.last_reading import LastReadingPowermeter class FakeSupervisor(addon.SupervisorClient): @@ -174,7 +175,9 @@ def test_single_power_entity_is_read_directly() -> None: cfg = config(BASE_OPTIONS) meter = cfg.powermeters(cfg.general())[0][0] assert isinstance(meter, HealthTrackingPowermeter) - source = meter.wrapped_powermeter + shared = meter.wrapped_powermeter + assert isinstance(shared, LastReadingPowermeter) + source = shared.wrapped_powermeter assert isinstance(source, HomeAssistant) assert source.power_calculate is False assert source.current_power_entity == ["sensor.current_power_in"] @@ -183,10 +186,12 @@ def test_single_power_entity_is_read_directly() -> None: def _inner_source(cfg: addon.AddonAppConfig) -> HomeAssistant: - """The Home Assistant source under the health wrapper the loader adds.""" + """The Home Assistant source under the wrappers the loader always adds.""" meter = cfg.powermeters(cfg.general())[0][0] assert isinstance(meter, HealthTrackingPowermeter) - source = meter.wrapped_powermeter + shared = meter.wrapped_powermeter + assert isinstance(shared, LastReadingPowermeter) + source = shared.wrapped_powermeter assert isinstance(source, HomeAssistant) return source @@ -244,6 +249,7 @@ def test_power_source_is_conditioned_by_the_options() -> None: "HampelPowermeter", "ThrottledPowermeter", "TransformedPowermeter", + "LastReadingPowermeter", ] diff --git a/src/astrameter/config/config_loader.py b/src/astrameter/config/config_loader.py index 94de27bd..765f586a 100644 --- a/src/astrameter/config/config_loader.py +++ b/src/astrameter/config/config_loader.py @@ -21,6 +21,10 @@ from astrameter.powermeter.base import Powermeter from astrameter.powermeter.wrappers.hampel import HampelPowermeter from astrameter.powermeter.wrappers.health import HealthTrackingPowermeter +from astrameter.powermeter.wrappers.last_reading import ( + PASSIVE_READING_MAX_AGE_S, + LastReadingPowermeter, +) from astrameter.powermeter.wrappers.pid import PidPowermeter from astrameter.powermeter.wrappers.smoothing import ( DeadbandPowermeter, @@ -355,6 +359,13 @@ def apply_signal_wrappers( Shared by every config backend, so a power source behaves the same however it was configured. *name* labels it in logs and MQTT Insights. """ + # Innermost, so passive readers see the source's own reading. The throttle + # above it lets a control read through at most once per interval, so the + # reading may be that much older before a passive reader reads for itself. + powermeter = LastReadingPowermeter( + powermeter, max_age=PASSIVE_READING_MAX_AGE_S + signal.throttle_interval + ) + if signal.offsets is not None or signal.multipliers is not None: offsets = signal.offsets if signal.offsets is not None else [0.0] multipliers = signal.multipliers if signal.multipliers is not None else [1.0] diff --git a/src/astrameter/main.py b/src/astrameter/main.py index 3b3fd82d..e134f397 100644 --- a/src/astrameter/main.py +++ b/src/astrameter/main.py @@ -68,7 +68,11 @@ async def read_ct_powermeter( async def _read_grid_phases(powermeters: list[ConfiguredPowermeter]) -> list[float]: """Raw three-phase reading from the meter that serves every client (or the first one), waiting briefly for a fresh push so an idle meter cannot pin - the caller.""" + the caller. + + Its callers only report the reading, so it reuses the one the batteries' + polls last fetched and reads a polled meter only when none did recently. + """ chosen = next( (c.powermeter for c in powermeters if c.client_filter.matches("0.0.0.0")), None ) diff --git a/src/astrameter/main_test.py b/src/astrameter/main_test.py index d865cfd5..9d94d7b6 100644 --- a/src/astrameter/main_test.py +++ b/src/astrameter/main_test.py @@ -1,10 +1,19 @@ import argparse +from dataclasses import replace from ipaddress import IPv4Network -from astrameter.config.config_loader import ClientFilter, new_config_parser +from astrameter.config.config_loader import ( + ClientFilter, + apply_signal_wrappers, + new_config_parser, +) from astrameter.config.ini_config import IniAppConfig -from astrameter.config.settings import ConfiguredPowermeter -from astrameter.main import _resolve_device_config, read_ct_powermeter +from astrameter.config.settings import ConfiguredPowermeter, SignalSettings +from astrameter.main import ( + _read_grid_phases, + _resolve_device_config, + read_ct_powermeter, +) from astrameter.powermeter import Powermeter @@ -87,6 +96,52 @@ async def test_read_ct_powermeter_swallows_timeout_and_serves_cached() -> None: assert result == [11.0, 22.0, 33.0] +class _PolledMeter(Powermeter): + """A pull source that counts how often it is actually read.""" + + def __init__(self, values: list[float]) -> None: + self.values = values + self.reads = 0 + + async def get_powermeter_watts(self) -> list[float]: + self.reads += 1 + return list(self.values) + + +_EVERYONE = ClientFilter([IPv4Network("0.0.0.0/0")]) + + +def _configured(source: Powermeter) -> list[ConfiguredPowermeter]: + """*source* behind the wrappers the config loader builds, smoothing on.""" + signal = replace(SignalSettings(), smooth_alpha=0.5) + return [ + ConfiguredPowermeter( + apply_signal_wrappers(source, "SCRIPT_1", signal), _EVERYONE, False + ) + ] + + +async def test_grid_phases_reuse_the_control_loops_read() -> None: + """The Marstek responder and cloud reporting only report the reading, so + while a battery polls the meter they must not read it again.""" + source = _PolledMeter([100.0, 200.0, 300.0]) + powermeters = _configured(source) + + await read_ct_powermeter(("10.0.0.5", 0), powermeters) + source.values = [0.0, 0.0, 0.0] + + assert await _read_grid_phases(powermeters) == [100.0, 200.0, 300.0] + assert source.reads == 1 + + +async def test_grid_phases_read_an_idle_meter_themselves() -> None: + source = _PolledMeter([100.0, 200.0, 300.0]) + powermeters = _configured(source) + + assert await _read_grid_phases(powermeters) == [100.0, 200.0, 300.0] + assert source.reads == 1 + + def _resolve(device_type: str) -> tuple[list[str], list[str]]: cfg = new_config_parser() cfg.add_section("GENERAL") diff --git a/src/astrameter/meter_pool.py b/src/astrameter/meter_pool.py index a88608b4..877a4e4f 100644 --- a/src/astrameter/meter_pool.py +++ b/src/astrameter/meter_pool.py @@ -12,7 +12,7 @@ from astrameter.config.logger import logger from astrameter.config.settings import ConfiguredPowermeter from astrameter.powermeter import Powermeter -from astrameter.powermeter.wrappers.health import HealthTrackingPowermeter +from astrameter.powermeter.wrappers.base import PowermeterWrapper #: How long a poll waits for a pushed reading before serving the last one. #: Batteries poll about once a second, so a longer wait would answer after the @@ -21,13 +21,10 @@ def powermeter_name(powermeter: Powermeter) -> str: - """The meter class behind the health wrapper every configured meter gets.""" - inner = ( - powermeter.wrapped_powermeter - if isinstance(powermeter, HealthTrackingPowermeter) - else powermeter - ) - return type(inner).__name__ + """The source class under the wrappers every configured meter gets.""" + while isinstance(powermeter, PowermeterWrapper): + powermeter = powermeter.wrapped_powermeter + return type(powermeter).__name__ def powermeter_for( diff --git a/src/astrameter/powermeter/base.py b/src/astrameter/powermeter/base.py index 3b8ffd60..e1edf0ff 100644 --- a/src/astrameter/powermeter/base.py +++ b/src/astrameter/powermeter/base.py @@ -40,10 +40,12 @@ async def get_powermeter_watts(self) -> list[float]: ... async def get_powermeter_watts_raw(self) -> list[float]: """Per-phase watts before section/global processing wrappers. - Used when a consumer (e.g. Marstek MQTT display) should match the physical - meter while control still uses :meth:`get_powermeter_watts`. Defaults to - the same values as :meth:`get_powermeter_watts` for sources with no inner - pipeline. + Used by the passive readers (the Marstek MQTT display, cloud reporting), + which should match the physical meter while control still uses + :meth:`get_powermeter_watts`, and must not add load on it: the + ``LastReadingPowermeter`` the config loader puts around every source + answers from the control loop's recent read. Defaults to the same values + as :meth:`get_powermeter_watts` for sources with no inner pipeline. """ return await self.get_powermeter_watts() diff --git a/src/astrameter/powermeter/wrappers/__init__.py b/src/astrameter/powermeter/wrappers/__init__.py index 02c59644..d61a0aec 100644 --- a/src/astrameter/powermeter/wrappers/__init__.py +++ b/src/astrameter/powermeter/wrappers/__init__.py @@ -1,15 +1,18 @@ from .base import PowermeterWrapper from .hampel import HampelPowermeter from .health import HealthTrackingPowermeter +from .last_reading import PASSIVE_READING_MAX_AGE_S, LastReadingPowermeter from .pid import PidPowermeter from .smoothing import DeadbandPowermeter, SmoothedPowermeter from .throttling import ThrottledPowermeter from .transform import TransformedPowermeter __all__ = [ + "PASSIVE_READING_MAX_AGE_S", "DeadbandPowermeter", "HampelPowermeter", "HealthTrackingPowermeter", + "LastReadingPowermeter", "PidPowermeter", "PowermeterWrapper", "SmoothedPowermeter", diff --git a/src/astrameter/powermeter/wrappers/health.py b/src/astrameter/powermeter/wrappers/health.py index a0ea01bd..b8b08a8a 100644 --- a/src/astrameter/powermeter/wrappers/health.py +++ b/src/astrameter/powermeter/wrappers/health.py @@ -5,6 +5,7 @@ from astrameter.powermeter.base import Powermeter from .base import PowermeterWrapper +from .last_reading import LastReadingPowermeter @dataclasses.dataclass(frozen=True, slots=True) @@ -85,7 +86,10 @@ def status_snapshot(self) -> PowermeterHealth: pipeline: list[str] = [] node: Powermeter = self.wrapped_powermeter while isinstance(node, PowermeterWrapper): - pipeline.append(type(node).__name__) + # Not a filter: it only shares the source's reading with passive + # readers, so the dashboard's pipeline leaves it out. + if not isinstance(node, LastReadingPowermeter): + pipeline.append(type(node).__name__) node = node.wrapped_powermeter return PowermeterHealth( name=self.name, diff --git a/src/astrameter/powermeter/wrappers/health_test.py b/src/astrameter/powermeter/wrappers/health_test.py index 042772ff..965217a2 100644 --- a/src/astrameter/powermeter/wrappers/health_test.py +++ b/src/astrameter/powermeter/wrappers/health_test.py @@ -5,7 +5,10 @@ from astrameter.powermeter.base import Powermeter +from .conftest import FakePowermeter from .health import HealthTrackingPowermeter +from .last_reading import LastReadingPowermeter +from .throttling import ThrottledPowermeter class _FakeClock: @@ -104,3 +107,12 @@ async def test_lifecycle_delegates_to_inner() -> None: await pm.stop() inner.start.assert_awaited_once() inner.stop.assert_awaited_once() + + +def test_pipeline_leaves_out_the_shared_reading() -> None: + """The dashboard's Filters row lists what shapes the reading; the wrapper + that only shares it with passive readers is not one of those.""" + pm = _make(ThrottledPowermeter(LastReadingPowermeter(FakePowermeter()), 1.0)) + snapshot = pm.status_snapshot() + assert snapshot.kind == "FakePowermeter" + assert snapshot.pipeline == ("ThrottledPowermeter",) diff --git a/src/astrameter/powermeter/wrappers/last_reading.py b/src/astrameter/powermeter/wrappers/last_reading.py new file mode 100644 index 00000000..657a77f8 --- /dev/null +++ b/src/astrameter/powermeter/wrappers/last_reading.py @@ -0,0 +1,58 @@ +import time +from collections.abc import Callable + +from astrameter.powermeter.base import Powermeter, stream_fresh + +from .base import PowermeterWrapper + +#: How old the control loop's last reading may be before a passive reader +#: (the Marstek app responder, cloud reporting) reads the source itself. +#: Batteries poll about once a second, so while any battery reads this meter +#: a passive reader never adds a request of its own. +PASSIVE_READING_MAX_AGE_S = 5.0 + + +class LastReadingPowermeter(PowermeterWrapper): + """Innermost wrapper that lets passive readers reuse the control loop's read. + + Every real read of the source passes through here and is remembered. + :meth:`get_powermeter_watts_raw` — what the diagnostic readers call — + serves that reading while it is at most ``max_age`` seconds old, and only + reads the source itself when nothing has read it for longer, so those + readers add no load on a meter the batteries are already polling. + + A push meter already answers from the reading it was sent, with no I/O, so + its raw reads go straight through and keep the meter's own staleness rule. + + ``reset()`` passes through without forgetting the reading: it clears the + filters' rolling state, and this is a fact about the meter. + """ + + def __init__( + self, + wrapped_powermeter: Powermeter, + *, + max_age: float = PASSIVE_READING_MAX_AGE_S, + clock: Callable[[], float] | None = None, + ) -> None: + super().__init__(wrapped_powermeter) + self.max_age = max_age + self._clock = clock or time.monotonic + self._values: list[float] | None = None + self._read_at: float | None = None + + async def get_powermeter_watts(self) -> list[float]: + values = await self.wrapped_powermeter.get_powermeter_watts() + if values: + self._values = list(values) + self._read_at = self._clock() + return values + + async def get_powermeter_watts_raw(self) -> list[float]: + if self.wrapped_powermeter.stream_online() is not None: + return await self.wrapped_powermeter.get_powermeter_watts_raw() + if self._values is not None and stream_fresh( + self._read_at, self.max_age, self._clock + ): + return list(self._values) + return await self.get_powermeter_watts() diff --git a/src/astrameter/powermeter/wrappers/last_reading_test.py b/src/astrameter/powermeter/wrappers/last_reading_test.py new file mode 100644 index 00000000..5ffa0eef --- /dev/null +++ b/src/astrameter/powermeter/wrappers/last_reading_test.py @@ -0,0 +1,131 @@ +import pytest + +from .conftest import FakePowermeter +from .last_reading import LastReadingPowermeter + + +class _FakeClock: + def __init__(self) -> None: + self.now = 0.0 + + def __call__(self) -> float: + return self.now + + +class _CountingSource(FakePowermeter): + """A polled source that counts how often it is actually read.""" + + def __init__(self, values: list[float]) -> None: + super().__init__(values) + self.reads = 0 + self.error: Exception | None = None + + async def get_powermeter_watts(self) -> list[float]: + self.reads += 1 + if self.error is not None: + raise self.error + return await super().get_powermeter_watts() + + +class _PushSource(_CountingSource): + def stream_online(self) -> bool | None: + return True + + async def get_powermeter_watts_raw(self) -> list[float]: + return await self.get_powermeter_watts() + + +def _make( + source: FakePowermeter, max_age: float = 5.0 +) -> tuple[LastReadingPowermeter, _FakeClock]: + clock = _FakeClock() + return LastReadingPowermeter(source, max_age=max_age, clock=clock), clock + + +async def test_raw_read_reuses_a_recent_control_read() -> None: + source = _CountingSource([100.0, 200.0]) + pm, clock = _make(source) + + assert await pm.get_powermeter_watts() == [100.0, 200.0] + source.set([999.0, 999.0]) + clock.now = 5.0 + + assert await pm.get_powermeter_watts_raw() == [100.0, 200.0] + assert source.reads == 1 + + +async def test_raw_read_reads_the_source_once_the_reading_is_too_old() -> None: + source = _CountingSource([100.0]) + pm, clock = _make(source) + await pm.get_powermeter_watts() + + source.set([150.0]) + clock.now = 5.1 + assert await pm.get_powermeter_watts_raw() == [150.0] + assert source.reads == 2 + + # That read is remembered, so the next passive read within max_age is free. + clock.now = 6.0 + assert await pm.get_powermeter_watts_raw() == [150.0] + assert source.reads == 2 + + +async def test_raw_read_reads_the_source_when_nothing_has_read_it() -> None: + source = _CountingSource([42.0]) + pm, _ = _make(source) + + assert await pm.get_powermeter_watts_raw() == [42.0] + assert source.reads == 1 + + +async def test_failed_or_empty_reads_are_not_remembered() -> None: + source = _CountingSource([100.0]) + pm, clock = _make(source) + await pm.get_powermeter_watts() + + clock.now = 10.0 + source.error = ValueError("offline") + with pytest.raises(ValueError): + await pm.get_powermeter_watts() + # The raw read does not fall back to the old reading: it asks the source. + with pytest.raises(ValueError): + await pm.get_powermeter_watts_raw() + + source.error = None + source.set([]) + assert await pm.get_powermeter_watts() == [] + source.set([7.0]) + assert await pm.get_powermeter_watts_raw() == [7.0] + + +async def test_served_reading_is_a_copy() -> None: + source = _CountingSource([1.0, 2.0]) + pm, _ = _make(source) + await pm.get_powermeter_watts() + + served = await pm.get_powermeter_watts_raw() + served[0] = 999.0 + assert await pm.get_powermeter_watts_raw() == [1.0, 2.0] + + +async def test_push_meter_raw_read_goes_straight_to_its_cache() -> None: + """A push meter answers without I/O and applies its own staleness rule, so + the wrapper must not serve a reading the meter itself would now refuse.""" + source = _PushSource([100.0]) + pm, _ = _make(source) + await pm.get_powermeter_watts() + + source.error = ValueError("stale") + with pytest.raises(ValueError, match="stale"): + await pm.get_powermeter_watts_raw() + + +async def test_reset_keeps_the_reading() -> None: + source = _CountingSource([100.0]) + pm, _ = _make(source) + await pm.get_powermeter_watts() + + pm.reset() + assert source.reset_count == 1 + assert await pm.get_powermeter_watts_raw() == [100.0] + assert source.reads == 1 From ce932921dd9e51ccc71ece4ddbd7341b64f0b057 Mon Sep 17 00:00:00 2001 From: tomquist <528585+tomquist@users.noreply.github.com> Date: Sun, 27 Sep 2026 22:22:21 +0000 Subject: [PATCH 2/7] Link the changelog entry to its pull request Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01Vv6VhaqsWHbc7sTdpnfNLw --- CHANGELOG.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 311748ca..65c8fb21 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,7 +2,7 @@ ## Next -- **Fixed** the Marstek app and cloud reporting sending extra requests to a polled power meter, such as a Shelly or an HTTP source, on top of the batteries' own polls; they now reuse the batteries' latest reading. +- **Fixed** the Marstek app and cloud reporting sending extra requests to a polled power meter, such as a Shelly or an HTTP source, on top of the batteries' own polls; they now reuse the batteries' latest reading ([#698](https://github.com/tomquist/astrameter/pull/698)). - **Fixed** a battery's Home Assistant entities staying unavailable until a restart since 2.3.1, after a network hiccup briefly dropped the battery and the MQTT connection at the same time ([#718](https://github.com/tomquist/astrameter/issues/718), [#719](https://github.com/tomquist/astrameter/pull/719)). From 441aba34bf418960c455c6f2273a2c99b5728759 Mon Sep 17 00:00:00 2001 From: tomquist <528585+tomquist@users.noreply.github.com> Date: Mon, 28 Sep 2026 06:33:20 +0000 Subject: [PATCH 3/7] Give the passive readers an explicit route The reuse of the control loop's reading hid behind get_powermeter_watts_raw, so nothing at the Marstek responder or cloud reporting said their reads were passive. Rename it get_passive_watts, and _read_grid_phases to _passive_grid_phases, and document the route on the base class. The throttle's override only restated the base wrapper's delegation, so it goes. The unused C++ get_powermeter_watts_raw is deleted to keep the mirrored symbols in step. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01Vv6VhaqsWHbc7sTdpnfNLw --- esphome/components/ct002/sensor_backed.h | 1 - esphome/components/ct002/wrapper_base.h | 10 --------- src/astrameter/main.py | 22 ++++++++++--------- src/astrameter/main_test.py | 10 ++++----- src/astrameter/powermeter/base.py | 16 ++++++++------ src/astrameter/powermeter/wrappers/base.py | 4 ++-- .../powermeter/wrappers/conftest.py | 2 +- src/astrameter/powermeter/wrappers/health.py | 4 ++-- .../powermeter/wrappers/health_test.py | 6 ++--- .../powermeter/wrappers/last_reading.py | 9 ++++---- .../powermeter/wrappers/last_reading_test.py | 22 +++++++++---------- .../powermeter/wrappers/pid_test.py | 2 +- .../powermeter/wrappers/throttling.py | 5 ----- .../powermeter/wrappers/throttling_test.py | 16 +++++++------- .../powermeter/wrappers/transform_test.py | 8 +++---- src/astrameter/shelly/shelly_udp_test.py | 2 +- 16 files changed, 64 insertions(+), 75 deletions(-) diff --git a/esphome/components/ct002/sensor_backed.h b/esphome/components/ct002/sensor_backed.h index 35d3a48e..1a12c232 100644 --- a/esphome/components/ct002/sensor_backed.h +++ b/esphome/components/ct002/sensor_backed.h @@ -24,7 +24,6 @@ class SensorBackedPowermeter : public Powermeter { uint32_t max_sensor_age_ms); std::vector get_powermeter_watts() override; - std::vector get_powermeter_watts_raw() override { return this->get_powermeter_watts(); } private: uint8_t num_phases_; diff --git a/esphome/components/ct002/wrapper_base.h b/esphome/components/ct002/wrapper_base.h index 1fbaf0da..bc9d5222 100644 --- a/esphome/components/ct002/wrapper_base.h +++ b/esphome/components/ct002/wrapper_base.h @@ -17,13 +17,6 @@ class Powermeter { // 3-element (three-phase) vectors; an empty vector signals "unavailable". virtual std::vector get_powermeter_watts() = 0; - // Raw (unfiltered) per-phase readings — used by mqtt_insights to publish - // the unsmoothed signal alongside the filtered one. The base wrapper - // delegates to upstream's raw, so the call chain unwinds to the source. - virtual std::vector get_powermeter_watts_raw() { - return this->get_powermeter_watts(); - } - // Clears any time-windowed state. Called by CT002 after long gaps where // accumulated state would be stale (e.g. Wi-Fi reconnect, sensor outage). virtual void reset() {} @@ -36,9 +29,6 @@ class PowermeterWrapper : public Powermeter { public: explicit PowermeterWrapper(Powermeter *wrapped) : wrapped_(wrapped) {} - std::vector get_powermeter_watts_raw() override { - return this->wrapped_->get_powermeter_watts_raw(); - } void reset() override { this->wrapped_->reset(); } protected: diff --git a/src/astrameter/main.py b/src/astrameter/main.py index e134f397..9a45a129 100644 --- a/src/astrameter/main.py +++ b/src/astrameter/main.py @@ -65,13 +65,15 @@ async def read_ct_powermeter( return three_phases(await read_fresh(configured)) -async def _read_grid_phases(powermeters: list[ConfiguredPowermeter]) -> list[float]: - """Raw three-phase reading from the meter that serves every client (or the - first one), waiting briefly for a fresh push so an idle meter cannot pin - the caller. - - Its callers only report the reading, so it reuses the one the batteries' - polls last fetched and reads a polled meter only when none did recently. +async def _passive_grid_phases(powermeters: list[ConfiguredPowermeter]) -> list[float]: + """Three-phase reading over the passive route, for callers that only + report it (the Marstek responder, cloud reporting). + + Reads the meter that serves every client (or the first one) through + :meth:`~astrameter.powermeter.Powermeter.get_passive_watts`, so it reuses + the reading the batteries' polls last fetched and reads a polled meter only + when none did recently. Waits briefly for a fresh push first, so an idle + push meter cannot pin the caller. """ chosen = next( (c.powermeter for c in powermeters if c.client_filter.matches("0.0.0.0")), None @@ -84,7 +86,7 @@ async def _read_grid_phases(powermeters: list[ConfiguredPowermeter]) -> list[flo await asyncio.wait_for( chosen.wait_for_next_message(), timeout=FRESH_READING_TIMEOUT_S ) - return [float(v) for v in three_phases(await chosen.get_powermeter_watts_raw())] + return [float(v) for v in three_phases(await chosen.get_passive_watts())] async def check_powermeter(powermeter: Powermeter, client_filter: ClientFilter) -> None: @@ -273,7 +275,7 @@ async def _bind_marstek_responder( device_id=device_id, ct_type=device.ct_type, mac=marstek_mac, - get_values=lambda: _read_grid_phases(powermeters), + get_values=lambda: _passive_grid_phases(powermeters), get_connected_slave_count=device.reporting_consumer_count, get_cd4_slave_csv=lambda: format_cd4_slave_csv( device.reporting_consumer_rows() @@ -346,7 +348,7 @@ def _start_cloud_reporting( async def gather() -> CtMeasurement: return _ct_measurement( - device, await _read_grid_phases(powermeters), mqtt_connected + device, await _passive_grid_phases(powermeters), mqtt_connected ) reporter = CloudReporter( diff --git a/src/astrameter/main_test.py b/src/astrameter/main_test.py index 9d94d7b6..30aeab14 100644 --- a/src/astrameter/main_test.py +++ b/src/astrameter/main_test.py @@ -10,7 +10,7 @@ from astrameter.config.ini_config import IniAppConfig from astrameter.config.settings import ConfiguredPowermeter, SignalSettings from astrameter.main import ( - _read_grid_phases, + _passive_grid_phases, _resolve_device_config, read_ct_powermeter, ) @@ -33,7 +33,7 @@ def __init__( async def get_powermeter_watts(self) -> list[float]: return list(self._values) - async def get_powermeter_watts_raw(self) -> list[float]: + async def get_passive_watts(self) -> list[float]: return list(self._values) async def wait_for_next_message(self, timeout: float = 5) -> None: @@ -75,7 +75,7 @@ async def test_read_ct_powermeter_calls_wait_with_2s_when_enabled() -> None: async def test_stub_powermeter_raw_matches_watts() -> None: pm = _StubPowermeter([3.0, 4.0, 5.0]) assert ( - await pm.get_powermeter_watts_raw() + await pm.get_passive_watts() == await pm.get_powermeter_watts() == [ 3.0, @@ -130,7 +130,7 @@ async def test_grid_phases_reuse_the_control_loops_read() -> None: await read_ct_powermeter(("10.0.0.5", 0), powermeters) source.values = [0.0, 0.0, 0.0] - assert await _read_grid_phases(powermeters) == [100.0, 200.0, 300.0] + assert await _passive_grid_phases(powermeters) == [100.0, 200.0, 300.0] assert source.reads == 1 @@ -138,7 +138,7 @@ async def test_grid_phases_read_an_idle_meter_themselves() -> None: source = _PolledMeter([100.0, 200.0, 300.0]) powermeters = _configured(source) - assert await _read_grid_phases(powermeters) == [100.0, 200.0, 300.0] + assert await _passive_grid_phases(powermeters) == [100.0, 200.0, 300.0] assert source.reads == 1 diff --git a/src/astrameter/powermeter/base.py b/src/astrameter/powermeter/base.py index e1edf0ff..dfb8499b 100644 --- a/src/astrameter/powermeter/base.py +++ b/src/astrameter/powermeter/base.py @@ -37,15 +37,17 @@ class Powermeter(ABC): @abstractmethod async def get_powermeter_watts(self) -> list[float]: ... - async def get_powermeter_watts_raw(self) -> list[float]: - """Per-phase watts before section/global processing wrappers. + async def get_passive_watts(self) -> list[float]: + """Per-phase watts for a reader that only reports them. - Used by the passive readers (the Marstek MQTT display, cloud reporting), - which should match the physical meter while control still uses - :meth:`get_powermeter_watts`, and must not add load on it: the + The passive route, for the Marstek MQTT display and cloud reporting: + the source's own reading, before offsets and filters, which never + feeds the processing chain and must not add load on the meter. The ``LastReadingPowermeter`` the config loader puts around every source - answers from the control loop's recent read. Defaults to the same values - as :meth:`get_powermeter_watts` for sources with no inner pipeline. + answers it from the control loop's recent read, and reads the source + only when nothing else has lately. Control uses + :meth:`get_powermeter_watts` instead. Defaults to the same values as + :meth:`get_powermeter_watts` for a source with no wrappers. """ return await self.get_powermeter_watts() diff --git a/src/astrameter/powermeter/wrappers/base.py b/src/astrameter/powermeter/wrappers/base.py index ee901857..afcb62fe 100644 --- a/src/astrameter/powermeter/wrappers/base.py +++ b/src/astrameter/powermeter/wrappers/base.py @@ -7,8 +7,8 @@ class PowermeterWrapper(Powermeter): def __init__(self, wrapped_powermeter: Powermeter) -> None: self.wrapped_powermeter = wrapped_powermeter - async def get_powermeter_watts_raw(self) -> list[float]: - return await self.wrapped_powermeter.get_powermeter_watts_raw() + async def get_passive_watts(self) -> list[float]: + return await self.wrapped_powermeter.get_passive_watts() def stream_online(self) -> bool | None: return self.wrapped_powermeter.stream_online() diff --git a/src/astrameter/powermeter/wrappers/conftest.py b/src/astrameter/powermeter/wrappers/conftest.py index 96eef1e3..e9adca5c 100644 --- a/src/astrameter/powermeter/wrappers/conftest.py +++ b/src/astrameter/powermeter/wrappers/conftest.py @@ -25,7 +25,7 @@ def set(self, values: list[float]) -> None: async def get_powermeter_watts(self) -> list[float]: return list(self._values) - async def get_powermeter_watts_raw(self) -> list[float]: + async def get_passive_watts(self) -> list[float]: return list(self._values) async def wait_for_message(self, timeout: float = 5) -> None: diff --git a/src/astrameter/powermeter/wrappers/health.py b/src/astrameter/powermeter/wrappers/health.py index b8b08a8a..c9ef14c6 100644 --- a/src/astrameter/powermeter/wrappers/health.py +++ b/src/astrameter/powermeter/wrappers/health.py @@ -109,8 +109,8 @@ async def get_powermeter_watts(self) -> list[float]: self._last_values = list(result) return result - async def get_powermeter_watts_raw(self) -> list[float]: - return await self._tracked(self.wrapped_powermeter.get_powermeter_watts_raw) + async def get_passive_watts(self) -> list[float]: + return await self._tracked(self.wrapped_powermeter.get_passive_watts) async def _tracked(self, fn: Callable[[], Awaitable[list[float]]]) -> list[float]: self._last_attempt = self._clock() diff --git a/src/astrameter/powermeter/wrappers/health_test.py b/src/astrameter/powermeter/wrappers/health_test.py index 965217a2..06d27854 100644 --- a/src/astrameter/powermeter/wrappers/health_test.py +++ b/src/astrameter/powermeter/wrappers/health_test.py @@ -78,12 +78,12 @@ async def test_empty_result_counts_as_not_ok() -> None: assert pm.last_outcome_ok is False -async def test_raw_read_also_tracked() -> None: +async def test_passive_read_also_tracked() -> None: inner = Mock(spec=Powermeter) - inner.get_powermeter_watts_raw = AsyncMock(return_value=[7.0]) + inner.get_passive_watts = AsyncMock(return_value=[7.0]) pm = _make(inner) - assert await pm.get_powermeter_watts_raw() == [7.0] + assert await pm.get_passive_watts() == [7.0] assert pm.last_outcome_ok is True diff --git a/src/astrameter/powermeter/wrappers/last_reading.py b/src/astrameter/powermeter/wrappers/last_reading.py index 657a77f8..16900ec7 100644 --- a/src/astrameter/powermeter/wrappers/last_reading.py +++ b/src/astrameter/powermeter/wrappers/last_reading.py @@ -16,13 +16,14 @@ class LastReadingPowermeter(PowermeterWrapper): """Innermost wrapper that lets passive readers reuse the control loop's read. Every real read of the source passes through here and is remembered. - :meth:`get_powermeter_watts_raw` — what the diagnostic readers call — + :meth:`get_passive_watts` — what the diagnostic readers call — serves that reading while it is at most ``max_age`` seconds old, and only reads the source itself when nothing has read it for longer, so those readers add no load on a meter the batteries are already polling. A push meter already answers from the reading it was sent, with no I/O, so - its raw reads go straight through and keep the meter's own staleness rule. + its passive reads go straight through and keep the meter's own staleness + rule. ``reset()`` passes through without forgetting the reading: it clears the filters' rolling state, and this is a fact about the meter. @@ -48,9 +49,9 @@ async def get_powermeter_watts(self) -> list[float]: self._read_at = self._clock() return values - async def get_powermeter_watts_raw(self) -> list[float]: + async def get_passive_watts(self) -> list[float]: if self.wrapped_powermeter.stream_online() is not None: - return await self.wrapped_powermeter.get_powermeter_watts_raw() + return await self.wrapped_powermeter.get_passive_watts() if self._values is not None and stream_fresh( self._read_at, self.max_age, self._clock ): diff --git a/src/astrameter/powermeter/wrappers/last_reading_test.py b/src/astrameter/powermeter/wrappers/last_reading_test.py index 5ffa0eef..3846c601 100644 --- a/src/astrameter/powermeter/wrappers/last_reading_test.py +++ b/src/astrameter/powermeter/wrappers/last_reading_test.py @@ -31,7 +31,7 @@ class _PushSource(_CountingSource): def stream_online(self) -> bool | None: return True - async def get_powermeter_watts_raw(self) -> list[float]: + async def get_passive_watts(self) -> list[float]: return await self.get_powermeter_watts() @@ -50,7 +50,7 @@ async def test_raw_read_reuses_a_recent_control_read() -> None: source.set([999.0, 999.0]) clock.now = 5.0 - assert await pm.get_powermeter_watts_raw() == [100.0, 200.0] + assert await pm.get_passive_watts() == [100.0, 200.0] assert source.reads == 1 @@ -61,12 +61,12 @@ async def test_raw_read_reads_the_source_once_the_reading_is_too_old() -> None: source.set([150.0]) clock.now = 5.1 - assert await pm.get_powermeter_watts_raw() == [150.0] + assert await pm.get_passive_watts() == [150.0] assert source.reads == 2 # That read is remembered, so the next passive read within max_age is free. clock.now = 6.0 - assert await pm.get_powermeter_watts_raw() == [150.0] + assert await pm.get_passive_watts() == [150.0] assert source.reads == 2 @@ -74,7 +74,7 @@ async def test_raw_read_reads_the_source_when_nothing_has_read_it() -> None: source = _CountingSource([42.0]) pm, _ = _make(source) - assert await pm.get_powermeter_watts_raw() == [42.0] + assert await pm.get_passive_watts() == [42.0] assert source.reads == 1 @@ -89,13 +89,13 @@ async def test_failed_or_empty_reads_are_not_remembered() -> None: await pm.get_powermeter_watts() # The raw read does not fall back to the old reading: it asks the source. with pytest.raises(ValueError): - await pm.get_powermeter_watts_raw() + await pm.get_passive_watts() source.error = None source.set([]) assert await pm.get_powermeter_watts() == [] source.set([7.0]) - assert await pm.get_powermeter_watts_raw() == [7.0] + assert await pm.get_passive_watts() == [7.0] async def test_served_reading_is_a_copy() -> None: @@ -103,9 +103,9 @@ async def test_served_reading_is_a_copy() -> None: pm, _ = _make(source) await pm.get_powermeter_watts() - served = await pm.get_powermeter_watts_raw() + served = await pm.get_passive_watts() served[0] = 999.0 - assert await pm.get_powermeter_watts_raw() == [1.0, 2.0] + assert await pm.get_passive_watts() == [1.0, 2.0] async def test_push_meter_raw_read_goes_straight_to_its_cache() -> None: @@ -117,7 +117,7 @@ async def test_push_meter_raw_read_goes_straight_to_its_cache() -> None: source.error = ValueError("stale") with pytest.raises(ValueError, match="stale"): - await pm.get_powermeter_watts_raw() + await pm.get_passive_watts() async def test_reset_keeps_the_reading() -> None: @@ -127,5 +127,5 @@ async def test_reset_keeps_the_reading() -> None: pm.reset() assert source.reset_count == 1 - assert await pm.get_powermeter_watts_raw() == [100.0] + assert await pm.get_passive_watts() == [100.0] assert source.reads == 1 diff --git a/src/astrameter/powermeter/wrappers/pid_test.py b/src/astrameter/powermeter/wrappers/pid_test.py index 207da318..10a9f658 100644 --- a/src/astrameter/powermeter/wrappers/pid_test.py +++ b/src/astrameter/powermeter/wrappers/pid_test.py @@ -10,7 +10,7 @@ def mock_powermeter() -> Mock: """Return a mock powermeter with async stubs for all interface methods.""" pm = Mock() pm.get_powermeter_watts = AsyncMock() - pm.get_powermeter_watts_raw = pm.get_powermeter_watts + pm.get_passive_watts = pm.get_powermeter_watts pm.wait_for_message = AsyncMock() pm.wait_for_next_message = AsyncMock() pm.start = AsyncMock() diff --git a/src/astrameter/powermeter/wrappers/throttling.py b/src/astrameter/powermeter/wrappers/throttling.py index f5a37cdd..0c2225fc 100644 --- a/src/astrameter/powermeter/wrappers/throttling.py +++ b/src/astrameter/powermeter/wrappers/throttling.py @@ -34,11 +34,6 @@ def __init__( self._last_values: list[float] | None = None self._pending_fetch: asyncio.Future[list[float]] | None = None - async def get_powermeter_watts_raw(self) -> list[float]: - # Raw reads skip throttle coalescing so the Marstek app can show sensor-level - # watts without being tied to the CT002 control cadence. - return await self.wrapped_powermeter.get_powermeter_watts_raw() - async def get_powermeter_watts(self) -> list[float]: if self.throttle_interval <= 0: return await self.wrapped_powermeter.get_powermeter_watts() diff --git a/src/astrameter/powermeter/wrappers/throttling_test.py b/src/astrameter/powermeter/wrappers/throttling_test.py index ad6e4c3c..b0a4dda6 100644 --- a/src/astrameter/powermeter/wrappers/throttling_test.py +++ b/src/astrameter/powermeter/wrappers/throttling_test.py @@ -10,7 +10,7 @@ async def test_no_throttling_always_fetches_fresh_values() -> None: """Test that when throttling is disabled, fresh values are always fetched.""" mock_pm = Mock() mock_pm.get_powermeter_watts = AsyncMock(return_value=[100.0, 200.0, 300.0]) - mock_pm.get_powermeter_watts_raw = mock_pm.get_powermeter_watts + mock_pm.get_passive_watts = mock_pm.get_powermeter_watts throttled = ThrottledPowermeter(mock_pm, throttle_interval=0) result1 = await throttled.get_powermeter_watts() @@ -25,7 +25,7 @@ async def test_throttling_waits_for_interval() -> None: """Test that throttling waits for remaining time before fetching new values.""" mock_pm = Mock() mock_pm.get_powermeter_watts = AsyncMock(return_value=[100.0, 200.0, 300.0]) - mock_pm.get_powermeter_watts_raw = mock_pm.get_powermeter_watts + mock_pm.get_passive_watts = mock_pm.get_powermeter_watts throttled = ThrottledPowermeter(mock_pm, throttle_interval=0.2) result1 = await throttled.get_powermeter_watts() @@ -48,7 +48,7 @@ async def test_throttling_fetches_fresh_after_interval() -> None: """Test that fresh values are fetched after throttling interval passes.""" mock_pm = Mock() mock_pm.get_powermeter_watts = AsyncMock(return_value=[100.0, 200.0, 300.0]) - mock_pm.get_powermeter_watts_raw = mock_pm.get_powermeter_watts + mock_pm.get_passive_watts = mock_pm.get_powermeter_watts throttled = ThrottledPowermeter(mock_pm, throttle_interval=0.1) result1 = await throttled.get_powermeter_watts() @@ -89,7 +89,7 @@ async def test_exception_handling_with_cache() -> None: mock_pm.start = AsyncMock() mock_pm.stop = AsyncMock() mock_pm.get_powermeter_watts = AsyncMock(return_value=[100.0, 200.0]) - mock_pm.get_powermeter_watts_raw = mock_pm.get_powermeter_watts + mock_pm.get_passive_watts = mock_pm.get_powermeter_watts throttled = ThrottledPowermeter(mock_pm, throttle_interval=0.1) @@ -108,7 +108,7 @@ async def test_exception_raises_without_cache() -> None: mock_pm.start = AsyncMock() mock_pm.stop = AsyncMock() mock_pm.get_powermeter_watts = AsyncMock(side_effect=Exception("Network error")) - mock_pm.get_powermeter_watts_raw = mock_pm.get_powermeter_watts + mock_pm.get_passive_watts = mock_pm.get_powermeter_watts throttled = ThrottledPowermeter(mock_pm, throttle_interval=0.1) @@ -116,14 +116,14 @@ async def test_exception_raises_without_cache() -> None: await throttled.get_powermeter_watts() -async def test_throttled_raw_bypasses_get_and_throttle_coalescing() -> None: +async def test_throttled_passive_read_bypasses_get_and_throttle_coalescing() -> None: mock_pm = Mock() get_m = AsyncMock(return_value=[1.0]) raw_m = AsyncMock(return_value=[2.0, 3.0, 4.0]) mock_pm.get_powermeter_watts = get_m - mock_pm.get_powermeter_watts_raw = raw_m + mock_pm.get_passive_watts = raw_m throttled = ThrottledPowermeter(mock_pm, throttle_interval=3600.0) - assert await throttled.get_powermeter_watts_raw() == [2.0, 3.0, 4.0] + assert await throttled.get_passive_watts() == [2.0, 3.0, 4.0] raw_m.assert_awaited_once() get_m.assert_not_called() diff --git a/src/astrameter/powermeter/wrappers/transform_test.py b/src/astrameter/powermeter/wrappers/transform_test.py index 1586e945..2c0c8602 100644 --- a/src/astrameter/powermeter/wrappers/transform_test.py +++ b/src/astrameter/powermeter/wrappers/transform_test.py @@ -9,7 +9,7 @@ def mock_powermeter() -> Mock: pm = Mock() pm.get_powermeter_watts = AsyncMock() - pm.get_powermeter_watts_raw = AsyncMock() + pm.get_passive_watts = AsyncMock() pm.wait_for_message = AsyncMock() pm.wait_for_next_message = AsyncMock() return pm @@ -19,11 +19,11 @@ async def test_transformed_raw_matches_wrapped_without_offset( mock_powermeter: Mock, ) -> None: mock_powermeter.get_powermeter_watts.return_value = [110.0, 210.0, 310.0] - mock_powermeter.get_powermeter_watts_raw.return_value = [100.0, 200.0, 300.0] + mock_powermeter.get_passive_watts.return_value = [100.0, 200.0, 300.0] t = TransformedPowermeter(mock_powermeter, [10.0], [1.0]) assert await t.get_powermeter_watts() == [120.0, 220.0, 320.0] - assert await t.get_powermeter_watts_raw() == [100.0, 200.0, 300.0] - mock_powermeter.get_powermeter_watts_raw.assert_awaited_once() + assert await t.get_passive_watts() == [100.0, 200.0, 300.0] + mock_powermeter.get_passive_watts.assert_awaited_once() async def test_identity_single_phase(mock_powermeter: Mock) -> None: diff --git a/src/astrameter/shelly/shelly_udp_test.py b/src/astrameter/shelly/shelly_udp_test.py index 1f90c88b..40c4df97 100644 --- a/src/astrameter/shelly/shelly_udp_test.py +++ b/src/astrameter/shelly/shelly_udp_test.py @@ -21,7 +21,7 @@ async def get_powermeter_watts(self) -> list[float]: self.call_count += 1 return [1.0] - async def get_powermeter_watts_raw(self) -> list[float]: + async def get_passive_watts(self) -> list[float]: # Same physical reading as get_powermeter_watts; do not bump call_count so # throttling/dedupe tests that only observe get_powermeter_watts stay stable. return [1.0] From cb6413dac47426a393048f40f1b7f0fc21084307 Mon Sep 17 00:00:00 2001 From: tomquist <528585+tomquist@users.noreply.github.com> Date: Tue, 13 Oct 2026 07:18:18 +0000 Subject: [PATCH 4/7] Let a passive read wait for a source read already in flight A passive read that found no fresh reading read the source even while a battery's read of it, or another passive reader's, was still running, so overlapping reads still cost extra requests. It now waits for the read in flight and checks the cache again before reading the source itself. A read that fails leaves nothing behind, so the passive reader still asks the source and never gets a stale value. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01Vv6VhaqsWHbc7sTdpnfNLw --- .../powermeter/wrappers/last_reading.py | 37 ++++++++-- .../powermeter/wrappers/last_reading_test.py | 67 +++++++++++++++++++ 2 files changed, 98 insertions(+), 6 deletions(-) diff --git a/src/astrameter/powermeter/wrappers/last_reading.py b/src/astrameter/powermeter/wrappers/last_reading.py index 16900ec7..7b20561c 100644 --- a/src/astrameter/powermeter/wrappers/last_reading.py +++ b/src/astrameter/powermeter/wrappers/last_reading.py @@ -1,3 +1,4 @@ +import asyncio import time from collections.abc import Callable @@ -25,6 +26,10 @@ class LastReadingPowermeter(PowermeterWrapper): its passive reads go straight through and keep the meter's own staleness rule. + A passive read that finds no fresh reading first waits for a read of the + source already in flight, a battery's or another passive reader's, and + only reads the source itself if that one left nothing fresh behind. + ``reset()`` passes through without forgetting the reading: it clears the filters' rolling state, and this is a fact about the meter. """ @@ -41,19 +46,39 @@ def __init__( self._clock = clock or time.monotonic self._values: list[float] | None = None self._read_at: float | None = None + # The most recent read of the source still running. It resolves to + # None however that read ends: a waiting passive reader looks at the + # cache afterwards rather than at the read's outcome. + self._in_flight: asyncio.Future[None] | None = None async def get_powermeter_watts(self) -> list[float]: - values = await self.wrapped_powermeter.get_powermeter_watts() - if values: - self._values = list(values) - self._read_at = self._clock() - return values + done: asyncio.Future[None] = asyncio.get_running_loop().create_future() + self._in_flight = done + try: + values = await self.wrapped_powermeter.get_powermeter_watts() + if values: + self._values = list(values) + self._read_at = self._clock() + return values + finally: + done.set_result(None) + if self._in_flight is done: + self._in_flight = None async def get_passive_watts(self) -> list[float]: if self.wrapped_powermeter.stream_online() is not None: return await self.wrapped_powermeter.get_passive_watts() + while True: + fresh = self._fresh_values() + if fresh is not None: + return fresh + if self._in_flight is None: + return await self.get_powermeter_watts() + await asyncio.shield(self._in_flight) + + def _fresh_values(self) -> list[float] | None: if self._values is not None and stream_fresh( self._read_at, self.max_age, self._clock ): return list(self._values) - return await self.get_powermeter_watts() + return None diff --git a/src/astrameter/powermeter/wrappers/last_reading_test.py b/src/astrameter/powermeter/wrappers/last_reading_test.py index 3846c601..f9a22085 100644 --- a/src/astrameter/powermeter/wrappers/last_reading_test.py +++ b/src/astrameter/powermeter/wrappers/last_reading_test.py @@ -1,3 +1,5 @@ +import asyncio + import pytest from .conftest import FakePowermeter @@ -27,6 +29,24 @@ async def get_powermeter_watts(self) -> list[float]: return await super().get_powermeter_watts() +class _SlowSource(_CountingSource): + """A polled source whose reads stay in flight until ``release()``.""" + + def __init__(self, values: list[float]) -> None: + super().__init__(values) + self._gate = asyncio.Event() + + def release(self) -> None: + self._gate.set() + + async def get_powermeter_watts(self) -> list[float]: + self.reads += 1 + await self._gate.wait() + if self.error is not None: + raise self.error + return list(self._values) + + class _PushSource(_CountingSource): def stream_online(self) -> bool | None: return True @@ -129,3 +149,50 @@ async def test_reset_keeps_the_reading() -> None: assert source.reset_count == 1 assert await pm.get_passive_watts() == [100.0] assert source.reads == 1 + + +async def test_passive_read_waits_for_a_control_read_in_flight() -> None: + source = _SlowSource([100.0]) + pm, _ = _make(source) + + control = asyncio.create_task(pm.get_powermeter_watts()) + await asyncio.sleep(0) + passive = asyncio.create_task(pm.get_passive_watts()) + await asyncio.sleep(0) + source.release() + + assert await control == [100.0] + assert await passive == [100.0] + assert source.reads == 1 + + +async def test_overlapping_passive_reads_share_one_source_read() -> None: + source = _SlowSource([42.0]) + pm, _ = _make(source) + + readers = [asyncio.create_task(pm.get_passive_watts()) for _ in range(3)] + await asyncio.sleep(0) + source.release() + + assert await asyncio.gather(*readers) == [[42.0], [42.0], [42.0]] + assert source.reads == 1 + + +async def test_passive_read_reads_itself_when_the_read_it_waited_for_fails() -> None: + source = _SlowSource([100.0]) + source.error = ValueError("offline") + pm, _ = _make(source) + + control = asyncio.create_task(pm.get_powermeter_watts()) + await asyncio.sleep(0) + passive = asyncio.create_task(pm.get_passive_watts()) + await asyncio.sleep(0) + source.release() + + with pytest.raises(ValueError): + await control + # The failure leaves nothing to reuse, so the passive reader asks the + # source and gets its own answer, never a stale value. + with pytest.raises(ValueError): + await passive + assert source.reads == 2 From 0e458b0b8867819c473f6a2870cd09a2209ef98d Mon Sep 17 00:00:00 2001 From: tomquist <528585+tomquist@users.noreply.github.com> Date: Tue, 13 Oct 2026 07:27:33 +0000 Subject: [PATCH 5/7] Track every source read in flight, not just the latest Without a throttle above it, battery polls can overlap on the wrapper. It only remembered the newest read, so when that one came back empty or failed first, a passive reader no longer saw the older read still running and started a request of its own. Keep every read in flight in a set and have a passive reader wait until none is left before reading the source. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01Vv6VhaqsWHbc7sTdpnfNLw --- .../powermeter/wrappers/last_reading.py | 22 ++++++---- .../powermeter/wrappers/last_reading_test.py | 41 +++++++++++++++++++ 2 files changed, 54 insertions(+), 9 deletions(-) diff --git a/src/astrameter/powermeter/wrappers/last_reading.py b/src/astrameter/powermeter/wrappers/last_reading.py index 7b20561c..8126a9b1 100644 --- a/src/astrameter/powermeter/wrappers/last_reading.py +++ b/src/astrameter/powermeter/wrappers/last_reading.py @@ -46,14 +46,15 @@ def __init__( self._clock = clock or time.monotonic self._values: list[float] | None = None self._read_at: float | None = None - # The most recent read of the source still running. It resolves to - # None however that read ends: a waiting passive reader looks at the - # cache afterwards rather than at the read's outcome. - self._in_flight: asyncio.Future[None] | None = None + # Every read of the source still running: battery polls can overlap + # when no throttle sits above. Each resolves to None however its read + # ends, so a waiting passive reader looks at the cache afterwards + # rather than at the read's outcome. + self._in_flight: set[asyncio.Future[None]] = set() async def get_powermeter_watts(self) -> list[float]: done: asyncio.Future[None] = asyncio.get_running_loop().create_future() - self._in_flight = done + self._in_flight.add(done) try: values = await self.wrapped_powermeter.get_powermeter_watts() if values: @@ -62,8 +63,7 @@ async def get_powermeter_watts(self) -> list[float]: return values finally: done.set_result(None) - if self._in_flight is done: - self._in_flight = None + self._in_flight.discard(done) async def get_passive_watts(self) -> list[float]: if self.wrapped_powermeter.stream_online() is not None: @@ -72,9 +72,13 @@ async def get_passive_watts(self) -> list[float]: fresh = self._fresh_values() if fresh is not None: return fresh - if self._in_flight is None: + if not self._in_flight: return await self.get_powermeter_watts() - await asyncio.shield(self._in_flight) + # asyncio.wait never cancels what it waits on, even if this + # reader is cancelled, so the reads it waits for run to the end. + await asyncio.wait( + set(self._in_flight), return_when=asyncio.FIRST_COMPLETED + ) def _fresh_values(self) -> list[float] | None: if self._values is not None and stream_fresh( diff --git a/src/astrameter/powermeter/wrappers/last_reading_test.py b/src/astrameter/powermeter/wrappers/last_reading_test.py index f9a22085..906b15bc 100644 --- a/src/astrameter/powermeter/wrappers/last_reading_test.py +++ b/src/astrameter/powermeter/wrappers/last_reading_test.py @@ -47,6 +47,24 @@ async def get_powermeter_watts(self) -> list[float]: return list(self._values) +class _PerReadSource(FakePowermeter): + """A polled source whose reads each finish when the test says so.""" + + def __init__(self) -> None: + super().__init__([]) + self.reads = 0 + self._pending: list[asyncio.Future[list[float]]] = [] + + def finish(self, index: int, values: list[float]) -> None: + self._pending[index].set_result(values) + + async def get_powermeter_watts(self) -> list[float]: + self.reads += 1 + read: asyncio.Future[list[float]] = asyncio.get_running_loop().create_future() + self._pending.append(read) + return await read + + class _PushSource(_CountingSource): def stream_online(self) -> bool | None: return True @@ -196,3 +214,26 @@ async def test_passive_read_reads_itself_when_the_read_it_waited_for_fails() -> with pytest.raises(ValueError): await passive assert source.reads == 2 + + +async def test_passive_read_waits_for_every_control_read_in_flight() -> None: + """Without a throttle, battery polls overlap. A newer one coming back + empty must not hide the older one still running from a passive reader.""" + source = _PerReadSource() + pm, _ = _make(source) + + older = asyncio.create_task(pm.get_powermeter_watts()) + newer = asyncio.create_task(pm.get_powermeter_watts()) + await asyncio.sleep(0) + passive = asyncio.create_task(pm.get_passive_watts()) + await asyncio.sleep(0) + + source.finish(1, []) + assert await newer == [] + await asyncio.sleep(0) + assert not passive.done() + + source.finish(0, [100.0]) + assert await older == [100.0] + assert await passive == [100.0] + assert source.reads == 2 From 22df0e396f4551ba9dfe5dc7c718e0aeb450b46e Mon Sep 17 00:00:00 2001 From: tomquist <528585+tomquist@users.noreply.github.com> Date: Tue, 13 Oct 2026 19:36:41 +0000 Subject: [PATCH 6/7] Bound passive reads during an outage and keep cache hits out of health The regression review found two problems with the passive route. A passive reader waited for every read of the source in flight, re-checked on each pass. With battery polls overlapping and a polled meter down, a new read was always running, so the Marstek responder or cloud report waited for as long as the outage lasted. It now waits only for the reads it found running, then reads the source itself, so the failure surfaces after one timeout of its own. The health wrapper recorded every passive read as an attempt, so a reading served from the cache flipped a meter the control loop had just seen fail back to online. It now records a passive read only when the source answered it, which LastReadingPowermeter.source_reads tells it. Failures are still recorded. Also clamp a negative THROTTLE_INTERVAL to 0 when sizing the cache window, so it no longer means a reading reused for ever, and drop the remaining "raw" from test names. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01Vv6VhaqsWHbc7sTdpnfNLw --- src/astrameter/config/config_loader.py | 5 +- src/astrameter/config/config_loader_test.py | 25 +++++++++ src/astrameter/main_test.py | 2 +- src/astrameter/powermeter/wrappers/health.py | 29 +++++++++- .../powermeter/wrappers/health_test.py | 55 ++++++++++++++++++ .../powermeter/wrappers/last_reading.py | 29 +++++++--- .../powermeter/wrappers/last_reading_test.py | 56 +++++++++++++++++-- .../powermeter/wrappers/transform_test.py | 2 +- 8 files changed, 187 insertions(+), 16 deletions(-) diff --git a/src/astrameter/config/config_loader.py b/src/astrameter/config/config_loader.py index 765f586a..091e47e0 100644 --- a/src/astrameter/config/config_loader.py +++ b/src/astrameter/config/config_loader.py @@ -362,8 +362,11 @@ def apply_signal_wrappers( # Innermost, so passive readers see the source's own reading. The throttle # above it lets a control read through at most once per interval, so the # reading may be that much older before a passive reader reads for itself. + # A negative interval means no throttle, as below, not a cache that never + # expires. powermeter = LastReadingPowermeter( - powermeter, max_age=PASSIVE_READING_MAX_AGE_S + signal.throttle_interval + powermeter, + max_age=PASSIVE_READING_MAX_AGE_S + max(0.0, signal.throttle_interval), ) if signal.offsets is not None or signal.multipliers is not None: diff --git a/src/astrameter/config/config_loader_test.py b/src/astrameter/config/config_loader_test.py index 4259a617..84b86e30 100644 --- a/src/astrameter/config/config_loader_test.py +++ b/src/astrameter/config/config_loader_test.py @@ -881,3 +881,28 @@ def test_read_all_configs_wait_for_next_message_section_override() -> None: # Section order is preserved by configparser, so SCRIPT_1 (override=false) # is first and SCRIPT_2 (inherits global true) is second. assert [wait for _, _, wait in powermeters] == [False, True] + + +@pytest.mark.parametrize(("throttle", "max_age"), [(0.0, 5.0), (3.0, 8.0), (-5.0, 5.0)]) +def test_shared_reading_ages_out_with_the_throttle( + throttle: float, max_age: float +) -> None: + """Passive readers may reuse a reading for as long as the throttle holds + control reads back; a negative interval means no throttle, not a reading + reused for ever.""" + from dataclasses import replace + + from astrameter.config.config_loader import apply_signal_wrappers + from astrameter.config.settings import SignalSettings + from astrameter.powermeter.script import Script + from astrameter.powermeter.wrappers.base import PowermeterWrapper + from astrameter.powermeter.wrappers.last_reading import LastReadingPowermeter + + signal = replace(SignalSettings(), throttle_interval=throttle) + node = apply_signal_wrappers(Script('echo "100"'), "SCRIPT_1", signal) + while isinstance(node, PowermeterWrapper) and not isinstance( + node, LastReadingPowermeter + ): + node = node.wrapped_powermeter + assert isinstance(node, LastReadingPowermeter) + assert node.max_age == max_age diff --git a/src/astrameter/main_test.py b/src/astrameter/main_test.py index 30aeab14..e1e35274 100644 --- a/src/astrameter/main_test.py +++ b/src/astrameter/main_test.py @@ -72,7 +72,7 @@ async def test_read_ct_powermeter_calls_wait_with_2s_when_enabled() -> None: assert pm._wait_calls == [2] -async def test_stub_powermeter_raw_matches_watts() -> None: +async def test_stub_powermeter_passive_matches_watts() -> None: pm = _StubPowermeter([3.0, 4.0, 5.0]) assert ( await pm.get_passive_watts() diff --git a/src/astrameter/powermeter/wrappers/health.py b/src/astrameter/powermeter/wrappers/health.py index c9ef14c6..f22ec476 100644 --- a/src/astrameter/powermeter/wrappers/health.py +++ b/src/astrameter/powermeter/wrappers/health.py @@ -59,6 +59,15 @@ def __init__( # Last successful processed read, so the health loop can publish the # most recent readings without issuing an extra read. self._last_values: list[float] | None = None + # The wrapper that serves passive reads from the control loop's last + # read, if the chain has one: such a read is only news about the + # meter when it made the source answer. + node: Powermeter = wrapped_powermeter + while isinstance(node, PowermeterWrapper) and not isinstance( + node, LastReadingPowermeter + ): + node = node.wrapped_powermeter + self._last_reading = node if isinstance(node, LastReadingPowermeter) else None @property def last_attempt(self) -> float | None: @@ -110,7 +119,25 @@ async def get_powermeter_watts(self) -> list[float]: return result async def get_passive_watts(self) -> list[float]: - return await self._tracked(self.wrapped_powermeter.get_passive_watts) + shared = self._last_reading + if shared is None or self.stream_online() is not None: + return await self._tracked(self.wrapped_powermeter.get_passive_watts) + # A reading served from the cache was already recorded by the read + # that fetched it; recording it again would mark a meter that has + # failed since then as healthy. A failure always comes from the + # source, so it is recorded. + reads_before = shared.source_reads + attempt = self._clock() + try: + result = await self.wrapped_powermeter.get_passive_watts() + except Exception: + self._last_attempt = attempt + self._last_outcome_ok = False + raise + if shared.source_reads != reads_before: + self._last_attempt = attempt + self._last_outcome_ok = bool(result) + return result async def _tracked(self, fn: Callable[[], Awaitable[list[float]]]) -> list[float]: self._last_attempt = self._clock() diff --git a/src/astrameter/powermeter/wrappers/health_test.py b/src/astrameter/powermeter/wrappers/health_test.py index 06d27854..05c68329 100644 --- a/src/astrameter/powermeter/wrappers/health_test.py +++ b/src/astrameter/powermeter/wrappers/health_test.py @@ -116,3 +116,58 @@ def test_pipeline_leaves_out_the_shared_reading() -> None: snapshot = pm.status_snapshot() assert snapshot.kind == "FakePowermeter" assert snapshot.pipeline == ("ThrottledPowermeter",) + + +class _FailingSource(FakePowermeter): + """A polled source that can be switched to raising on every read.""" + + def __init__(self, values: list[float]) -> None: + super().__init__(values) + self.error: Exception | None = None + + async def get_powermeter_watts(self) -> list[float]: + if self.error is not None: + raise self.error + return await super().get_powermeter_watts() + + +def _shared( + source: Powermeter, clock: _FakeClock +) -> tuple[HealthTrackingPowermeter, LastReadingPowermeter]: + shared = LastReadingPowermeter(source, max_age=5.0, clock=clock) + return _make(shared, clock=clock), shared + + +async def test_cached_passive_read_does_not_mark_a_failing_meter_healthy() -> None: + """The reading a passive reader is served from the cache predates the + failure the control loop just saw, so it must not paper over it.""" + clock = _FakeClock() + source = _FailingSource([100.0]) + pm, _ = _shared(source, clock) + await pm.get_powermeter_watts() + + clock.advance(1.0) + source.error = OSError("offline") + with pytest.raises(OSError): + await pm.get_powermeter_watts() + assert (pm.last_attempt, pm.last_outcome_ok) == (1.0, False) + + clock.advance(1.0) + assert await pm.get_passive_watts() == [100.0] + assert (pm.last_attempt, pm.last_outcome_ok) == (1.0, False) + + +async def test_passive_read_that_reads_the_source_is_recorded() -> None: + clock = _FakeClock() + source = _FailingSource([100.0]) + pm, _ = _shared(source, clock) + + clock.advance(1.0) + assert await pm.get_passive_watts() == [100.0] + assert (pm.last_attempt, pm.last_outcome_ok) == (1.0, True) + + clock.advance(10.0) + source.error = OSError("offline") + with pytest.raises(OSError): + await pm.get_passive_watts() + assert (pm.last_attempt, pm.last_outcome_ok) == (11.0, False) diff --git a/src/astrameter/powermeter/wrappers/last_reading.py b/src/astrameter/powermeter/wrappers/last_reading.py index 8126a9b1..697df95b 100644 --- a/src/astrameter/powermeter/wrappers/last_reading.py +++ b/src/astrameter/powermeter/wrappers/last_reading.py @@ -26,9 +26,13 @@ class LastReadingPowermeter(PowermeterWrapper): its passive reads go straight through and keep the meter's own staleness rule. - A passive read that finds no fresh reading first waits for a read of the - source already in flight, a battery's or another passive reader's, and - only reads the source itself if that one left nothing fresh behind. + A passive read that finds no fresh reading first waits for the reads of + the source already in flight when it arrived, a battery's or another + passive reader's, and reads the source itself only if they left nothing + fresh behind. Reads that start while it waits are not waited for, so a + meter that keeps failing under continuous battery polls answers the + passive reader after the reads it found running and one of its own, + rather than keeping it waiting for as long as the polls overlap. ``reset()`` passes through without forgetting the reading: it clears the filters' rolling state, and this is a fact about the meter. @@ -51,12 +55,23 @@ def __init__( # ends, so a waiting passive reader looks at the cache afterwards # rather than at the read's outcome. self._in_flight: set[asyncio.Future[None]] = set() + self._source_reads = 0 + + @property + def source_reads(self) -> int: + """How many reads of the source have returned, whatever they returned. + + Lets the health wrapper tell a passive read the source answered from + one served from this cache, which proves nothing about the meter now. + """ + return self._source_reads async def get_powermeter_watts(self) -> list[float]: done: asyncio.Future[None] = asyncio.get_running_loop().create_future() self._in_flight.add(done) try: values = await self.wrapped_powermeter.get_powermeter_watts() + self._source_reads += 1 if values: self._values = list(values) self._read_at = self._clock() @@ -68,17 +83,17 @@ async def get_powermeter_watts(self) -> list[float]: async def get_passive_watts(self) -> list[float]: if self.wrapped_powermeter.stream_online() is not None: return await self.wrapped_powermeter.get_passive_watts() + pending = set(self._in_flight) while True: fresh = self._fresh_values() if fresh is not None: return fresh - if not self._in_flight: + pending = {read for read in pending if not read.done()} + if not pending: return await self.get_powermeter_watts() # asyncio.wait never cancels what it waits on, even if this # reader is cancelled, so the reads it waits for run to the end. - await asyncio.wait( - set(self._in_flight), return_when=asyncio.FIRST_COMPLETED - ) + await asyncio.wait(pending, return_when=asyncio.FIRST_COMPLETED) def _fresh_values(self) -> list[float] | None: if self._values is not None and stream_fresh( diff --git a/src/astrameter/powermeter/wrappers/last_reading_test.py b/src/astrameter/powermeter/wrappers/last_reading_test.py index 906b15bc..c2245415 100644 --- a/src/astrameter/powermeter/wrappers/last_reading_test.py +++ b/src/astrameter/powermeter/wrappers/last_reading_test.py @@ -58,6 +58,9 @@ def __init__(self) -> None: def finish(self, index: int, values: list[float]) -> None: self._pending[index].set_result(values) + def fail(self, index: int, error: Exception) -> None: + self._pending[index].set_exception(error) + async def get_powermeter_watts(self) -> list[float]: self.reads += 1 read: asyncio.Future[list[float]] = asyncio.get_running_loop().create_future() @@ -80,7 +83,7 @@ def _make( return LastReadingPowermeter(source, max_age=max_age, clock=clock), clock -async def test_raw_read_reuses_a_recent_control_read() -> None: +async def test_passive_read_reuses_a_recent_control_read() -> None: source = _CountingSource([100.0, 200.0]) pm, clock = _make(source) @@ -92,7 +95,7 @@ async def test_raw_read_reuses_a_recent_control_read() -> None: assert source.reads == 1 -async def test_raw_read_reads_the_source_once_the_reading_is_too_old() -> None: +async def test_passive_read_reads_the_source_once_the_reading_is_too_old() -> None: source = _CountingSource([100.0]) pm, clock = _make(source) await pm.get_powermeter_watts() @@ -108,7 +111,7 @@ async def test_raw_read_reads_the_source_once_the_reading_is_too_old() -> None: assert source.reads == 2 -async def test_raw_read_reads_the_source_when_nothing_has_read_it() -> None: +async def test_passive_read_reads_the_source_when_nothing_has_read_it() -> None: source = _CountingSource([42.0]) pm, _ = _make(source) @@ -125,7 +128,7 @@ async def test_failed_or_empty_reads_are_not_remembered() -> None: source.error = ValueError("offline") with pytest.raises(ValueError): await pm.get_powermeter_watts() - # The raw read does not fall back to the old reading: it asks the source. + # The passive read does not fall back to the old reading: it asks the source. with pytest.raises(ValueError): await pm.get_passive_watts() @@ -146,7 +149,7 @@ async def test_served_reading_is_a_copy() -> None: assert await pm.get_passive_watts() == [1.0, 2.0] -async def test_push_meter_raw_read_goes_straight_to_its_cache() -> None: +async def test_push_meter_passive_read_goes_straight_to_its_cache() -> None: """A push meter answers without I/O and applies its own staleness rule, so the wrapper must not serve a reading the meter itself would now refuse.""" source = _PushSource([100.0]) @@ -237,3 +240,46 @@ async def test_passive_read_waits_for_every_control_read_in_flight() -> None: assert await older == [100.0] assert await passive == [100.0] assert source.reads == 2 + + +async def test_passive_read_does_not_wait_for_reads_started_after_it() -> None: + """A meter that keeps failing while battery polls overlap must still answer + a passive reader, with an error, rather than keep it waiting for ever.""" + source = _PerReadSource() + pm, _ = _make(source) + + first = asyncio.create_task(pm.get_powermeter_watts()) + await asyncio.sleep(0) + passive = asyncio.create_task(pm.get_passive_watts()) + await asyncio.sleep(0) + later = asyncio.create_task(pm.get_powermeter_watts()) + await asyncio.sleep(0) + + source.fail(0, OSError("timeout")) + with pytest.raises(OSError): + await first + await asyncio.sleep(0) + # The passive reader asks the source itself instead of waiting for the + # battery read that started after it. + assert source.reads == 3 + source.fail(2, OSError("timeout")) + with pytest.raises(OSError): + await passive + + source.finish(1, [100.0]) + assert await later == [100.0] + + +async def test_source_reads_counts_every_answer_but_not_cache_hits() -> None: + source = _CountingSource([100.0]) + pm, _ = _make(source) + + await pm.get_powermeter_watts() + await pm.get_passive_watts() + source.set([]) + await pm.get_powermeter_watts() + source.error = ValueError("offline") + with pytest.raises(ValueError): + await pm.get_powermeter_watts() + + assert pm.source_reads == 2 diff --git a/src/astrameter/powermeter/wrappers/transform_test.py b/src/astrameter/powermeter/wrappers/transform_test.py index 2c0c8602..cc7dbf3d 100644 --- a/src/astrameter/powermeter/wrappers/transform_test.py +++ b/src/astrameter/powermeter/wrappers/transform_test.py @@ -15,7 +15,7 @@ def mock_powermeter() -> Mock: return pm -async def test_transformed_raw_matches_wrapped_without_offset( +async def test_transformed_passive_matches_wrapped_without_offset( mock_powermeter: Mock, ) -> None: mock_powermeter.get_powermeter_watts.return_value = [110.0, 210.0, 310.0] From 287d012593e89134513679a1083e5f27f148e652 Mon Sep 17 00:00:00 2001 From: tomquist <528585+tomquist@users.noreply.github.com> Date: Tue, 13 Oct 2026 20:13:43 +0000 Subject: [PATCH 7/7] Record a passive read's health only when it reached the source itself A passive read waiting on overlapping battery reads can be served the one that succeeded after another had already failed. The shared read counter still moved, so the health wrapper recorded the cached success over the newer failure. Track, per calling task, whether the passive read reached the source, and record only then. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01Vv6VhaqsWHbc7sTdpnfNLw --- src/astrameter/powermeter/wrappers/health.py | 28 +++++------ .../powermeter/wrappers/health_test.py | 43 +++++++++++++++++ .../powermeter/wrappers/last_reading.py | 46 ++++++++++++++----- .../powermeter/wrappers/last_reading_test.py | 30 +++++++----- 4 files changed, 109 insertions(+), 38 deletions(-) diff --git a/src/astrameter/powermeter/wrappers/health.py b/src/astrameter/powermeter/wrappers/health.py index f22ec476..4f7071df 100644 --- a/src/astrameter/powermeter/wrappers/health.py +++ b/src/astrameter/powermeter/wrappers/health.py @@ -5,7 +5,7 @@ from astrameter.powermeter.base import Powermeter from .base import PowermeterWrapper -from .last_reading import LastReadingPowermeter +from .last_reading import LastReadingPowermeter, probe_passive_read @dataclasses.dataclass(frozen=True, slots=True) @@ -119,22 +119,22 @@ async def get_powermeter_watts(self) -> list[float]: return result async def get_passive_watts(self) -> list[float]: - shared = self._last_reading - if shared is None or self.stream_online() is not None: + if self._last_reading is None or self.stream_online() is not None: return await self._tracked(self.wrapped_powermeter.get_passive_watts) # A reading served from the cache was already recorded by the read - # that fetched it; recording it again would mark a meter that has - # failed since then as healthy. A failure always comes from the - # source, so it is recorded. - reads_before = shared.source_reads + # that fetched it, and reads that finished while this one waited may + # have recorded something newer; recording it again could mark a + # meter that has failed since as healthy. Only an answer this call got + # from the source is news, and a failure always comes from the source. attempt = self._clock() - try: - result = await self.wrapped_powermeter.get_passive_watts() - except Exception: - self._last_attempt = attempt - self._last_outcome_ok = False - raise - if shared.source_reads != reads_before: + with probe_passive_read() as probe: + try: + result = await self.wrapped_powermeter.get_passive_watts() + except Exception: + self._last_attempt = attempt + self._last_outcome_ok = False + raise + if probe.reached_source: self._last_attempt = attempt self._last_outcome_ok = bool(result) return result diff --git a/src/astrameter/powermeter/wrappers/health_test.py b/src/astrameter/powermeter/wrappers/health_test.py index 05c68329..c3fdb811 100644 --- a/src/astrameter/powermeter/wrappers/health_test.py +++ b/src/astrameter/powermeter/wrappers/health_test.py @@ -1,3 +1,4 @@ +import asyncio from typing import Any from unittest.mock import AsyncMock, Mock @@ -171,3 +172,45 @@ async def test_passive_read_that_reads_the_source_is_recorded() -> None: with pytest.raises(OSError): await pm.get_passive_watts() assert (pm.last_attempt, pm.last_outcome_ok) == (11.0, False) + + +class _PerReadSource(FakePowermeter): + """A polled source whose reads each finish when the test says so.""" + + def __init__(self) -> None: + super().__init__([]) + self._pending: list[asyncio.Future[list[float]]] = [] + + def finish(self, index: int, values: list[float]) -> None: + self._pending[index].set_result(values) + + def fail(self, index: int, error: Exception) -> None: + self._pending[index].set_exception(error) + + async def get_powermeter_watts(self) -> list[float]: + read: asyncio.Future[list[float]] = asyncio.get_running_loop().create_future() + self._pending.append(read) + return await read + + +async def test_passive_read_served_while_waiting_keeps_a_later_failure() -> None: + """A passive reader waiting on battery reads can be served the one that + succeeded after another has already failed; that failure is the newer + fact about the meter, so the passive read must not overwrite it.""" + clock = _FakeClock() + source = _PerReadSource() + pm, _ = _shared(source, clock) + + succeeding = asyncio.create_task(pm.get_powermeter_watts()) + failing = asyncio.create_task(pm.get_powermeter_watts()) + await asyncio.sleep(0) + passive = asyncio.create_task(pm.get_passive_watts()) + await asyncio.sleep(0) + + source.finish(0, [100.0]) + source.fail(1, OSError("timeout")) + assert await succeeding == [100.0] + with pytest.raises(OSError): + await failing + assert await passive == [100.0] + assert pm.last_outcome_ok is False diff --git a/src/astrameter/powermeter/wrappers/last_reading.py b/src/astrameter/powermeter/wrappers/last_reading.py index 697df95b..71ff85eb 100644 --- a/src/astrameter/powermeter/wrappers/last_reading.py +++ b/src/astrameter/powermeter/wrappers/last_reading.py @@ -1,6 +1,8 @@ import asyncio +import contextlib import time -from collections.abc import Callable +from collections.abc import Callable, Iterator +from contextvars import ContextVar from astrameter.powermeter.base import Powermeter, stream_fresh @@ -13,6 +15,34 @@ PASSIVE_READING_MAX_AGE_S = 5.0 +class PassiveReadProbe: + """Says whether one passive read was answered by the source.""" + + reached_source = False + + +_probe: ContextVar[PassiveReadProbe | None] = ContextVar( + "passive_read_probe", default=None +) + + +@contextlib.contextmanager +def probe_passive_read() -> Iterator[PassiveReadProbe]: + """Report whether the passive reads made inside reached the source. + + The health wrapper uses it to tell a reading the source just gave from one + served from the cache, which proves nothing about the meter now. It goes + by the calling task, so reads other tasks make meanwhile, the batteries' + or other passive readers', do not count. + """ + probe = PassiveReadProbe() + token = _probe.set(probe) + try: + yield probe + finally: + _probe.reset(token) + + class LastReadingPowermeter(PowermeterWrapper): """Innermost wrapper that lets passive readers reuse the control loop's read. @@ -55,23 +85,12 @@ def __init__( # ends, so a waiting passive reader looks at the cache afterwards # rather than at the read's outcome. self._in_flight: set[asyncio.Future[None]] = set() - self._source_reads = 0 - - @property - def source_reads(self) -> int: - """How many reads of the source have returned, whatever they returned. - - Lets the health wrapper tell a passive read the source answered from - one served from this cache, which proves nothing about the meter now. - """ - return self._source_reads async def get_powermeter_watts(self) -> list[float]: done: asyncio.Future[None] = asyncio.get_running_loop().create_future() self._in_flight.add(done) try: values = await self.wrapped_powermeter.get_powermeter_watts() - self._source_reads += 1 if values: self._values = list(values) self._read_at = self._clock() @@ -90,6 +109,9 @@ async def get_passive_watts(self) -> list[float]: return fresh pending = {read for read in pending if not read.done()} if not pending: + probe = _probe.get() + if probe is not None: + probe.reached_source = True return await self.get_powermeter_watts() # asyncio.wait never cancels what it waits on, even if this # reader is cancelled, so the reads it waits for run to the end. diff --git a/src/astrameter/powermeter/wrappers/last_reading_test.py b/src/astrameter/powermeter/wrappers/last_reading_test.py index c2245415..cf91bb0a 100644 --- a/src/astrameter/powermeter/wrappers/last_reading_test.py +++ b/src/astrameter/powermeter/wrappers/last_reading_test.py @@ -3,7 +3,7 @@ import pytest from .conftest import FakePowermeter -from .last_reading import LastReadingPowermeter +from .last_reading import LastReadingPowermeter, probe_passive_read class _FakeClock: @@ -270,16 +270,22 @@ async def test_passive_read_does_not_wait_for_reads_started_after_it() -> None: assert await later == [100.0] -async def test_source_reads_counts_every_answer_but_not_cache_hits() -> None: - source = _CountingSource([100.0]) - pm, _ = _make(source) +async def test_probe_reports_only_this_tasks_source_reads() -> None: + source = _SlowSource([100.0]) + pm, clock = _make(source) - await pm.get_powermeter_watts() - await pm.get_passive_watts() - source.set([]) - await pm.get_powermeter_watts() - source.error = ValueError("offline") - with pytest.raises(ValueError): - await pm.get_powermeter_watts() + control = asyncio.create_task(pm.get_powermeter_watts()) + await asyncio.sleep(0) + with probe_passive_read() as probe: + passive = asyncio.create_task(pm.get_passive_watts()) + await asyncio.sleep(0) + source.release() + assert await passive == [100.0] + await control + # Served from the battery read it waited for, not from a read of its own. + assert probe.reached_source is False - assert pm.source_reads == 2 + clock.now = 10.0 + with probe_passive_read() as probe: + assert await pm.get_passive_watts() == [100.0] + assert probe.reached_source is True