From c4cbc02a1516b14e5afd5fc1b95a20cb200164bc Mon Sep 17 00:00:00 2001 From: jack Date: Mon, 13 Jul 2026 12:26:33 -0700 Subject: [PATCH] fix(stream_reader): add ASCII hex path + binary resync guard MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- .../Tabs/components/StreamControlPanel.qml | 17 +- .../backend/controllers/session_controller.py | 29 +++- src/pygui/serialcomm/stream_reader.py | 150 +++++++--------- tests/test_session_controller.py | 71 +++++++- tests/test_stream_reader.py | 162 +++++++++++------- 5 files changed, 269 insertions(+), 160 deletions(-) diff --git a/src/pygui/ISC/Tabs/components/StreamControlPanel.qml b/src/pygui/ISC/Tabs/components/StreamControlPanel.qml index 8908cfb..d742c1c 100644 --- a/src/pygui/ISC/Tabs/components/StreamControlPanel.qml +++ b/src/pygui/ISC/Tabs/components/StreamControlPanel.qml @@ -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 } diff --git a/src/pygui/backend/controllers/session_controller.py b/src/pygui/backend/controllers/session_controller.py index 3edf110..412b8ae 100644 --- a/src/pygui/backend/controllers/session_controller.py +++ b/src/pygui/backend/controllers/session_controller.py @@ -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) diff --git a/src/pygui/serialcomm/stream_reader.py b/src/pygui/serialcomm/stream_reader.py index 4444973..87b81a1 100644 --- a/src/pygui/serialcomm/stream_reader.py +++ b/src/pygui/serialcomm/stream_reader.py @@ -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() diff --git a/tests/test_session_controller.py b/tests/test_session_controller.py index c577ff5..a73eb79 100644 --- a/tests/test_session_controller.py +++ b/tests/test_session_controller.py @@ -99,9 +99,10 @@ def test_leaving_live_mode_stops_stream(controller): def test_binary_frame_decode_bounded_to_loaded_layout(controller, monkeypatch): """Regression for phantom sensor #239 polluting MIN/MAX/AVG: the wire payload carries a fixed-width channel array (up to 244 slots) wider than - any single wafer family's real sensor count. Family "A" (aepwafer.yaml) - has 48 real sensors — everything past that in the payload is an - unpopulated hardware slot and must never reach Frame.values.""" + any single wafer family's real sensor count. Family "X" (xwafer.yaml, + the only family live streaming is supported on) has 80 real sensors — + everything past that in the payload is an unpopulated hardware slot and + must never reach Frame.values.""" fake_transport = MagicMock() fake_transport.read.return_value = b"" monkeypatch.setattr( @@ -109,13 +110,13 @@ def test_binary_frame_decode_bounded_to_loaded_layout(controller, monkeypatch): lambda *a, **k: fake_transport, ) - controller.startStream("COM_FAKE", "A") + controller.startStream("COM_FAKE", "X") try: - assert len(controller._sensors) == 48 # sanity: aepwafer.yaml loaded + assert len(controller._sensors) == 80 # sanity: xwafer.yaml loaded parse = controller._reader._parse - payload = bytes(244 * 2) # full-width wire payload, well beyond the real 48 + payload = bytes(244 * 2) # full-width wire payload, well beyond the real 80 frame = parse(payload, seq=1) - assert len(frame.values) == 48 + assert len(frame.values) == 80 finally: controller.stopStream() @@ -135,6 +136,62 @@ def test_live_error_surfaces_message_and_stops_stream(controller): assert controller._reader is None +def test_start_stream_refuses_non_x_family(controller, monkeypatch): + """C# parity: only X-family wafer firmware supports live streaming. + Family A produced ~10s of unrelated ASCII output, not real telemetry — + startStream must refuse before ever opening the port.""" + open_port = MagicMock() + monkeypatch.setattr( + "pygui.serialcomm.serial_port.SerialPort.open_port", open_port + ) + error_seen = MagicMock() + controller.liveError.connect(error_seen) + + controller.startStream("COM_FAKE", "A") + + open_port.assert_not_called() + error_seen.assert_called_once() + assert "X-family" in error_seen.call_args[0][0] + assert controller._reader is None + + +def test_start_stream_allows_x_family(controller, monkeypatch): + fake_transport = MagicMock() + fake_transport.read.return_value = b"" + monkeypatch.setattr( + "pygui.serialcomm.serial_port.SerialPort.open_port", + lambda *a, **k: fake_transport, + ) + error_seen = MagicMock() + controller.liveError.connect(error_seen) + + controller.startStream("COM_FAKE", "X") + try: + error_seen.assert_not_called() + assert controller._reader is not None + finally: + controller.stopStream() + + +def test_stop_stream_skips_d2s_write_when_reader_already_exited(controller): + """Reader's own worker thread can exit and close the transport (e.g. + after a resync give-up) before stopStream() runs on the main thread. + Writing D2S to that already-closed transport is a guaranteed EBADF — + skip it instead of racing the close.""" + controller.setMode("live") + fake_reader = MagicMock() + fake_reader._thread = MagicMock() + fake_reader._thread.is_alive.return_value = False + fake_transport = MagicMock() + fake_transport.is_open = True + fake_reader._transport = fake_transport + controller._reader = fake_reader + + controller.stopStream() + + fake_transport.write.assert_not_called() + + def test_compare_files_integration(controller): # Mock the comparisonResult signal mock_slot = MagicMock() diff --git a/tests/test_stream_reader.py b/tests/test_stream_reader.py index d45ea34..619c1ee 100644 --- a/tests/test_stream_reader.py +++ b/tests/test_stream_reader.py @@ -17,66 +17,8 @@ class FakeTransport: res = self._data[self._pos:self._pos+n] self._pos += n return res - def readline(self): - time.sleep(0.001) - if self._pos >= len(self._data): - return b"" - idx = self._data.find(b"\n", self._pos) - if idx == -1: - res = self._data[self._pos:] - self._pos = len(self._data) - return res - res = self._data[self._pos:idx+1] - self._pos = idx + 1 - return res def close(self): self.closed = True -def parse_line(raw: str, seq: int) -> Frame: - parts = [float(x) for x in raw.split(",")] - return Frame(seq = seq, time = parts[0], values = parts[1:]) - -def test_reads_frames_and_counts_errors(): - got, errors = [], [] - transport = FakeTransport([b"0.0,149,148\n", b"garbage\n", b"0.5,150,149\n"]) - r = StreamReader(transport, parse_line, - on_frame = got.append, on_error = lambda e: errors.append(e)) - - r.start() - time.sleep(0.1) - r.stop() - assert [f.values for f in got] == [[149.0, 148.0], [150.0, 149.0]] - assert r.error_count == 1 - assert transport.closed - - -class WedgedTransport: - """Simulates a worker thread stuck in a blocking read() that ignores - the stop event — read() never returns on its own. stop() must close - the port to unblock it rather than relying on the read loop to notice.""" - def __init__(self): - self.closed = False - - def read(self, n=1): - while not self.closed: - time.sleep(0.001) - raise OSError("port closed") - - def close(self): - self.closed = True - - -def test_stop_closes_transport_even_if_worker_thread_is_wedged(): - r = StreamReader(WedgedTransport(), parse_line, - on_frame=lambda f: None, on_error=lambda e: None) - r.start() - time.sleep(0.05) # let the thread enter the blocking read - r.stop() - assert r._transport.closed, ( - "stop() must close the transport itself — otherwise a wedged " - "worker thread keeps the port open and a subsequent startStream() " - "opens a second competing handle to the same port" - ) - def make_packet(payload: bytes, seq: int = 1) -> bytes: """Build a wire packet exactly as the device does: AA 88, msg type, @@ -98,22 +40,24 @@ def test_crc16_known_vector(): def test_binary_stream_parses_crc_valid_packets(): got = [] data = make_packet(b"\x01\x02", seq=1) + make_packet(b"\x03\x04", seq=2) - r = StreamReader(FakeTransport([data]), parse_payload, + transport = FakeTransport([data]) + r = StreamReader(transport, parse_payload, on_frame=got.append, on_error=lambda e: None) r.start() time.sleep(0.2) r.stop() assert [f.seq for f in got] == [1, 2] assert r.error_count == 0 + assert transport.closed def test_corrupt_crc_packet_dropped_and_stream_continues(): - got = [] + got, errors = [], [] bad = bytearray(make_packet(b"\x09\x09", seq=1)) bad[-1] ^= 0xFF # break the CRC data = bytes(bad) + make_packet(b"\x01\x02", seq=2) r = StreamReader(FakeTransport([data]), parse_payload, - on_frame=got.append, on_error=lambda e: None) + on_frame=got.append, on_error=errors.append) r.start() time.sleep(0.2) r.stop() @@ -146,19 +90,111 @@ def test_quiet_port_between_scans_is_not_a_resync_failure(): r = StreamReader(FakeTransport([make_packet(b"\x01\x02", seq=1)]), parse_payload, on_frame=got.append, on_error=errors.append) r.start() - time.sleep(0.5) # ~100 empty reads after the single packet drains + time.sleep(0.5) # long idle stretch after the single packet drains r.stop() assert [f.seq for f in got] == [1] assert errors == [] +def test_resync_giveup_debug_log_has_bytes_error_log_does_not(caplog): + """Regression: the give-up log used to dump `buf` after it had already + been cleared for the new read, so the hex-diagnostic came back empty — + useless for diagnosing what was actually on the wire. Fix: real bytes + go to DEBUG only; the ERROR line the terminal/Activity Log show stays + free of raw internal buffer content.""" + import logging + errors = [] + # 0xAA/0x88 excluded so this exercises _run_binary's resync path even + # though it's called directly (bypassing the ASCII/binary dispatch, + # which is exercised separately — see test_ascii_hex_dump_stream_*). + garbage = b"\x42" * 20000 + r = StreamReader(FakeTransport([garbage]), parse_payload, + on_frame=lambda f: None, on_error=errors.append) + with caplog.at_level(logging.DEBUG, logger="pygui.serialcomm.stream_reader"): + r._run_binary(bytearray()) + assert len(errors) == 1 + error_lines = [r.message for r in caplog.records if r.levelname == "ERROR"] + debug_lines = [r.message for r in caplog.records if r.levelname == "DEBUG"] + assert any("giving up" in m for m in error_lines) + assert not any("42424242" in m for m in error_lines), ( + "raw wire bytes must not appear in the ERROR line users see" + ) + assert any("42424242" in m for m in debug_lines) + + +def test_resync_gives_up_after_16kb_of_garbage(): + """Give-up cap counts discarded BYTES, not read attempts, so it fires + at a fixed data volume regardless of transfer rate.""" + errors = [] + got = [] + garbage = b"\x01" * 20000 # never contains 0xAA 0x88; well past the cap + r = StreamReader(FakeTransport([garbage]), parse_payload, + on_frame=got.append, on_error=errors.append) + r._run_binary(bytearray()) # deterministic give-up — no thread needed + assert got == [] + assert len(errors) == 1 + assert "resync" in str(errors[0]).lower() + + +def test_ascii_hex_dump_stream_decodes_real_wafer_capture(): + """Regression using an exact byte capture from real hardware (family A): + the wire is plain ASCII hex text, 4 chars/word, no 0xAA 0x88 framing at + all — see docs/adr/0007. _run() must route to the ASCII path and the + words must decode to plausible room-temperature values, not stall.""" + got = [] + words = ["280B", "1C0B", "220B", "1B0B", "200B", "1F0B", "190B", "1D0B", + "1B0B", "240B", "180B", "210B", "2B0B", "260B", "290B", "2B0B"] + # aepwafer.yaml's family "A" has 48 real sensors; pad to one full block + # (256 words) the way the real device does — extra slots are unpopulated. + block = "".join(words) + "0000" * (256 - len(words)) + r = StreamReader(FakeTransport([block.encode()]), parse_payload, + on_frame=got.append, on_error=lambda e: None, + family_code="A") + r.start() + time.sleep(0.2) + r.stop() + assert len(got) == 1 + assert len(got[0].values) == 48 # sensor_count_for("A") + # First word "280B" byte-swaps to "0B28" -> ~22.31C via the AEP formula. + assert 22.0 < got[0].values[0] < 22.5 + + +class WedgedTransport: + """Simulates a worker thread stuck in a blocking read() that ignores + the stop event — read() never returns on its own. stop() must close + the port to unblock it rather than relying on the read loop to notice.""" + def __init__(self): + self.closed = False + + def read(self, n=1): + while not self.closed: + time.sleep(0.001) + raise OSError("port closed") + + def close(self): + self.closed = True + + +def test_stop_closes_transport_even_if_worker_thread_is_wedged(): + r = StreamReader(WedgedTransport(), parse_payload, + on_frame=lambda f: None, on_error=lambda e: None) + r.start() + time.sleep(0.05) # let the thread enter the blocking read + r.stop() + assert r._transport.closed, ( + "stop() must close the transport itself — otherwise a wedged " + "worker thread keeps the port open and a subsequent startStream() " + "opens a second competing handle to the same port" + ) + + def test_stop_does_not_report_the_close_it_caused_as_an_error(): """stop()'s own transport.close() unblocks the worker thread's read() by making it raise — that's expected teardown, not a stream fault, and must not reach on_error (Activity Log would otherwise show a fake "Live stream error" on every ordinary Stop).""" errors = [] - r = StreamReader(WedgedTransport(), parse_line, + r = StreamReader(WedgedTransport(), parse_payload, on_frame=lambda f: None, on_error=errors.append) r.start() time.sleep(0.05)