fix: discriminate callback AABB MPPT frames from FC4

Framed and AT collectors harvest bare aabb side-channel telemetry without
flipping forward mode or stealing correlated FC4 replies.

Author: mikehardy

CHANGELOG.md

@@ -9,6 +9,8 @@ the GitHub release body should be rendered from the matching version section her
 
 ### Fixed
 
+- Collector framed/AT transports now discriminate bare MPPT `aabb` frames from EyeBond FC4 on the callback socket, complete AABB waiters without corrupting negotiated forward mode, and project last-PV snapshots into `collector_info` for drivers that solicit `5AA5…02`.
+
 - Callback identity probes no longer start a process-wide shared listener they cannot tear down. Attempts borrow an already-owned discovery listener or report `callback_listener_unavailable`, and expected onboarding outcomes (timeouts, silent callbacks, wire retries) log at debug/info instead of warning.
 
 - Read-only cloud analysis now retains its results for collector-only entries

custom_components/eybond_local/collector/transport/__init__.py

@@ -23,6 +23,7 @@
     _acquire_shared_listener,
     _acquire_shared_payload_listener,
     _release_shared_listener,
+    _resolve_registered_listener,
 )
 from .proxy import SharedProxyCaptureRoute
 from .shared_at import SharedCollectorAtTransport

custom_components/eybond_local/collector/transport/common.py

@@ -24,6 +24,7 @@
 from ...link_models import EybondLinkRoute, LinkRoute, RawSerialLinkRoute
 from ...link_transport import PayloadLinkTransport
 from ...models import CollectorInfo
