Add wait_for_next_message so CT002 serves fresh powermeter data (#322)

* Add wait_for_next_message so CT002 serves fresh powermeter data

The CT002's before_send callback now awaits wait_for_next_message()
before reading the powermeter, ensuring push-based meters (MQTT, SMA,
HomeWizard, Home Assistant) deliver a fresh measurement for every
battery request instead of re-serving stale cached values.

* Fix wait_for_next_message returning early in multi-topic mode

In multi-topic configurations, wait_for_next_message could return after
receiving a message on just one topic, leaving other phase values as None
and causing get_powermeter_watts() to fail. Now loops with a deadline
until all subscribed values are populated, matching wait_for_message
semantics but always waiting for a new message first.

https://claude.ai/code/session_013EAvo4q396i36rhT45LNkp

* Revert changelog entry for wait_for_next_message fix

https://claude.ai/code/session_013EAvo4q396i36rhT45LNkp

* Assert wait_for_next_message actually blocks in test

Replace `await task` with `assert task.done()` after
wait_for_next_message returns, so the test fails if the method
returns before the setter task completes.

https://claude.ai/code/session_013EAvo4q396i36rhT45LNkp

* Format mqtt_test.py with ruff

https://claude.ai/code/session_013EAvo4q396i36rhT45LNkp

---------

Co-authored-by: Claude <noreply@anthropic.com>

Author: tomquist

src/astrameter/ct002/ct002.py

@@ -457,8 +457,9 @@ def _build_response_fields(self, request_fields, values):
         ]
 
         phase_values = self._collect_reports_by_phase()
+        phase_power = [phase_a, phase_b, phase_c]
         for phase, idx in (("A", 0), ("B", 1), ("C", 2)):
-            if phase_values[phase]["active"]:
+            if phase_values[phase]["active"] or phase_power[idx] != 0:
                 response_fields[8 + idx] = "1"
             response_fields[15 + idx] = str(phase_values[phase]["chrg_power"])
             response_fields[20 + idx] = str(phase_values[phase]["dchrg_power"])
@@ -579,14 +580,13 @@ async def _handle_request(self, data, addr, transport):
             return
         self._last_response_time[addr] = current_time
 
-        if not in_inspection_mode:
-            meter_dev_type = fields[0] if len(fields) > 0 else ""
-            self._update_consumer_report(
-                consumer_id,
-                phase=reported_phase,
-                power=reported_power,
-                device_type=meter_dev_type,
-            )
+        meter_dev_type = fields[0] if len(fields) > 0 else ""
+        self._update_consumer_report(
+            consumer_id,
+            phase=reported_phase if not in_inspection_mode else "A",
+            power=reported_power,
+            device_type=meter_dev_type,
+        )
 
         updated = await self._call_before_send(addr, fields, consumer_id)
         if updated is not None:

src/astrameter/main.py

@@ -211,6 +211,7 @@ async def update_readings(addr, _fields=None, _consumer_id=None):
             if powermeter is None:
                 logger.debug(f"No powermeter found for client {addr[0]}")
                 return None
+            await powermeter.wait_for_next_message()
             values = await powermeter.get_powermeter_watts()
             value1 = values[0] if len(values) > 0 else 0
             value2 = values[1] if len(values) > 1 else 0

src/astrameter/powermeter/base.py

@@ -6,6 +6,15 @@ async def get_powermeter_watts(self) -> list[float]:
     async def wait_for_message(self, timeout=5):
         pass
 
+    async def wait_for_next_message(self, timeout=5):
+        """Block until a *new* measurement arrives (push-based powermeters).
+
+        Unlike ``wait_for_message`` (which returns immediately once data has
+        been received *at least once*), this method waits for the *next*
+        update, ensuring callers always get fresh data.  Polling-based
+        powermeters leave the default no-op.
+        """
+
     # --- Lifecycle (no-op by default, override for push-based powermeters) ---
 
     async def start(self):

src/astrameter/powermeter/homeassistant.py

