diff --git a/src/astrameter/ct002/ct002.py b/src/astrameter/ct002/ct002.py index 2bcdd12f..5081cf34 100644 --- a/src/astrameter/ct002/ct002.py +++ b/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: diff --git a/src/astrameter/main.py b/src/astrameter/main.py index b753c426..d0bac83e 100644 --- a/src/astrameter/main.py +++ b/src/astrameter/main.py @@ -61,6 +61,7 @@ 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. @@ -68,6 +69,10 @@ async def read_ct_powermeter( 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) diff --git a/src/astrameter/mqtt_insights/service.py b/src/astrameter/mqtt_insights/service.py index 259491b0..3af6898e 100644 --- a/src/astrameter/mqtt_insights/service.py +++ b/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(), diff --git a/src/astrameter/powermeter/tibber_pulse.py b/src/astrameter/powermeter/tibber_pulse.py index db7de893..8cab3859 100644 --- a/src/astrameter/powermeter/tibber_pulse.py +++ b/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: diff --git a/src/astrameter/powermeter/wrappers/throttling.py b/src/astrameter/powermeter/wrappers/throttling.py index afb0af55..a5c49a4e 100644 --- a/src/astrameter/powermeter/wrappers/throttling.py +++ b/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