+from ...payload.short_ascii import AABB_WIRE
 from ..profile import apply_collector_profile
 from ..protocol import (
     EybondHeader,
@@ -207,6 +208,142 @@ def _looks_like_plain_raw_response_start(chunk: bytes) -> bool:
     return 0x20 <= value <= 0x7E
 
 
+def _record_aabb_telemetry(collector: CollectorInfo, frame: bytes) -> bool:
+    """Harvest one fixed-length ``aabb`` PV telemetry frame onto collector info.
+
+    Returns True when ``frame`` was recognized as ``aabb`` (even non-PV subtypes)
+    so the caller can consume it from the TCP stream.
+    """
+
+    if len(frame) < AABB_WIRE.frame_length or not frame.startswith(AABB_WIRE.magic):
+        return False
+    frame = frame[: AABB_WIRE.frame_length]
+    subtype_bytes = frame[len(AABB_WIRE.magic) : AABB_WIRE.data_offset]
+    subtype = subtype_bytes.hex()
+    frame_hex = frame.hex()
+    seen = monotonic()
+    collector.aabb_last_any = (subtype, frame_hex, seen)
+    # PV/controller block observed as subtype ``0200`` in the Anern proxy capture.
+    # Subtype ``0202`` is the MPPT *settings* reply (SI20210402T2), not live PV.
+    if subtype_bytes == AABB_WIRE.pv_subtype:
+        collector.aabb_last_pv = (frame_hex, seen)
+        # Compatibility mirrors describe the PV snapshot only. Keeping a
+        # non-PV subtype out of these fields prevents projection from combining
+        # a settings subtype with stale PV bytes and timestamp.
+        collector.aabb_last_hex = frame_hex
+        collector.aabb_last_monotonic = seen
+        collector.aabb_last_subtype = subtype
+        logger.debug(
+            "Harvested aabb PV telemetry remote=%s subtype=%s frame=%s",
+            collector.remote_ip,
+            subtype,
+            frame.hex(),
+        )
+    else:
+        logger.debug(
+            "Ignored non-PV aabb frame remote=%s subtype=%s frame=%s",
+            collector.remote_ip,
+            subtype,
+            frame.hex(),
+        )
+    return True
+
+
+def _complete_pending_aabb_raw_response(
+    pending: asyncio.Future[bytes] | None,
+    *,
+    protocol: str,
+    frame: bytes,
+) -> bool:
+    """Complete a raw ``payload_protocol=aabb`` waiter with one harvested frame."""
+
+    if pending is None or pending.done():
+        return False
+    if str(protocol or "").strip().lower() != "aabb":
+        return False
+    if (
+        len(frame) < AABB_WIRE.frame_length
+        or not frame.startswith(AABB_WIRE.magic)
+        or frame[len(AABB_WIRE.magic) : AABB_WIRE.data_offset]
+        != AABB_WIRE.pv_subtype
+    ):
+        return False
+    pending.set_result(frame[: AABB_WIRE.frame_length])
+    return True
+
+
+async def _discriminate_aabb_magic(
+    reader: "_PrefixedAsyncReader",
+    magic: bytes,
+    *,
+    timeout: float,
+    is_correlated_header: Callable[[EybondHeader], bool] | None = None,
+) -> tuple[str, bytes]:
+    """Decide whether ``AA BB…`` is unsolicited aabb PV or an FC4 header (tid ``0xAABB``).
+
+    Caller has already consumed the two-byte ``\\xaa\\xbb`` magic into ``magic``.
+
+    Returns one of:
+    - ``(\"aabb\", frame21)`` — harvest and continue
+    - ``(\"eybond_header\", header8)`` — process as a normal EyeBond frame header
+    - ``(\"incomplete\", b\"\")`` — ambiguous bytes were ``push_front``'d; wait for more
+      underlying data (or discard on stall) before retrying
+    """
+
+    if magic != AABB_WIRE.magic:
+        raise ValueError("aabb_magic_required")
+
+    buf = bytearray(magic)
+    # Always inspect the full EyeBond header before committing to AABB. A valid
+    # FC4 response can have tid=0xAABB and devcode=0x0200, making its first four
+    # bytes identical to the AABB magic + PV subtype. Explicit correlation with
+    # an outstanding TID/function is stronger evidence than that subtype.
+    while len(buf) < HEADER_SIZE:
+        try:
+            buf += await asyncio.wait_for(reader.readexactly(1), timeout=timeout)
+        except asyncio.TimeoutError:
+            reader.push_front(bytes(buf))
+            return ("incomplete", b"")
+
+    header_candidate = bytes(buf[:HEADER_SIZE])
+    plausible_header = _looks_like_eybond_frame_start(header_candidate)
+    if plausible_header and is_correlated_header is not None:
+        header = decode_header(header_candidate)
+        if is_correlated_header(header):
+            return ("eybond_header", header_candidate)
+
+    subtype = bytes(buf[len(AABB_WIRE.magic) : AABB_WIRE.data_offset])
+    known_aabb_subtype = subtype in AABB_WIRE.known_subtypes
+    target_length = AABB_WIRE.frame_length if known_aabb_subtype else HEADER_SIZE
+    while len(buf) < target_length:
+        try:
+            buf += await asyncio.wait_for(reader.readexactly(1), timeout=timeout)
+        except asyncio.TimeoutError:
+            reader.push_front(bytes(buf))
+            return ("incomplete", b"")
+
+    if known_aabb_subtype:
+        return ("aabb", bytes(buf[: AABB_WIRE.frame_length]))
+
+    if plausible_header:
+        leftover = bytes(buf[HEADER_SIZE:])
+        if leftover:
+            reader.push_front(leftover)
+        return ("eybond_header", header_candidate)
+
+    while len(buf) < AABB_WIRE.frame_length:
+        try:
+            buf += await asyncio.wait_for(reader.readexactly(1), timeout=timeout)
+        except asyncio.TimeoutError:
+            reader.push_front(bytes(buf))
+            return ("incomplete", b"")
+
+    frame = bytes(buf[: AABB_WIRE.frame_length])
+    leftover = bytes(buf[AABB_WIRE.frame_length :])
+    if leftover:
+        reader.push_front(leftover)
+    return ("aabb", frame)
+
 def _short_ascii(value: bytes, *, limit: int = 160) -> str:
     text = "".join(chr(byte) if 0x20 <= byte <= 0x7E else "." for byte in value[:limit])
     if len(value) > limit:
@@ -435,6 +572,11 @@ def _copy_collector_info(collector: CollectorInfo) -> CollectorInfo:
             collector_virtual_bridge=collector.collector_virtual_bridge,
             collector_bridge_kind=collector.collector_bridge_kind,
             collector_bridge_version=collector.collector_bridge_version,
+            aabb_last_any=collector.aabb_last_any,
+            aabb_last_hex=collector.aabb_last_hex,
+            aabb_last_monotonic=collector.aabb_last_monotonic,
+            aabb_last_subtype=collector.aabb_last_subtype,
+            aabb_last_pv=collector.aabb_last_pv,
         )
     )
 
@@ -444,6 +586,33 @@ def __init__(self, reader: asyncio.StreamReader, initial_bytes: bytes = b"") ->
         self._reader = reader
         self._buffer = bytearray(initial_bytes)
 
+    def push_front(self, data: bytes) -> None:
+        """Return bytes to the front of the read buffer (unread)."""
+
+        if data:
+            self._buffer[:0] = data
+
+    def discard_buffered(self, count: int) -> int:
+        """Drop up to ``count`` bytes from the front of the buffer. Return dropped."""
+
+        n = max(0, min(int(count), len(self._buffer)))
+        if n:
+            del self._buffer[:n]
+        return n
+
+    def buffered_startswith(self, prefix: bytes) -> bool:
+        return bytes(self._buffer[: len(prefix)]) == prefix
+
+    async def await_underlying_byte(self, timeout: float) -> bool:
+        """Wait for one new network byte into the buffer. False on timeout."""
+
+        try:
+            byte = await asyncio.wait_for(self._reader.readexactly(1), timeout=timeout)
+        except asyncio.TimeoutError:
+            return False
+        self._buffer.extend(byte)
+        return True
+
     async def readexactly(self, size: int) -> bytes:
         if size <= 0:
             return b""
@@ -458,7 +627,6 @@ async def readexactly(self, size: int) -> bytes:
             return data
         data += await self._reader.readexactly(size - len(data))
         return data
-
     async def readuntil(self, separator: bytes = b"\n") -> bytes:
         index = self._buffer.find(separator)
         if index >= 0: