Feat: Support graceful fallback target (TIMEOUT_FALLBACK_W)

src/astrameter/ct002/ct002.py

@@ -227,9 +227,11 @@ def __init__(
         saturation_grace_seconds=SATURATION_GRACE_SECONDS,
         saturation_stall_timeout_seconds=SATURATION_STALL_TIMEOUT_SECONDS,
         device_id="",
+        timeout_fallback_w=None,
         clock=None,
         reset_fn=None,
     ) -> None:
+        self.timeout_fallback_w = timeout_fallback_w
         self.udp_port = udp_port
         self.ct_mac = ct_mac
         self.ct_type = ct_type
@@ -898,28 +900,64 @@ async def _call_before_send(self, addr, fields, consumer_id):
             # the same rate-limit behaviour as production.
             self._before_send_failure_count += 1
             now = self._clock()
-            if (
-                self._before_send_failure_count == 1
-                or now - self._before_send_last_warn >= 30.0
-            ):
-                logger.warning(
-                    "CT002 before_send failed (%d in a row) for %s: %s. "
-                    "The CT002 emulator is sending a zero adjustment so "
-                    "batteries hold their current output until the "
-                    "powermeter recovers.",
-                    self._before_send_failure_count,
-                    addr,
-                    exc,
-                    exc_info=debug_traceback(),
-                )
+            if self._before_send_failure_count == 1:
+                # One-shot engagement log — easy to grep / alert on.
+                if self.timeout_fallback_w is not None:
+                    logger.warning(
+                        "CT002: Fallback ENGAGED for %s. "
+                        "Powermeter offline (%s), targeting %gW.",
+                        addr,
+                        exc,
+                        self.timeout_fallback_w,
+                        exc_info=False,
+                    )
+                else:
+                    logger.warning(
+                        "CT002: Powermeter offline for %s (%s). "
+                        "Holding current output until recovery.",
+                        addr,
+                        exc,
+                        exc_info=debug_traceback(),
+                    )
+                self._before_send_last_warn = now
+            elif now - self._before_send_last_warn >= 30.0:
+                if self.timeout_fallback_w is not None:
+                    logger.warning(
+                        "CT002 before_send still failing (%d in a row) for %s: %s. "
+                        "Fallback active, targeting %gW.",
+                        self._before_send_failure_count,
+                        addr,
+                        exc,
+                        self.timeout_fallback_w,
+                        exc_info=False,
+                    )
+                else:
+                    logger.warning(
+                        "CT002 before_send still failing (%d in a row) for %s: %s. "
+                        "Holding current output.",
+                        self._before_send_failure_count,
+                        addr,
+                        exc,
+                        exc_info=debug_traceback(),
+                    )
                 self._before_send_last_warn = now
             return None, True
         # Success path: if we were in a failure spell, log the recovery.
         if self._before_send_failure_count > 0:
-            logger.info(
-                "CT002 before_send recovered after %d consecutive failures",
-                self._before_send_failure_count,
-            )
+            if self.timeout_fallback_w is not None:
+                logger.warning(
+                    "CT002: Fallback DISENGAGED for %s. "
+                    "Powermeter recovered after %d failure(s).",
+                    addr,
+                    self._before_send_failure_count,
+                    exc_info=False,
+                )
+            else:
+                logger.info(
+                    "CT002 before_send recovered after %d consecutive failures for %s",
+                    self._before_send_failure_count,
+                    addr,
+                )
             self._before_send_failure_count = 0
             self._before_send_last_warn = 0.0
         return result, False
@@ -1046,15 +1084,32 @@ async def _handle_request(self, data, addr, transport):
 
             if meter_failed:
                 # Powermeter unavailable: do NOT re-drive control from the stale
-                # cached reading.  The CT002 instruction is a delta
+                # cached reading. The CT002 instruction is a delta
                 # (``new_target = current_power + grid_field``), so re-issuing a
                 # delta derived from a frozen reading winds the battery up in
