fix: preserve partial PI30 reads and reconcile cloud recovery

Author: Maksym Shtembuliak

CHANGELOG.md

@@ -81,6 +81,20 @@ the GitHub release body should be rendered from the matching version section her
 
 ### Fixed
 
+- PI30 polling preserves successful replies when another command times out,
+  returns NAK or has an invalid payload (#57). Required readings are carried
+  for a bounded period and then withdrawn individually; their commands remain
+  enabled. Diagnostics identify missed commands and sample age. Empty reads
+  no longer advance runtime success time or confirm a recovered session.
+
+- Collector connection recovery after cloud learning/proxy capture checks live
+  endpoint and apply state before repeating any changes (#46). An already
+  restored connection no longer needs another endpoint write. Recovery excludes
+  background polling, checks uncertain write outcomes without resending them,
+  and retains the recovery warning when application cannot be confirmed.
+  Support diagnostics distinguish write stages and retain the last failure
+  even after a successful verification read.
+
 - Device learning keeps the original cloud API when the collector is temporarily
   redirected to Home Assistant. A local port or AT traffic cannot replace a
   known ValueCloud, SmartValue or SmartESS server with another provider (#46).

custom_components/eybond_local/collector/management.py

@@ -111,6 +111,7 @@ def __init__(self, message: str, *, query_parameter: int | None = None) -> None:
         super().__init__(message)
         # Numeric request context only; never endpoint values or response text.
         self.query_parameter = query_parameter
+        self.operation_phase = ""
 
     @property
     def diagnostic_code(self) -> str:
@@ -120,27 +121,46 @@ def diagnostic_code(self) -> str:
             return type(self.__cause__).__name__
         # Directly constructed typed failures may already carry a class code.
         code = str(self)
-        if code in {failure.__name__ for failure in _TRANSPORT_FAILURES}:
+        if code in {
+            "collector_management_session_changed",
+            *(failure.__name__ for failure in _TRANSPORT_FAILURES),
+        }:
             return code
         return type(self).__name__
 
     @property
     def request_diagnostics(self) -> dict[str, object]:
         """Project wire context here so runtime need not interpret parameters."""
 
-        if self.query_parameter is None:
-            return {}
-        return {
-            "protocol": "eybond_framed",
-            "function": 2,
-            "parameter": self.query_parameter,
-        }
+        result: dict[str, object] = {}
+        if self.query_parameter is not None:
+            result.update(protocol="eybond_framed", function=2, parameter=self.query_parameter)
+        if self.operation_phase:
+            result["phase"] = self.operation_phase
+        return result
 
 
 class CollectorManagementConfirmationError(CollectorManagementError):
     """A management command was sent but the collector did not confirm it."""
 
 
+async def _endpoint_step(phase: str, request):
+    """Annotate compound-operation failures without leaking endpoint values."""
+
+    try:
+        return await request
+    except CollectorManagementTransportError as exc:
+        exc.operation_phase = phase
+        raise
+    except CollectorManagementError:
+        raise
+    except Exception as exc:
+        wrapped = _wrap_wire_call(exc)
+        if isinstance(wrapped, CollectorManagementTransportError):
+            wrapped.operation_phase = phase
+        raise wrapped from exc
+
+
 # ---------------------------------------------------------------------------
 # Normalized models
 # ---------------------------------------------------------------------------
@@ -458,26 +478,29 @@ async def async_write_endpoint(
         self, endpoint: str, *, apply_changes: bool = True
     ) -> CollectorEndpointWriteResult:
         session, _ = self._session()
-        previous = await self._query_flag(session, SET_SERVER_ENDPOINT)
-        try:
-            set_response = await session.set_collector(SET_SERVER_ENDPOINT, endpoint)
-        except CollectorManagementError:
-            raise
-        except Exception as exc:  # noqa: BLE001
-            raise _wrap_wire_call(exc) from exc
+        previous = await _endpoint_step(
+            "read_previous", self._query_flag(session, SET_SERVER_ENDPOINT),
+        )
+        set_response = await _endpoint_step(
+            "write_endpoint", session.set_collector(SET_SERVER_ENDPOINT, endpoint),
+        )
         if set_response.status != 0 or set_response.parameter != SET_SERVER_ENDPOINT:
             raise CollectorManagementConfirmationError(
                 f"collector_set_unconfirmed:parameter={SET_SERVER_ENDPOINT}:"
                 f"status={set_response.status}"
             )
-        readback = await self._query_confirmed(session, SET_SERVER_ENDPOINT)
-        reboot_required = await self._query_flag(session, QUERY_REBOOT_REQUIRED)
+        readback = await _endpoint_step(
+            "readback", self._query_confirmed(session, SET_SERVER_ENDPOINT),
+        )
+        reboot_required = await _endpoint_step(
+            "read_apply_state", self._query_flag(session, QUERY_REBOOT_REQUIRED),
+        )
         confirmation_source = _confirm_write(endpoint, readback, ack_source="set_ack")
 
         apply_performed = False
         warnings: tuple[str, ...] = ()
         if apply_changes:
-            await self._apply(session)
+            await _endpoint_step("apply", self._apply(session))
             apply_performed = True
             warnings = (self._APPLY_WARNING,)
 
@@ -741,20 +764,24 @@ async def async_write_endpoint(
         self, endpoint: str, *, apply_changes: bool = True
     ) -> CollectorEndpointWriteResult:
         transport = self._resolve_transport(needs_write=True)
-        previous = await self._query(transport, _AT_ENDPOINT_COMMAND)
+        previous = await _endpoint_step(
+            "read_previous", self._query(transport, _AT_ENDPOINT_COMMAND),
+        )
         # CLDSRVHOST1 write ack: a W000 status OR a documented exact endpoint echo.
         # A Wxxx != W000 is NOT an ack (readback must then confirm).
-        write_value = await self._send_write(transport, _AT_ENDPOINT_COMMAND, endpoint)
+        write_value = await _endpoint_step(
+            "write_endpoint", self._send_write(transport, _AT_ENDPOINT_COMMAND, endpoint),
+        )
         write_ack = write_value == _AT_SUCCESS_STATUS or _endpoints_match(write_value, endpoint)
-        readback = await self._query(transport, _AT_ENDPOINT_COMMAND)
+        readback = await _endpoint_step("readback", self._query(transport, _AT_ENDPOINT_COMMAND))
         confirmation_source = _confirm_at_write(endpoint, readback, write_ack=write_ack)
 
         warnings: tuple[str, ...] = ()
         extra: dict[str, object] = {}
         apply_performed = False
         if apply_changes:
             # An unconfirmed apply raises (never swallowed into a warning).
-            extra["at_apply_response"] = await self._apply(transport)
+            extra["at_apply_response"] = await _endpoint_step("apply", self._apply(transport))
             apply_performed = True
             warnings = (self._APPLY_WARNING,)
 

custom_components/eybond_local/drivers/pi30.py

@@ -40,6 +40,7 @@
     parse_qt_clock,
     parse_serial_number,
     q1_output_keys,
+    qpigs_output_keys,
     qpiws_output_keys,
 )
 from ..poll_policy import PollPolicy
@@ -101,6 +102,12 @@ class Pi30CommandSpec:
     Pi30CommandSpec(command="Q1", parser=parse_q1, optional=True),
 )
 
+# A transient miss can carry a prior runtime sample for at most two minutes,
+# or three configured poll intervals for deliberately slow polling. This is
+# runtime-only freshness bookkeeping, not a second measurement cache.
+_REQUIRED_SAMPLE_MAX_AGE_SECONDS = 120.0
+_REQUIRED_LAST_SUCCESS_KEY = "pi30_required_last_success"
+
 # A minimal subset read only during onboarding to enrich the confirmation UI --
 # not a runtime cadence.
 _ONBOARDING_RUNTIME_COMMAND_SPECS: tuple[Pi30CommandSpec, ...] = (
@@ -612,13 +619,15 @@ def _pi30_command_output_keys() -> dict[str, frozenset[str]]:
     """Command (stable cache key) -> the runtime value keys it owns.
 
     Ownership lives here next to the specs/parsers (never in the hub): it maps
-    each optional command to the direct parser keys AND any value derived from
-    them, so a command that reaches a final unsupported verdict can invalidate
-    exactly the keys it produced. Q1/QPIWS key sets are derived from the payload
+    each command to the direct parser keys AND any value derived from them, so
+    an expired sample or final unsupported verdict can invalidate exactly the
+    keys it produced. QPIGS/Q1/QPIWS key sets are derived from the payload
     layouts so they cannot drift from the parsers.
     """
 
     return {
+        "QPIGS": qpigs_output_keys(),
+        "QMOD": frozenset({"operating_mode_code", "operating_mode"}),
         "QPIWS": qpiws_output_keys() | frozenset({"alarm_status"}),
         "Q1": q1_output_keys() | frozenset({"inverter_charge_state"}),
         "QET": frozenset({"pv_generation_sum"}),
@@ -719,6 +728,7 @@ async def _async_collect_values(
     runtime_state: dict[str, Any] | None = None,
     now_monotonic: float | None = None,
     command_timings: list[tuple[str, str, int, str]] | None = None,
+    allow_partial: bool = False,
 ) -> dict[str, Any]:
     values: dict[str, Any] = {}
 
@@ -737,19 +747,61 @@ async def _async_collect_values(
             _record_pi30_command_stat(
                 runtime_state, spec.command, spec.command, duration_ms, _command_outcome(exc), command_timings
             )
-            if not spec.optional:
+            if not spec.optional and not allow_partial:
                 raise
-            _record_command_failure(runtime_state, spec.command)
+            if spec.optional:
+                _record_command_failure(runtime_state, spec.command)
         else:
             duration_ms = int(round((time.monotonic() - started) * 1000.0))
             _record_pi30_command_stat(
                 runtime_state, spec.command, spec.command, duration_ms, "ok", command_timings
             )
             _record_command_success(runtime_state, spec.command)
+            if allow_partial and not spec.optional and runtime_state is not None:
+                runtime_state.setdefault(_REQUIRED_LAST_SUCCESS_KEY, {})[spec.command] = (
+                    time.monotonic() if now_monotonic is None else now_monotonic
+                )
 
     return values
 
 
+def _required_sample_freshness(
+    *,
+    runtime_state: dict[str, Any] | None,
+    command_timings: list[tuple[str, str, int, str]],
+    now: float,
+    poll_interval: float | None,
+) -> tuple[frozenset[str], dict[str, Any]]:
+    """Bound carried core values without disabling their required commands."""
+
+    max_age = max(_REQUIRED_SAMPLE_MAX_AGE_SECONDS, 3.0 * (poll_interval or 0.0))
+    successful = {key for key, _wire, _ms, outcome in command_timings if outcome == "ok"}
+    last_success = (runtime_state or {}).get(_REQUIRED_LAST_SUCCESS_KEY, {})
+    output_keys = _pi30_command_output_keys()
+    removed: set[str] = set()
+    stale: list[str] = []
+    expired: list[str] = []
+    diagnostics: dict[str, Any] = {"pi30_required_value_max_age_seconds": max_age}
+    for spec in _RUNTIME_COMMAND_SPECS:
+        if spec.optional:
+            continue
+        sampled_at = last_success.get(spec.command)
+        age = max(0.0, now - sampled_at) if sampled_at is not None else None
+        if age is not None:
+            diagnostics[f"pi30_{spec.command.lower()}_age_seconds"] = round(age, 3)
+        if spec.command in successful:
+            continue
+        stale.append(spec.command)
+        # No timestamp means this session has never confirmed that command.
+        # Remove detection/previous-session values instead of granting a new TTL.
+        if age is None or age >= max_age:
+            expired.append(spec.command)
+            removed.update(output_keys[spec.command])
+    diagnostics["pi30_required_stale_commands"] = ", ".join(stale)
+    diagnostics["pi30_required_expired_commands"] = ", ".join(expired)
+    return frozenset(removed), diagnostics
+
+
 def _pi30_unreachable_commands(unsupported: frozenset[str]) -> frozenset[str]:
     """Return the commands made unreachable by a finally-unsupported prerequisite.
 
@@ -801,6 +853,9 @@ def _build_pi30_read_meta(
         "pi30_poll_succeeded": sum(1 for _k, _w, _ms, o in attempted if o == "ok"),
         "pi30_poll_timeout": sum(1 for _k, _w, _ms, o in attempted if o == "timeout"),
         "pi30_poll_error": sum(1 for _k, _w, _ms, o in attempted if o == "error"),
+        "pi30_poll_failed_commands": ", ".join(
+            key for key, _wire, _ms, outcome in attempted if outcome in {"timeout", "error"}
+        ),
     }
 
     if runtime_state is None:
@@ -863,19 +918,20 @@ async def _async_collect_runtime_values(
     There is no fast/medium/slow scheduling and no empty cycle: every poll
     executes QPIGS, QMOD, then the optional QPIWS / Q1 (auto-skipped when finally
     unsupported), then the reachable energy chain (existing order + early-exit),
-    back to back with no artificial pause. The result is always a DELTA (see
-    :meth:`Pi30Driver.async_read_values`): a transient failure keeps last-good and
-    a final unsupported verdict invalidates the command's values via
-    ``removed_keys``. ``poll_interval`` / ``now_monotonic`` are accepted for the
-    driver contract but no longer gate anything -- the whole-cycle wall-clock is
-    the honest cost the neutral auto poll policy consumes.
+    back to back with no artificial pause. Per-command payload errors do not
+    discard other successful replies. Required commands are never disabled;
+    their last-good values expire through ``removed_keys``. Optional commands
+    retain the existing unsupported-command policy. Poll interval and monotonic
+    time bound freshness only; they do not gate command scheduling.
     """
 
     command_timings: list[tuple[str, str, int, str]] = []
     cycle_started = time.monotonic()
 
     values = await _async_collect_values(
-        session, _RUNTIME_COMMAND_SPECS, runtime_state=runtime_state, command_timings=command_timings
+        session, _RUNTIME_COMMAND_SPECS, runtime_state=runtime_state,
+        command_timings=command_timings, allow_partial=True,
+        now_monotonic=now_monotonic,
     )
     values.update(
         await _async_collect_energy_values(
@@ -885,13 +941,18 @@ async def _async_collect_runtime_values(
 
     cycle_wall_ms = int(round((time.monotonic() - cycle_started) * 1000.0))
 
-    removed_keys: frozenset[str] = frozenset()
+    removed_keys, freshness = _required_sample_freshness(
+        runtime_state=runtime_state,
+        command_timings=command_timings,
+        now=time.monotonic() if now_monotonic is None else now_monotonic,
+        poll_interval=poll_interval,
+    )
     if runtime_state is not None:
         _commit_cycle_failures(runtime_state)
         # Invalidate values owned by any command that reached a FINAL unsupported
         # verdict (incl. energy-chain reachability). Never remove a value read
         # THIS cycle; unsupported-command reporting is a separate diagnostic.
-        removed_keys = frozenset(
+        removed_keys |= frozenset(
             _pi30_removed_keys_for_unsupported(
                 frozenset(_unsupported_commands(runtime_state))
             )
@@ -904,6 +965,7 @@ async def _async_collect_runtime_values(
         cycle_wall_ms=cycle_wall_ms,
     )
     diagnostics.update(_unsupported_command_diagnostics(runtime_state))
+    diagnostics.update(freshness)
     return values, removed_keys, diagnostics
 
 

custom_components/eybond_local/payload/pi30.py

@@ -532,6 +532,15 @@ def _decode_value(raw: str, kind: str) -> Any:
     (None, "int"),
 )
 
+def qpigs_output_keys() -> frozenset[str]:
+    """Return all QPIGS fields, including locally derived parser values."""
+
+    return frozenset(name for name, _kind in _QPIGS_LAYOUT_24) | frozenset({
+        "qpigs_field_count", "pv_input_power", "battery_power_balance_current",
+        "status_bits_raw",
+    })
+
+
 def qpiws_output_keys() -> frozenset[str]:
     """Return the runtime value keys produced by :func:`parse_qpiws`."""