fix(stream_reader): add ASCII hex path + binary resync guard

Family-A wafers never send 0xAA 0x88 framing — _run() now routes to _run_ascii() on first byte dispatch instead of treating the whole stream as corrupt binary. Binary path gets a 16 KB give-up cap (bytes, not attempts) so a permanently-corrupt port can't stall the reader indefinitely. Raw wire bytes go to DEBUG only; the ERROR line that surfaces in the Activity Log is kept free of internal buffer content. - add family_code param to StreamReader for ASCII decode routing - add _run_binary give-up cap + DEBUG-only hex diagnostic - fix stop(): close transport on wedged thread (binary + ASCII paths) - refactor test_stream_reader: consolidate ASCII regression + resync give-up + DEBUG-vs-ERROR log split tests; drop redundant parse_line - extend test_session_controller: live-stream gate for non-X family - StreamControlPanel.qml: thread-safe stop on tab leave
2026-07-13 12:26:33 -07:00
parent 425f3731cd
commit c4cbc02a15
5 changed files with 269 additions and 160 deletions
@@ -110,8 +110,13 @@ Rectangle {

                TabButton {
                    id: liveTab
+                    // Only X-family wafer firmware supports live streaming
+                    // (C# parity, Form1.cs btnConnect_Click) — other families
+                    // have no live-stream protocol to speak.
+                    readonly property string detectedFamily: (deviceController.lastWaferInfo && deviceController.lastWaferInfo.length > 0) ? deviceController.lastWaferInfo[0] : ""
+                    readonly property bool familySupportsStream: detectedFamily === "X"
                    text: "Live"
-                    enabled: deviceController.connectionStatus === "Connected"
+                    enabled: deviceController.connectionStatus === "Connected" && familySupportsStream
                    opacity: enabled ? 1.0 : 0.4
                    Layout.fillWidth: true
                    implicitHeight: 24
@@ -120,7 +125,9 @@ Rectangle {

                    AppToolTip {
                        visible: disabledHover.containsMouse && !liveTab.enabled
-                        text: "Connect a wafer to enable Live mode"
+                        text: deviceController.connectionStatus !== "Connected"
+                            ? "Connect a wafer to enable Live mode"
+                            : "Live mode is only supported on X-family wafers (detected " + liveTab.detectedFamily + ")"
                        delay: 300
                    }

@@ -359,12 +366,14 @@ Rectangle {
            text: streamController.mode === "live" ? "STOP" : "START"
            enabled: streamController.mode === "live"
                     ? true
-                     : deviceController.connectionStatus === "Connected"
+                     : (deviceController.connectionStatus === "Connected" && liveTab.familySupportsStream)
            opacity: enabled ? 1.0 : 0.4

            AppToolTip {
                visible: primaryBtn.hovered && !primaryBtn.enabled
-                text: "Connect a wafer to enable Live mode"
+                text: deviceController.connectionStatus !== "Connected"
+                    ? "Connect a wafer to enable Live mode"
+                    : "Live mode is only supported on X-family wafers (detected " + liveTab.detectedFamily + ")"
                delay: 300
            }

@@ -687,6 +687,16 @@ class SessionController(QObject):
            log.warning("startStream: StreamReader is already running.")
            return

+        # C# parity (Form1.cs btnConnect_Click: "currently only X wafers
+        # support streaming"). Other families' firmware has no live-stream
+        # protocol — a D2 send on family A produced a burst of unrelated
+        # ASCII output for ~10s then went quiet, not real sensor telemetry.
+        if family_code and family_code != "X":
+            message = f"Live stream is only supported on X-family wafers (detected {family_code})"
+            log.warning(message)
+            self.liveError.emit(message)
+            return
+
        from pygui.backend.wafer.wafer_layouts import load_layout_for_wafer_id
        from pygui.serialcomm.serial_port import SerialPort  # transport open

@@ -742,6 +752,13 @@ class SessionController(QObject):
        self.liveStatsChanged.emit()

        transport = SerialPort.open_port(port, timeout=1)
+        # Discard anything left over from a prior session (e.g. the device
+        # kept transmitting after our last D2S landed) — stale bytes here
+        # misalign the very first packet, corrupting every reading in it.
+        try:
+            transport.reset_input_buffer()
+        except Exception as exc:
+            log.warning("Could not reset input buffer before start: %s", exc)
        # Send 'D2' command padded to 512 bytes to start the stream
        cmd = b"D2" + SerialPort.COMMAND_PAD.encode()
        transport.write(cmd)
@@ -784,14 +801,24 @@ class SessionController(QObject):
        self.liveStatsChanged.emit()
        if self._reader:
            transport = self._reader._transport
+            reader_thread = self._reader._thread
+            # If the worker thread already exited on its own (e.g. a fatal
+            # stream error), it already closed the transport in its finally
+            # block — writing here races that close and raises EBADF. Nothing
+            # is listening for D2S anyway once the reader has given up.
+            reader_still_running = reader_thread is not None and reader_thread.is_alive()

            # Send 'D2S' command padded to 512 bytes to stop the stream
            # Send BEFORE stopping the reader which will close the port on thread exit
-            if transport and transport.is_open:
+            if reader_still_running and transport and transport.is_open:
                try:
                    cmd = b"D2S" + (b"F" * 509)
                    transport.write(cmd)
                    transport.flush()
+                    # Device may keep transmitting a few more bytes after D2S
+                    # lands — discard them so they don't survive into the
+                    # next Live session and misalign its first packet.
+                    transport.reset_input_buffer()
                except Exception as exc:
                    log.error("Error sending stop command: %s", exc)

@@ -14,10 +14,13 @@ log = logging.getLogger(__name__)

 ParseFrame = Callable[[bytes, int], Frame]

-# Max consecutive resync attempts before giving up on a corrupt stream.
-# Counted only when data WAS read and still no sync marker (pure garbage);
-# an idle port never counts — C#'s findMessageStart waits forever on quiet.
-_MAX_RESYNC_ATTEMPTS = 50
+# Max garbage bytes discarded (no sync marker found) before giving up on a
+# corrupt stream. Counted in BYTES, not read attempts — attempts scale with
+# data rate (a slow trickle could take minutes to trip an attempt-count cap),
+# bytes don't. Real frames are <=500B, so a healthy stream never comes close;
+# 16KB is ~32 frames of slack, ~1.4s of pure garbage at 115200 baud. Empty
+# reads (idle port) never count — C#'s findMessageStart waits forever on quiet.
+_MAX_RESYNC_BYTES = 16384

 # Largest believable payload_len. Real frames are ≤ 488 bytes (244 words);
 # anything bigger means we latched onto a false 0xAA 0x88 inside data and
@@ -85,39 +88,31 @@ class StreamReader:

    def _run(self) -> None:
        try:
-            # Peek at the first few bytes to determine protocol.
-            peek_buf = bytearray()
-            while not self._stop.is_set() and len(peek_buf) < 32:
-                b = self._transport.read(1)
-                if not b:
-                    continue
-                peek_buf.extend(b)
-                if b[0] == 10:  # newline
-                    break
-                if len(peek_buf) == 2 and peek_buf[0] == 0xAA and peek_buf[1] == 0x88:
-                    break  # binary sync marker — decisive, stop peeking
-
-            if self._stop.is_set() or not peek_buf:
+            # Protocol is decided by the first byte: 0xAA is the binary sync
+            # marker's lead byte and can never appear in the ASCII hex-dump
+            # stream (0-9/A-F only) real hardware has been observed sending
+            # — see docs/adr/0007. Anything else is ASCII.
+            first = b""
+            while not first and not self._stop.is_set():
+                first = self._transport.read(1)
+            if self._stop.is_set() or not first:
                return

-            raw = bytes(peek_buf)
-
-            # Binary sync bytes are the strongest signal.
-            if raw.startswith(b"\xaa\x88"):
-                self._run_binary(raw)
+            if first[0] == 0xAA:
+                second = b""
+                while not second and not self._stop.is_set():
+                    second = self._transport.read(1)
+                if self._stop.is_set():
+                    return
+                if second and second[0] == 0x88:
+                    self._run_binary(bytearray(first + second))
+                    return
+                # Lone 0xAA, not a real marker — feed both bytes into the
+                # ASCII path instead of discarding them.
+                self._run_ascii(bytes(first) + bytes(second))
                return

-            # Text-lines (CSV) requires BOTH a newline AND commas in the
-            # pre-newline content.  This avoids misclassifying a binary
-            # payload that happens to contain 0x2C as text.
-            has_newline = b"\n" in raw or b"\r" in raw
-            line_before_nl = raw.split(b"\n")[0].split(b"\r")[0]
-            if has_newline and b"," in line_before_nl:
-                self._run_text_lines(raw)
-                return
-
-            # Default: ASCII hex dump stream.
-            self._run_ascii(raw)
+            self._run_ascii(bytes(first))
        except Exception as exc:
            if self._stop.is_set():
                # stop() closes the transport to unblock a pending read —
@@ -139,7 +134,8 @@ class StreamReader:
    def _run_binary(self, initial_bytes: bytes | bytearray) -> None:
        log.info("StreamReader: starting binary stream parsing")
        buf = bytearray(initial_bytes)
-        resync_attempts = 0
+        resync_bytes = 0
+        logged_head = False

        while not self._stop.is_set():
            # Find sync marker in buffer.
@@ -152,7 +148,11 @@ class StreamReader:
                idx += 1

            if synced:
-                resync_attempts = 0
+                if not logged_head:
+                    log.debug("StreamReader: first bytes on wire: %s",
+                            bytes(buf[:64]).hex())
+                    logged_head = True
+                resync_bytes = 0
                buf = buf[idx:]
                # Accumulate header (5 bytes after sync).
                while len(buf) < 7 and not self._stop.is_set():
@@ -223,13 +223,24 @@ class StreamReader:
                    # The device pauses between scans; C#'s findMessageStart
                    # waits forever. Only actual garbage counts toward the cap.
                    continue
-                resync_attempts += 1
+                if not logged_head:
+                    log.debug("StreamReader: first bytes on wire: %s",
+                            bytes((buf + chunk)[:64]).hex())
+                    logged_head = True
                self.resync_count += 1
-                if resync_attempts > _MAX_RESYNC_ATTEMPTS:
+                resync_bytes += len(chunk)
+                if resync_bytes > _MAX_RESYNC_BYTES:
+                    # buf was already cleared/reset at the top of this
+                    # iteration — the bytes worth dumping are in buf+chunk
+                    # (the carryover byte, if any, plus what was just read).
+                    # Raw wire content is DEBUG-only — never in the ERROR
+                    # line the terminal/Activity Log surface to the user.
+                    log.debug("StreamReader: discarded bytes before giving up: %s",
+                            bytes((buf + chunk)[:64]).hex())
                    log.error(
-                        "StreamReader: giving up after %d resync attempts "
+                        "StreamReader: giving up after %d garbage bytes "
                        "(no 0xAA 0x88 sync marker found in stream)",
-                        resync_attempts,
+                        resync_bytes,
                    )
                    self._on_error(
                        TimeoutError("Binary stream resync failed: no sync marker")
@@ -238,6 +249,9 @@ class StreamReader:
                buf.extend(chunk)

    def _run_ascii(self, initial_bytes: bytes) -> None:
+        """Parse the ASCII hex-dump live stream: fixed-width blocks of
+        4-char hex words, one word per channel slot (up to 256 slots; only
+        the family's real sensor_count are populated, see docs/adr/0007)."""
        log.info("StreamReader: starting ASCII hex dump stream parsing")

        valid_words = sensor_count_for(self._family_code)
@@ -246,12 +260,9 @@ class StreamReader:
        chars_needed = block_words * word_char_len  # total hex chars per block

        hex_chars = bytearray()
-        if initial_bytes:
-            for b in initial_bytes:
-                if chr(b) in "0123456789ABCDEFabcdef":
-                    hex_chars.append(b)
-                elif b == 0xAA:
-                    hex_chars.append(b)
+        for b in initial_bytes:
+            if chr(b) in "0123456789ABCDEFabcdef":
+                hex_chars.append(b)

        seq = 0
        while not self._stop.is_set():
@@ -259,24 +270,15 @@ class StreamReader:
            while len(hex_chars) < chars_needed and not self._stop.is_set():
                chunk = self._read_chunk(min_bytes=chars_needed - len(hex_chars))
                for byte in chunk:
-                    if byte == 0xAA:
-                        # Sync marker appeared – hand off to binary parser.
-                        hex_chars.append(byte)
-                        break
                    if chr(byte) in "0123456789ABCDEFabcdef":
                        hex_chars.append(byte)
-                else:
-                    continue
-                break  # broke out of inner loop due to 0xAA

-            # Check for early termination (0xAA sync marker anywhere in buffer).
-            aa_idx = hex_chars.find(bytes([0xAA]))
-            if aa_idx >= 0:
-                log.info("StreamReader: Detected 0xAA at position %d, switching to binary mode.", aa_idx)
-                self._run_binary(hex_chars[aa_idx:])
+            if self._stop.is_set():
                return

-            # Parse words from the accumulated hex characters.
+            # Parse one block's worth of words (buffered reads may overshoot
+            # chars_needed — cap here so the remainder rolls into the next
+            # iteration instead of being folded into this frame and lost).
            word_buffer = []
            for i in range(0, min(len(hex_chars), chars_needed), word_char_len):
                word_str = hex_chars[i:i + word_char_len]
@@ -285,7 +287,6 @@ class StreamReader:
                word_buffer.append(bytes(word_str))

            if not word_buffer:
-                time.sleep(0.01)
                continue

            hex_chars = hex_chars[len(word_buffer) * word_char_len:]
@@ -295,7 +296,7 @@ class StreamReader:
            for hex_val in valid_hex_words:
                try:
                    swapped = hex_val[2:4] + hex_val[0:2]
-                    t = _convert_hex_to_temp(swapped, self._family_code)  # type: ignore[arg-type]
+                    t = _convert_hex_to_temp(swapped.decode(), self._family_code)
                    values.append(t)
                except Exception:
                    values.append(0.0)
@@ -305,31 +306,10 @@ class StreamReader:
                self._on_frame(frame)
                seq += 1
            except Exception as exc:
-                log.error("Error emitting ASCII frame: %s", exc)
-
-    def _run_text_lines(self, initial_bytes: bytes) -> None:
-        log.info("StreamReader: starting line-by-line ASCII text parsing")
-        buf = bytearray(initial_bytes)
-        seq = 0
-
-        while not self._stop.is_set():
-            # Drain complete lines from the buffer.
-            while b"\n" in buf:
-                line_bytes, _, buf = buf.partition(b"\n")
-                line = line_bytes.decode('utf-8', errors='ignore').strip()
-                if line:
-                    try:
-                        frame = self._parse(line, seq)  # type: ignore[arg-type]
-                        if frame:
-                            self._on_frame(frame)
-                            seq += 1
-                    except Exception as exc:
-                        self.error_count += 1
-                        self._on_error(exc)
-
-            # Read more data using buffered reads.
-            chunk = self._read_chunk()
-            buf.extend(chunk)
+                self.error_count += 1
+                if self.error_count <= 10:
+                    log.warning("Error emitting ASCII frame: %s", exc)
+                self._on_error(exc)

    def stop(self) -> None:
        self._stop.set()