-                # active control, and feeds frozen per-phase values into a phase
-                # self-diagnosis in inspection mode (issue #403).  Send a zero
-                # adjustment instead so each battery holds its current output —
-                # matching the ESPHome component, which uses ``[0, 0, 0]`` when
-                # its sensor ages out (see esphome/components/ct002/ct002.cpp).
-                values = [0, 0, 0]
+                # active control.
+                if self.timeout_fallback_w is not None:
+                    # Calculate the precise delta required to reach the fallback
+                    # target based on the battery's currently reported output.
+                    delta_w = self.timeout_fallback_w - reported_power
+                    if reported_phase in ("A", "B", "C"):
+                        # Per-phase battery reads only its phase field; place
+                        # the full delta there so it converges in one step.
+                        phase_idx = {"A": 0, "B": 1, "C": 2}[reported_phase]
+                        values = [0.0, 0.0, 0.0]
+                        values[phase_idx] = float(delta_w)
+                    else:
+                        # Combined ("D") battery reads the summed field.
+                        values = [delta_w / 3.0, delta_w / 3.0, delta_w / 3.0]
+                    logger.debug(
+                        "CT002: Fallback delta. Target: %gW, Reported: %gW, "
+                        "Sending delta: %gW (phase %s)",
+                        self.timeout_fallback_w, reported_power, delta_w,
+                        reported_phase or "D",
+                    )
+                else:
+                    # Old behaviour: Send a zero adjustment instead so each battery holds its current output
+                    values = [0, 0, 0]
             else:
                 values = self._get_consumer_value(consumer_id)
                 if values is None:

src/astrameter/main.py