@@ -77,6 +77,7 @@ def __init__(
         self._session: aiohttp.ClientSession | None = None
         self._ws_task: asyncio.Task[None] | None = None
         self._entities_ready = asyncio.Event()
+        self._message_event = asyncio.Event()
 
     def _collect_entities(self) -> set[str]:
         if self.power_calculate:
@@ -233,6 +234,7 @@ def _update_entity_value(self, entity_id: str, state_val: object) -> None:
             self._entity_values[entity_id] = None
             self._entity_update_time[entity_id] = None
         self._check_entities_ready()
+        self._message_event.set()
 
     def _check_entities_ready(self) -> None:
         ready = all(
@@ -288,3 +290,10 @@ async def wait_for_message(self, timeout: float = 5) -> None:
             await asyncio.wait_for(self._entities_ready.wait(), timeout=timeout)
         except asyncio.TimeoutError:
             raise TimeoutError("Timeout waiting for Home Assistant state") from None
+
+    async def wait_for_next_message(self, timeout: float = 5) -> None:
+        self._message_event.clear()
+        try:
+            await asyncio.wait_for(self._message_event.wait(), timeout=timeout)
+        except asyncio.TimeoutError:
+            raise TimeoutError("Timeout waiting for Home Assistant state") from None

src/astrameter/powermeter/homeassistant_test.py

@@ -1,3 +1,4 @@
+import asyncio
 import json
 from unittest.mock import AsyncMock, patch
 
@@ -359,6 +360,34 @@ async def test_wait_for_message_timeout():
         await pm.wait_for_message(timeout=0)
 
 
+# wait_for_next_message tests
+
+
+async def test_wait_for_next_message_blocks_until_new():
+    pm = _create_powermeter()
+    await _simulate_auth_and_states(
+        pm, [{"entity_id": "sensor.current_power", "state": "100"}]
+    )
+
+    async def _push_later():
+        await asyncio.sleep(0.05)
+        pm._update_entity_value("sensor.current_power", "200")
+
+    task = asyncio.create_task(_push_later())
+    await pm.wait_for_next_message(timeout=2)
+    await task
+    assert await pm.get_powermeter_watts() == [200.0]
+
+
+async def test_wait_for_next_message_timeout():
+    pm = _create_powermeter()
+    await _simulate_auth_and_states(
+        pm, [{"entity_id": "sensor.current_power", "state": "100"}]
+    )
+    with pytest.raises(TimeoutError):
+        await pm.wait_for_next_message(timeout=0)
+
+
 # subscribe_entities entity list test
 
 

src/astrameter/powermeter/homewizard.py

@@ -238,3 +238,10 @@ async def wait_for_message(self, timeout: float = 5) -> None:
             await asyncio.wait_for(self._message_event.wait(), timeout=timeout)
         except asyncio.TimeoutError:
             raise TimeoutError("Timeout waiting for HomeWizard measurement") from None
+
+    async def wait_for_next_message(self, timeout: float = 5) -> None:
+        self._message_event.clear()
+        try:
+            await asyncio.wait_for(self._message_event.wait(), timeout=timeout)
+        except asyncio.TimeoutError:
+            raise TimeoutError("Timeout waiting for HomeWizard measurement") from None

src/astrameter/powermeter/homewizard_test.py

@@ -252,6 +252,27 @@ async def test_wait_for_message_timeout():
         await pm.wait_for_message(timeout=0)
 
 
+async def test_wait_for_next_message_blocks_until_new():
+    pm = _create_powermeter()
+    pm._handle_measurement({"power_w": 100})
+
+    async def _push_later():
+        await asyncio.sleep(0.05)
+        pm._handle_measurement({"power_w": 200})
+
+    task = asyncio.create_task(_push_later())
+    await pm.wait_for_next_message(timeout=2)
+    await task
+    assert await pm.get_powermeter_watts() == [200]
+
+
+async def test_wait_for_next_message_timeout():
+    pm = _create_powermeter()
+    pm._handle_measurement({"power_w": 100})
+    with pytest.raises(TimeoutError):
+        await pm.wait_for_next_message(timeout=0)
+
+
 # --- Category F: Lifecycle ---
 
 

src/astrameter/powermeter/mqtt.py

@@ -172,3 +172,19 @@ async def wait_for_message(self, timeout=5):
                 raise TimeoutError("Timeout waiting for MQTT message") from None
             if all(v is not None for v in self.values):
                 return
+
+    async def wait_for_next_message(self, timeout=5):
+        self._message_event.clear()
+        loop = asyncio.get_running_loop()
+        deadline = loop.time() + timeout
+        while True:
+            remaining = deadline - loop.time()
+            if remaining <= 0:
+                raise TimeoutError("Timeout waiting for MQTT message")
+            try:
+                await asyncio.wait_for(self._message_event.wait(), timeout=remaining)
+            except asyncio.TimeoutError:
+                raise TimeoutError("Timeout waiting for MQTT message") from None
+            if all(v is not None for v in self.values):
+                return
+            self._message_event.clear()

src/astrameter/powermeter/mqtt_test.py

@@ -100,6 +100,50 @@ async def _set_later():
     assert pm.value == 99.0
 
 
+async def test_wait_for_next_message_blocks_until_new():
+    pm = _make_pm()
+    pm.value = 1.0
+    pm._message_event.set()
+
+    async def _set_later():
+        await asyncio.sleep(0.05)
+        pm.value = 42.0
+        pm._message_event.set()
+
+    task = asyncio.create_task(_set_later())
+    await pm.wait_for_next_message(timeout=2)
+    assert task.done(), (
+        "wait_for_next_message returned before the setter task completed"
+    )
+    assert pm.value == 42.0
+
+
+async def test_wait_for_next_message_times_out():
+    pm = _make_pm()
+    pm.value = 1.0
+    pm._message_event.set()
+    with pytest.raises(TimeoutError, match="Timeout waiting"):
+        await pm.wait_for_next_message(timeout=0.1)
+
+
+async def test_wait_for_next_message_multi_topic_cold_start():
+    pm = _make_pm(topic=["t1", "t2"])
+    # Cold start: no values set, event not set
+
+    async def _set_staggered():
+        await asyncio.sleep(0.05)
+        pm.values[0] = 100.0
+        pm._message_event.set()
+        await asyncio.sleep(0.05)
+        pm.values[1] = 200.0
+        pm._message_event.set()
+
+    task = asyncio.create_task(_set_staggered())
+    await pm.wait_for_next_message(timeout=2)
+    await task
+    assert await pm.get_powermeter_watts() == [100.0, 200.0]
+
+
 # ---------------------------------------------------------------------------
 # Multi-phase constructor unit tests
 # ---------------------------------------------------------------------------

src/astrameter/powermeter/pid.py

@@ -95,6 +95,9 @@ async def wait_for_message(self, timeout=5):
         """Pass through to wrapped powermeter."""
         return await self.wrapped_powermeter.wait_for_message(timeout)
 
+    async def wait_for_next_message(self, timeout=5):
+        return await self.wrapped_powermeter.wait_for_next_message(timeout)
+
     async def start(self):
         """Pass through to wrapped powermeter."""
         await self.wrapped_powermeter.start()