Add configurable wait-for-fresh-push behavior for event-driven powermeters (#330)

* Stop wait_for_next_message regressions from breaking battery responses

PR #322 wired wait_for_next_message into update_readings so push-based
powermeters serve fresh data on every CT002 request. Two regressions
followed:

1. Slow event-driven meters (e.g. Home Assistant wrapping a P1 that
   ticks every 10s, or any source with high THROTTLE_INTERVAL) raise
   TimeoutError on every request, which surfaces as "CT002 before_send
   failed" warnings and skips the cached value that would have been
   perfectly usable (issue #327).

2. MQTT multi-topic kept clearing the event whenever the waker happened
   to be a phase that had already published, so any topic still at None
   would force the full 5s timeout even though a fresh message had
   arrived.

Treat wait_for_next_message as best-effort freshness:

- update_readings caps the wait at 2s and swallows TimeoutError (cached
  value is served, which matches behaviour pre-#322).
- Add WAIT_FOR_NEXT_MESSAGE (global [GENERAL] + per-section) so users
  can opt out of the wait entirely for sources that always update
  slower than the cap. Defaults to true.
- MQTT wait_for_next_message returns on any next message instead of
  re-arming the event when not all topics have ever published.

Refactor the closure inside run_device into a module-level
read_ct_powermeter helper so the swallow-timeout behaviour is
unit-testable.

* Revert CHANGELOG entry for wait_for_next_message fix

* Expose WAIT_FOR_NEXT_MESSAGE in the Home Assistant app config

Adds a "Wait For Fresh Home Assistant Reading" toggle (default true) to
the HA app so users with slow-updating source sensors (e.g. P1 smart
meters) can opt out of the up-to-2s freshness wait and serve the
last-known value immediately.

* Honor WAIT_FOR_NEXT_MESSAGE in the Shelly emulator path

The Shelly UDP handler was ignoring the per-powermeter wait flag, so a
Shelly-emulated device would never wait for a fresh push from an
event-driven powermeter (behaviour matched pre-#322). Mirror the CT path:
read the flag when matching addr to powermeter, then best-effort await
``wait_for_next_message(timeout=2)`` with ``TimeoutError`` swallowed.

Also consolidate the duplicate ``[HOMEASSISTANT]`` per-powermeter throttle
example in config.ini.example: keep the more informative
``THROTTLE_INTERVAL = 2`` block (with the "2-3 seconds due to network
latency" note) and drop the stray ``= 1`` duplicate that predated the
``WAIT_FOR_NEXT_MESSAGE`` example.

---------

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

Author: tomquist

README.md

@@ -174,10 +174,20 @@ SKIP_POWERMETER_TEST = False
 # Set to 0 to disable throttling (default). Recommended: 1-3 seconds for slow data sources
 # Can be overridden per powermeter section
 THROTTLE_INTERVAL = 0
+# Briefly wait (up to 2s) for a fresh push from event-driven powermeters
+# (MQTT, Home Assistant, HomeWizard, SMA, ...) before responding to the
+# battery. Set to false to skip the wait and always serve the last-known
+# value — recommended when the underlying meter updates slower than 2s
+# (e.g. P1 smart meter behind Home Assistant) so that the inevitable timeout
+# doesn't add latency to every CT002 response. Default: true.
+# Can be overridden per powermeter section.
+#WAIT_FOR_NEXT_MESSAGE = true
 ```
 
-Per-powermeter options (e.g. in `[TASMOTA]`):
+Per-powermeter options (apply in any powermeter section, e.g. `[TASMOTA]` or `[HOMEASSISTANT]`):
 - **THROTTLE_INTERVAL** — Override global throttling for this powermeter
+- **WAIT_FOR_NEXT_MESSAGE** — Override the global wait-for-fresh-push behaviour
+  for this powermeter (set to `false` to opt out of the wait entirely)
 
 CT002/CT003 active-steering options (all under `[CT002]` or `[CT003]`):
 - **ACTIVE_CONTROL** — When true (default), the emulator smooths the grid reading, splits

config.ini.example

@@ -17,6 +17,15 @@ SKIP_POWERMETER_TEST = False
 # Recommended values: 1-3 seconds for slower data sources
 # Can be overridden per powermeter section
 THROTTLE_INTERVAL = 0
+# Wait briefly (up to 2s) for a fresh push from event-driven powermeters
+# (MQTT, Home Assistant, HomeWizard, SMA energy meter, ...) before responding
+# to the battery. Keeps battery control loops on the freshest data.
+# Set to false to skip the wait entirely and always serve the last-known
+# value — recommended for sources that update slower than 2s (e.g. a P1 smart
+# meter behind Home Assistant) or whenever you'd rather have an instant
+# response than a fresh one. Defaults to true.
+# Can be overridden per powermeter section.
+#WAIT_FOR_NEXT_MESSAGE = true
 
 #[CT002]
 ## CT type is derived from the emulated device (ct002 -> HME-4, ct003 -> HME-3).
@@ -194,12 +203,15 @@ THROTTLE_INTERVAL = 0
 #POWER_INPUT_ALIAS = sensor.phase1_input, sensor.phase2_input, sensor.phase3_input
 #POWER_OUTPUT_ALIAS = sensor.phase1_output, sensor.phase2_output, sensor.phase3_output
 
-## Per-powermeter throttling override (optional)
-#THROTTLE_INTERVAL = 1
-
 ## Per-powermeter throttling override (optional)
 ## HomeAssistant typically needs 2-3 seconds due to network latency
 #THROTTLE_INTERVAL = 2
+## Skip the up-to-2s wait for a fresh push from this powermeter and always
+## serve the last-known value. Useful when the underlying sensor (e.g. P1
+## smart meter) updates slower than 2s, so the wait would always time out
+## and the cached value would be served anyway. Defaults to the global
+## [GENERAL] WAIT_FOR_NEXT_MESSAGE setting (true).
+#WAIT_FOR_NEXT_MESSAGE = false
 
 #[VZLOGGER]
 #IP = 192.168.1.106

ha_addon/config.yaml

@@ -34,6 +34,7 @@ options:
   power_output_alias: ""
   device_types: "shellypro3em"
   throttle_interval: 0
+  wait_for_next_message: true
   marstek_auto_register_ct_device: false
   marstek_mailbox: ""
   marstek_password: ""
@@ -45,6 +46,7 @@ schema:
   power_output_alias: str?
   device_types: str
   throttle_interval: float(0,)
+  wait_for_next_message: bool
   ct_mac: str?
   marstek_auto_register_ct_device: bool?
   marstek_mailbox: str?

ha_addon/run.sh

@@ -126,6 +126,7 @@ else
         echo "PORT=80"
         echo "API_PATH_PREFIX=/core"
         echo "ACCESSTOKEN=$SUPERVISOR_TOKEN"
+        echo "WAIT_FOR_NEXT_MESSAGE=$(bashio::config 'wait_for_next_message')"
         if bashio::config.has_value 'power_output_alias'; then
             echo "POWER_CALCULATE=True"
             echo "POWER_INPUT_ALIAS=$(bashio::config 'power_input_alias')"

ha_addon/translations/en.yaml

@@ -14,6 +14,9 @@ configuration:
   throttle_interval:
     name: Throttling Interval
     description: "Minimum time interval (in seconds) between power meter readings. When set to 0, throttling is disabled for maximum performance. For slower data sources or to prevent control instability, set to 1-3 seconds. Supports decimal values (e.g., 1.5 for 1.5 seconds). Increase this value if you're seeing oscillations or instability with slow data sources."
+  wait_for_next_message:
+    name: Wait For Fresh Home Assistant Reading
+    description: "When enabled (default), the app waits briefly (up to 2 seconds) for Home Assistant to push a fresh value before responding to the battery, keeping control loops on the freshest data. Disable this if your source sensor (e.g. a P1 smart meter) updates slower than 2 seconds — turning it off avoids the needless wait and always serves the last-known value immediately."
   log_level:
     name: Log Level
     description: "Controls the verbosity of log messages. Select 'critical' for only critical errors, 'error' for error messages and above, 'warning' for warnings and above, 'info' for informational messages and above (recommended), or 'debug' for detailed debugging information. Use 'debug' for troubleshooting connection or configuration issues."

src/astrameter/config/config_loader.py

@@ -145,11 +145,14 @@ def parse_float_list(value: str, key_name: str, section: str) -> list[float]:
 
 def read_all_powermeter_configs(
     config: configparser.ConfigParser,
-) -> list[tuple[Powermeter, ClientFilter]]:
-    powermeters = []
+) -> list[tuple[Powermeter, ClientFilter, bool]]:
+    powermeters: list[tuple[Powermeter, ClientFilter, bool]] = []
     global_throttle_interval = config.getfloat(
         "GENERAL", "THROTTLE_INTERVAL", fallback=0.0
     )
+    global_wait_for_next_message = config.getboolean(
+        "GENERAL", "WAIT_FOR_NEXT_MESSAGE", fallback=True
+    )
     global_pid_kp = config.getfloat("GENERAL", "PID_KP", fallback=0.0)
     global_pid_ki = config.getfloat("GENERAL", "PID_KI", fallback=0.0)
     global_pid_kd = config.getfloat("GENERAL", "PID_KD", fallback=0.0)
@@ -237,7 +240,10 @@ def read_all_powermeter_configs(
                 )
 
             client_filter = create_client_filter(section, config)
-            powermeters.append((powermeter, client_filter))
+            wait_for_next_message = config.getboolean(
+                section, "WAIT_FOR_NEXT_MESSAGE", fallback=global_wait_for_next_message
+            )
+            powermeters.append((powermeter, client_filter, wait_for_next_message))
     return powermeters
 
 

src/astrameter/config/config_loader_test.py

@@ -583,7 +583,7 @@ def test_read_all_configs_with_power_transform():
 
     powermeters = read_all_powermeter_configs(config)
     assert len(powermeters) == 1
-    pm, _ = powermeters[0]
+    pm, _, _ = powermeters[0]
     assert isinstance(pm, TransformedPowermeter)
     assert pm.offsets == [-50.0]
     assert pm.multipliers == [1.05]
@@ -600,7 +600,7 @@ def test_read_all_configs_with_per_phase_transform():
 
     powermeters = read_all_powermeter_configs(config)
     assert len(powermeters) == 1
-    pm, _ = powermeters[0]
+    pm, _, _ = powermeters[0]
     assert isinstance(pm, TransformedPowermeter)
     assert pm.offsets == [-10.0, -20.0, -30.0]
     assert pm.multipliers == [1.05, 1.02, 1.03]
@@ -616,7 +616,7 @@ def test_read_all_configs_offset_only():
 
     powermeters = read_all_powermeter_configs(config)
     assert len(powermeters) == 1
-    pm, _ = powermeters[0]
+    pm, _, _ = powermeters[0]
     assert isinstance(pm, TransformedPowermeter)
     assert pm.offsets == [10.0]
     assert pm.multipliers == [1.0]
@@ -632,7 +632,7 @@ def test_read_all_configs_zero_multiplier_accepted():
 
     powermeters = read_all_powermeter_configs(config)
     assert len(powermeters) == 1
-    pm, _ = powermeters[0]
+    pm, _, _ = powermeters[0]
     assert isinstance(pm, TransformedPowermeter)
     assert pm.multipliers == [0.0]
 
@@ -646,5 +646,42 @@ def test_read_all_configs_no_transform_when_not_configured():
 
     powermeters = read_all_powermeter_configs(config)
     assert len(powermeters) == 1
-    pm, _ = powermeters[0]
+    pm, _, _ = powermeters[0]
     assert not isinstance(pm, TransformedPowermeter)
+
+
+def test_read_all_configs_wait_for_next_message_default_true():
+    """No config means waiting is enabled (preserves PR #322 behaviour)."""
+    config = configparser.ConfigParser()
+    config["SCRIPT_1"] = {"COMMAND": 'echo "100"'}
+    powermeters = read_all_powermeter_configs(config)
+    assert len(powermeters) == 1
+    _, _, wait_for_next = powermeters[0]
+    assert wait_for_next is True
+
+
+def test_read_all_configs_wait_for_next_message_global_off():
+    """[GENERAL] WAIT_FOR_NEXT_MESSAGE=false applies to every section."""
+    config = configparser.ConfigParser()
+    config["GENERAL"] = {"WAIT_FOR_NEXT_MESSAGE": "false"}
+    config["SCRIPT_1"] = {"COMMAND": 'echo "100"'}
+    config["SCRIPT_2"] = {"COMMAND": 'echo "200"'}
+    powermeters = read_all_powermeter_configs(config)
+    assert len(powermeters) == 2
+    assert all(wait is False for _, _, wait in powermeters)
+
+
+def test_read_all_configs_wait_for_next_message_section_override():
+    """Per-section WAIT_FOR_NEXT_MESSAGE overrides the global default."""
+    config = configparser.ConfigParser()
+    config["GENERAL"] = {"WAIT_FOR_NEXT_MESSAGE": "true"}
+    config["SCRIPT_1"] = {
+        "COMMAND": 'echo "100"',
+        "WAIT_FOR_NEXT_MESSAGE": "false",
+    }
+    config["SCRIPT_2"] = {"COMMAND": 'echo "200"'}
+    powermeters = read_all_powermeter_configs(config)
+    assert len(powermeters) == 2
+    # 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]

src/astrameter/main.py

@@ -33,6 +33,43 @@ def get_ct_section(device_type: str, cfg: configparser.ConfigParser) -> str:
     return section
 
 
+async def read_ct_powermeter(
+    addr: tuple[str, int],
+    powermeters: list[tuple[Powermeter, ClientFilter, bool]],
+) -> list[float] | None:
+    """Pick the powermeter matching *addr* and return up to three phase values.
+
+    Optionally awaits a fresh push (with a 2 s cap) when the matched
+    powermeter has ``WAIT_FOR_NEXT_MESSAGE`` enabled. A timeout there is
+    swallowed so the cached value is still served — `update_readings`
+    callers should never see a stale-meter `TimeoutError`.
+    """
+    powermeter = None
+    wait_for_next = False
+    for pm, client_filter, wait_flag in powermeters:
+        if client_filter.matches(addr[0]):
+            powermeter = pm
+            wait_for_next = wait_flag
+            break
+    if powermeter is None:
+        logger.debug(f"No powermeter found for client {addr[0]}")
+        return None
+    if wait_for_next:
+        try:
+            await powermeter.wait_for_next_message(timeout=2)
+        except TimeoutError:
+            logger.debug(
+                "Powermeter %s produced no fresh message within 2s; "
+                "serving last known value",
+                type(powermeter).__name__,
+            )
+    values = await powermeter.get_powermeter_watts()
+    value1 = values[0] if len(values) > 0 else 0
+    value2 = values[1] if len(values) > 1 else 0
+    value3 = values[2] if len(values) > 2 else 0
+    return [value1, value2, value3]
+
+
 async def test_powermeter(powermeter: Powermeter, client_filter: ClientFilter):
     """Test powermeter configuration with minimal retry logic for edge cases."""
     max_retries = 3
@@ -70,7 +107,7 @@ async def run_device(
     device_type: str,
     cfg: configparser.ConfigParser,
     args: argparse.Namespace,
-    powermeters: list[tuple[Powermeter, ClientFilter]],
+    powermeters: list[tuple[Powermeter, ClientFilter, bool]],
     device_id: str | None = None,
     insights: MqttInsightsService | None = None,
 ):
@@ -203,21 +240,7 @@ async def run_device(
         )
 
         async def update_readings(addr, _fields=None, _consumer_id=None):
-            powermeter = None
-            for pm, client_filter in powermeters:
-                if client_filter.matches(addr[0]):
-                    powermeter = pm
-                    break
-            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
-            value3 = values[2] if len(values) > 2 else 0
-
-            return [value1, value2, value3]
+            return await read_ct_powermeter(addr, powermeters)
 
         device.before_send = update_readings
 
@@ -336,19 +359,19 @@ async def async_main(
                 await web_server.stop()
             web_server = None
 
-    powermeters: list[tuple[Powermeter, ClientFilter]] = []
+    powermeters: list[tuple[Powermeter, ClientFilter, bool]] = []
     insights: MqttInsightsService | None = None
 
     try:
         # Create powermeters
         powermeters = read_all_powermeter_configs(cfg)
 
         # Start powermeter lifecycle
-        for pm, _ in powermeters:
+        for pm, _, _ in powermeters:
             await pm.start()
 
         if not skip_test:
-            for powermeter, client_filter in powermeters:
+            for powermeter, client_filter, _ in powermeters:
                 await test_powermeter(powermeter, client_filter)
 
         # MQTT Insights (optional)
@@ -379,7 +402,7 @@ async def async_main(
                 logger.info("MQTT Insights service stopped")
             except Exception:
                 logger.exception("Error stopping MQTT Insights service")
-        for pm, _ in powermeters:
+        for pm, _, _ in powermeters:
             try:
                 await pm.stop()
             except Exception:

src/astrameter/main_test.py

@@ -0,0 +1,68 @@
+from ipaddress import IPv4Network
+
+from astrameter.config.config_loader import ClientFilter
+from astrameter.main import read_ct_powermeter
+from astrameter.powermeter import Powermeter
+
+
+class _StubPowermeter(Powermeter):
+    """Minimal powermeter stub for testing ``read_ct_powermeter``."""
+
+    def __init__(
+        self,
+        values: list[float],
+        wait_raises: BaseException | None = None,
+        wait_calls: list[float] | None = None,
+    ):
+        self._values = values
+        self._wait_raises = wait_raises
+        self._wait_calls = wait_calls if wait_calls is not None else []
+
+    async def get_powermeter_watts(self) -> list[float]:
+        return list(self._values)
+
+    async def wait_for_next_message(self, timeout=5):
+        self._wait_calls.append(timeout)
+        if self._wait_raises is not None:
+            raise self._wait_raises
+
+
+_LOCAL = ClientFilter([IPv4Network("127.0.0.1/32")])
+
+
+async def test_read_ct_powermeter_returns_none_when_no_match():
+    pm = _StubPowermeter([10.0])
+    powermeters = [(pm, _LOCAL, True)]
+    assert await read_ct_powermeter(("10.0.0.1", 0), powermeters) is None
+
+
+async def test_read_ct_powermeter_pads_to_three_phases():
+    pm = _StubPowermeter([42.0])
+    powermeters = [(pm, _LOCAL, False)]
+    assert await read_ct_powermeter(("127.0.0.1", 0), powermeters) == [42.0, 0, 0]
+
+
+async def test_read_ct_powermeter_skips_wait_when_disabled():
+    pm = _StubPowermeter([1.0, 2.0, 3.0])
+    powermeters = [(pm, _LOCAL, False)]
+    result = await read_ct_powermeter(("127.0.0.1", 0), powermeters)
+    assert result == [1.0, 2.0, 3.0]
+    assert pm._wait_calls == []
+
+
+async def test_read_ct_powermeter_calls_wait_with_2s_when_enabled():
+    pm = _StubPowermeter([1.0, 2.0, 3.0])
+    powermeters = [(pm, _LOCAL, True)]
+    await read_ct_powermeter(("127.0.0.1", 0), powermeters)
+    assert pm._wait_calls == [2]
+
+
+async def test_read_ct_powermeter_swallows_timeout_and_serves_cached():
+    """Issue #327: a slow push meter must not break CT002 responses."""
+    pm = _StubPowermeter(
+        [11.0, 22.0, 33.0],
+        wait_raises=TimeoutError("simulated slow meter"),
+    )
+    powermeters = [(pm, _LOCAL, True)]
+    result = await read_ct_powermeter(("127.0.0.1", 0), powermeters)
+    assert result == [11.0, 22.0, 33.0]

src/astrameter/powermeter/mqtt.py

@@ -175,16 +175,7 @@ async def wait_for_message(self, timeout=5):
 
     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()
+        try:
+            await asyncio.wait_for(self._message_event.wait(), timeout=timeout)
+        except asyncio.TimeoutError:
+            raise TimeoutError("Timeout waiting for MQTT message") from None