@@ -61,13 +61,18 @@ def get_ct_section(device_type: str, cfg: configparser.ConfigParser) -> str:
 async def read_ct_powermeter(
     addr: tuple[str, int],
     powermeters: list[tuple[Powermeter, ClientFilter, bool]],
+    bypass_cache: bool = False,
 ) -> 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`.
+
+    When *bypass_cache* is True the matching :class:`ThrottledPowermeter`
+    wrapper (if any) is told to propagate errors instead of returning
+    stale cached values, so the caller's fallback logic can activate.
     """
     powermeter = None
     wait_for_next = False
@@ -79,6 +84,14 @@ async def read_ct_powermeter(
     if powermeter is None:
         logger.debug(f"No powermeter found for client {addr[0]}")
         return None
+    # Lazily enable bypass on the wrapper chain for this specific
+    # powermeter so other powermeters in the shared list are unaffected.
+    if bypass_cache:
+        node = powermeter
+        while hasattr(node, "wrapped_powermeter"):
+            if hasattr(node, "bypass_cache_on_error"):
+                node.bypass_cache_on_error = True
+            node = node.wrapped_powermeter
     if wait_for_next:
         try:
             await powermeter.wait_for_next_message(timeout=2)
@@ -95,8 +108,18 @@ async def read_ct_powermeter(
     return [value1, value2, value3]
 
 
-async def test_powermeter(powermeter: Powermeter, client_filter: ClientFilter):
-    """Test powermeter configuration with minimal retry logic for edge cases."""
+async def test_powermeter(
+    powermeter: Powermeter,
+    client_filter: ClientFilter,
+    allow_failure: bool = False,
+):
+    """Test powermeter configuration with minimal retry logic for edge cases.
+
+    When *allow_failure* is True (e.g. because a TIMEOUT_FALLBACK_W is
+    configured), a persistent test failure logs an error but does **not**
+    crash the startup — the control loop's own fallback logic will handle
+    the dead meter at runtime.
+    """
     max_retries = 3
     retry_delay = 5  # seconds
 
@@ -123,6 +146,15 @@ async def test_powermeter(powermeter: Powermeter, client_filter: ClientFilter):
                 continue
             else:
                 # Last attempt failed
+                if allow_failure:
+                    logger.error(
+                        "Failed to test powermeter after %d attempts: %s. "
+                        "Continuing because TIMEOUT_FALLBACK_W is configured.",
+                        max_retries + 1,
+                        e,
+                        exc_info=False,
+                    )
+                    return
                 raise RuntimeError(
                     f"Failed to test powermeter after {max_retries + 1} attempts: {e}"
                 ) from e
@@ -241,6 +273,14 @@ async def run_device(
             ct_section, "SATURATION_DECAY_FACTOR", fallback=0.995
         )
         min_dc_output = cfg.getfloat(ct_section, "MIN_DC_OUTPUT", fallback=0.0)
+        timeout_fallback_raw = cfg.get(ct_section, "TIMEOUT_FALLBACK_W", fallback=None)
+        timeout_fallback_w = float(timeout_fallback_raw) if timeout_fallback_raw is not None else None
+        if timeout_fallback_w is not None and timeout_fallback_w < min_dc_output:
+            logger.warning(
+                "TIMEOUT_FALLBACK_W (%gW) is below MIN_DC_OUTPUT (%gW). Clamping fallback to %gW.",
+                timeout_fallback_w, min_dc_output, min_dc_output
+            )
+            timeout_fallback_w = min_dc_output
         if 0 < min_dc_output < min_target_for_saturation:
             logger.warning(
                 "MIN_DC_OUTPUT (%gW) is below MIN_TARGET_FOR_SATURATION (%dW): a "
@@ -317,11 +357,14 @@ async def run_device(
             min_dc_output=min_dc_output,
             saturation_decay_factor=saturation_decay_factor,
             device_id=device_id or "",
+            timeout_fallback_w=timeout_fallback_w,
             reset_fn=lambda: _reset_all_powermeters(powermeters),
         )
 
         async def update_readings(addr, _fields=None, _consumer_id=None):
-            return await read_ct_powermeter(addr, powermeters)
+            return await read_ct_powermeter(
+                addr, powermeters, bypass_cache=timeout_fallback_w is not None
+            )
 
         device.before_send = update_readings
 
@@ -516,7 +559,38 @@ async def _cloud_gather(
                         await asyncio.wait_for(
                             chosen.wait_for_next_message(), timeout=2.0
                         )
-                    vs = await chosen.get_powermeter_watts_raw()
+                    try:
+                        vs = await chosen.get_powermeter_watts_raw()
+                        chosen.in_fallback_mode = False
+                    except Exception as e:
+                        min_dc = cfg.getfloat("CT002", "MIN_DC_OUTPUT", fallback=0.0)
+                        timeout_fallback_raw = cfg.get("CT002", "TIMEOUT_FALLBACK_W", fallback=None)
+                        timeout_fallback_w = float(timeout_fallback_raw) if timeout_fallback_raw is not None else 0.0
+                        if timeout_fallback_raw is not None and timeout_fallback_w < min_dc:
+                            timeout_fallback_w = min_dc
+                        if not getattr(chosen, "in_fallback_mode", False):
+                            if timeout_fallback_raw is not None:
+                                logger.warning(
+                                    "Powermeter %s failed (%s: %s). Using fallback %gW",
+                                    getattr(chosen, 'name', 'unknown'),
+                                    type(e).__name__,
+                                    e,
+                                    timeout_fallback_w,
+                                    exc_info=False,
+                                )
+                            else:
+                                logger.warning(
+                                    "Powermeter %s failed (%s: %s). Holding output (no fallback configured).",
+                                    getattr(chosen, 'name', 'unknown'),
+                                    type(e).__name__,
+                                    e,
+                                    exc_info=False,
+                                )
+                        chosen.in_fallback_mode = True
+                        # Prevent spin-loop DoS on persistent network errors
+                        await asyncio.sleep(2.0)
+                        # Powermeter timed out. Inject safe fallback target to prevent drain.
+                        vs = [timeout_fallback_w / 3.0, timeout_fallback_w / 3.0, timeout_fallback_w / 3.0]
                     phases = [float(vs[i]) if i < len(vs) else 0.0 for i in range(3)]
                 ap, bp, cp = (round(p) for p in phases)
                 buckets = _dev.reporting_phase_buckets()
@@ -622,8 +696,21 @@ async def async_main(
             await pm.start()
 
         if not skip_test:
+            # Intentionally broad: if *any* device section configures a
+            # fallback target, we allow all meters to fail startup.  The
+            # alternative (crashing on a dead meter while another device
+            # is ready to handle it gracefully) would be worse.  Per-meter
+            # scoping would require config_loader to associate powermeters
+            # with specific device sections, which it currently doesn't.
+            has_fallback = any(
+                cfg.get(s, "TIMEOUT_FALLBACK_W", fallback=None) is not None
+                for s in cfg.sections()
+                if s.upper().startswith("CT")
+            )
             for powermeter, client_filter, _ in powermeters:
-                await test_powermeter(powermeter, client_filter)
+                await test_powermeter(
+                    powermeter, client_filter, allow_failure=has_fallback
+                )
 
         # MQTT Insights (optional)
         insights_cfg = read_mqtt_insights_config(cfg)

src/astrameter/mqtt_insights/service.py

@@ -1002,8 +1002,9 @@ async def _powermeter_health_loop(self, client: aiomqtt.Client) -> None:
                 if not name:
                     continue
                 online, values = await self._powermeter_status(pm)
+                fallback = getattr(pm, "in_fallback_mode", False)
                 await self._publish_powermeter_health(
-                    client, base, cfg, name, online, values
+                    client, base, cfg, name, online, values, fallback
                 )
             await asyncio.sleep(interval)
 
@@ -1067,9 +1068,14 @@ async def _publish_powermeter_health(
         name: str,
         online: bool,
         values: list[float] | None,
+        fallback_active: bool = False,
     ) -> None:
         pm_id = _sanitize_id(name)
-        state = {"online": online, "grid_power": self._grid_power_payload(values)}
+        state = {
+            "online": online, 
+            "grid_power": self._grid_power_payload(values),
+            "fallback_active": fallback_active
+        }
         await client.publish(
             f"{base}/powermeter/{pm_id}",
             payload=json.dumps(state).encode(),

src/astrameter/powermeter/tibber_pulse.py

@@ -68,11 +68,12 @@ def __init__(
     async def start(self) -> None:
         if self.session:
             return
-        # Fail fast: the battery polls ~1/s, so a slow source should error
-        # quickly and let the next poll retry rather than pin a handler.
+        # Give the Pulse enough time to respond over WiFi — a transient
+        # slowdown shouldn't trigger fallback mode.  Genuine failures (HTTP
+        # errors, undecodable SML) surface quickly regardless of timeout.
         self.session = aiohttp.ClientSession(
             auth=BasicAuth(self.user, self.password),
-            timeout=ClientTimeout(total=2, connect=1),
+            timeout=ClientTimeout(total=5, connect=3),
         )
 
     async def stop(self) -> None:

src/astrameter/powermeter/wrappers/throttling.py

@@ -22,6 +22,12 @@ def __init__(self, wrapped_powermeter: Powermeter, throttle_interval: float = 0.
         super().__init__(wrapped_powermeter)
         self.throttle_interval = throttle_interval
 
+        # When True, errors always propagate instead of returning cached
+        # values.  Set by the caller (e.g. CT002 when TIMEOUT_FALLBACK_W is
+        # configured) so the control loop can activate its own fallback
+        # target instead of silently holding on stale data.
+        self.bypass_cache_on_error: bool = False
+
         # Coalescing fetch pattern: when a fetch is in flight (including the
         # throttle sleep), concurrent callers await the same future so every
         # consumer gets fresh data without hammering the source.
@@ -74,7 +80,7 @@ async def get_powermeter_watts(self) -> list[float]:
             # Update timestamp even on failure so we respect the throttle
             # interval before retrying — avoids hammering a failing source.
             self._last_update_time = time.monotonic()
-            if self._last_values is not None:
+            if self._last_values is not None and not self.bypass_cache_on_error:
                 logger.warning(
                     "Throttling: Error getting fresh values: %s", e, exc_info=True
                 )
@@ -88,6 +94,7 @@ async def get_powermeter_watts(self) -> list[float]:
                 return cached
             if not self._pending_fetch.done():
                 self._pending_fetch.set_exception(e)
+                self._pending_fetch.exception()  # Retrieve to prevent asyncio log spam
             raise
         finally:
             self._pending_fetch = None