From 425f3731cd7836e49a3b5dc3fa5dc300084d891b Mon Sep 17 00:00:00 2001 From: jack Date: Mon, 13 Jul 2026 11:54:11 -0700 Subject: [PATCH] fix(stream): payload-length cap, quiet-port resync, main-thread error teardown --- .../backend/controllers/session_controller.py | 33 +++-- src/pygui/backend/license/license_model.py | 9 ++ src/pygui/serialcomm/stream_reader.py | 95 ++++++++++++-- tests/test_session_controller.py | 39 ++++++ tests/test_stream_reader.py | 120 +++++++++++++++++- 5 files changed, 276 insertions(+), 20 deletions(-) diff --git a/src/pygui/backend/controllers/session_controller.py b/src/pygui/backend/controllers/session_controller.py index ab68cb0..3edf110 100644 --- a/src/pygui/backend/controllers/session_controller.py +++ b/src/pygui/backend/controllers/session_controller.py @@ -21,6 +21,7 @@ from pygui.backend.models.sensor_editor import SensorEditor from pygui.backend.models.session_model import SessionModel, SessionUpdate from pygui.backend.models.threshold_classifier import ThresholdConfig from pygui.backend.utils import slot_error_boundary +from pygui.backend.wafer.family_spec import sensor_count_for from pygui.backend.wafer.wafer_layouts import ec_pairs_for_wafer_id from pygui.backend.wafer.zwafer_models import Sensor from pygui.backend.wafer.zwafer_parser import ZWaferParser @@ -43,6 +44,7 @@ class SessionController(QObject): loadFileError = Signal(str) clusterAveragingEnabledChanged = Signal() liveStatsChanged = Signal() # emitted when receivedCount or errorCount change + liveError = Signal(str) # fatal live-stream error message, for Activity Log # trend: per-frame avg for live graph trendDelta = Signal(str) # JSON [[elapsed_s, avg]] — one new point per live frame trendReset = Signal() # live trend buffer cleared (new stream started) @@ -53,7 +55,7 @@ class SessionController(QObject): segmentExported = Signal(dict) # dict with success, path | error # private: marshal a worker-thread frame onto the main thread _liveFrame = Signal(object) # Frame - _liveError = Signal() # worker-thread error ping + _liveError = Signal(str) # worker-thread error ping, carries exception text def __init__(self, parent: QObject | None = None) -> None: super().__init__(parent) @@ -700,9 +702,16 @@ class SessionController(QObject): if not self._active_clusters: self._active_clusters = group_sensors_by_radius(self._sensors) - # The new binary protocol sends payload of (sensorCount * 2) bytes - # Each sensor is a big-endian 16-bit value (Sign-Magnitude). - expected_sensors = 80 if (family_code or "") == "X" else 244 + # The binary protocol's payload carries a fixed-size channel array + # (sensorCount * 2 bytes, big-endian Sign-Magnitude each) that's wider + # than any one wafer family's real sensor count — trailing channels + # are unpopulated hardware slots, not real readings. Bound decoding to + # the just-loaded layout's actual sensor count so those float/noise + # channels never reach values (and therefore never reach MIN/MAX/AVG + # in compute_stats, which has no sensor-count knowledge of its own). + # Fall back to the family YAML lookup only if no layout loaded yet + # (e.g. startStream called with an unknown family_code). + expected_sensors = len(self._sensors) or sensor_count_for(family_code or "") def parse_binary_frame(payload: bytes, seq: int) -> Frame: values = [] @@ -744,8 +753,11 @@ class SessionController(QObject): log.error("Live stream error: %s", exc) ref = weak_self() if ref is not None: - ref._liveError.emit() - ref.stopStream() + # Marshal onto the main thread via the queued signal — this + # callback runs on StreamReader's own worker thread, and + # stopStream() joins that same thread, so calling it directly + # here would deadlock/raise ("cannot join current thread"). + ref._liveError.emit(str(exc)) def on_frame(frame: Frame): ref = weak_self() @@ -815,10 +827,15 @@ class SessionController(QObject): self.frameUpdated.emit() self.stateChanged.emit() - def _on_live_error(self) -> None: - """Main-thread callback for worker-thread stream errors.""" + def _on_live_error(self, message: str) -> None: + """Main-thread callback for worker-thread stream errors. StreamReader + only calls its on_error for resync exhaustion or an unhandled parse + exception — both unrecoverable for the current stream — so this ends + the session and surfaces the reason (Activity Log wiring in QML).""" self._error_count += 1 self.liveStatsChanged.emit() + self.liveError.emit(message) + self.stopStream() # ---- recording ---- @Slot(str, str) diff --git a/src/pygui/backend/license/license_model.py b/src/pygui/backend/license/license_model.py index e6d6068..094edf5 100644 --- a/src/pygui/backend/license/license_model.py +++ b/src/pygui/backend/license/license_model.py @@ -207,3 +207,12 @@ class LicenseModel(QObject): return True age_days = (time.time() - marker.stat().st_mtime) / 86400.0 return age_days > TRIAL_DAYS + + @Slot(result=bool) + def reviewAccessBlocked(self) -> bool: + """True only once a started trial has actually expired (ADR-0005). + + A "none" state (trial never started) is not a block — nothing to + enforce yet. "permanent" (a real license) never blocks. + """ + return self.replayLicenseState() == "temporary" and self.replayTrialExpired() diff --git a/src/pygui/serialcomm/stream_reader.py b/src/pygui/serialcomm/stream_reader.py index 280e674..4444973 100644 --- a/src/pygui/serialcomm/stream_reader.py +++ b/src/pygui/serialcomm/stream_reader.py @@ -15,8 +15,30 @@ 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 +# Largest believable payload_len. Real frames are ≤ 488 bytes (244 words); +# anything bigger means we latched onto a false 0xAA 0x88 inside data and +# read garbage as a length — without this cap, _accumulate would swallow up +# to 64KB of good frames waiting for a packet that doesn't exist. +# ponytail: loose cap; exact per-family length check if false frames slip CRC. +_MAX_PAYLOAD_LEN = 4096 + + +def _crc16(data: bytes) -> int: + """CRC16/MODBUS (poly 0xA001, init 0xFFFF) — mirrors C# CalculateCRC16.""" + crc = 0xFFFF + for byte in data: + crc ^= byte + for _ in range(8): + if crc & 1: + crc = (crc >> 1) ^ 0xA001 + else: + crc >>= 1 + return crc + class StreamReader: def __init__(self, transport, parse_frame: ParseFrame, @@ -72,6 +94,8 @@ class StreamReader: 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: return @@ -95,11 +119,17 @@ class StreamReader: # Default: ASCII hex dump stream. self._run_ascii(raw) except Exception as exc: - log.exception("StreamReader thread encountered error: %s", exc) - try: - self._on_error(exc) - except Exception: - pass + if self._stop.is_set(): + # stop() closes the transport to unblock a pending read — + # that self-inflicted close raises here too (e.g. EBADF). + # Not a real stream fault, so don't report it as one. + log.debug("StreamReader thread exiting after stop(): %s", exc) + else: + log.exception("StreamReader thread encountered error: %s", exc) + try: + self._on_error(exc) + except Exception: + pass finally: try: self._transport.close() @@ -133,6 +163,12 @@ class StreamReader: return payload_len = int.from_bytes(buf[5:7], byteorder='little') + if payload_len > _MAX_PAYLOAD_LEN: + # False sync marker inside data — the "length" is garbage. + # Skip past the marker and rescan what we already have. + self.resync_count += 1 + buf = buf[2:] + continue total_packet_len = 2 + 5 + payload_len + 2 # Accumulate the rest of the packet in one buffered read. @@ -144,6 +180,22 @@ class StreamReader: if self._stop.is_set(): return + # CRC16 over sync+header+payload, trailing 2 bytes LE — + # mirrors C# VerifyData. A failed CRC is a corrupt or + # misframed packet: drop it, rescan, keep streaming. + received_crc = int.from_bytes( + buf[7 + payload_len:9 + payload_len], byteorder='little' + ) + if _crc16(bytes(buf[:7 + payload_len])) != received_crc: + self.error_count += 1 + if self.error_count <= 10: + log.warning( + "CRC mismatch on binary packet (seq bytes %s) — dropped", + bytes(buf[3:5]).hex(), + ) + buf = buf[2:] + continue + header = buf[2:7] seq = int.from_bytes(header[1:3], byteorder='little') payload = buf[7:7 + payload_len] @@ -160,6 +212,17 @@ class StreamReader: buf = buf[total_packet_len:] else: # Resync: keep only trailing 0xAA if present, then read more. + if len(buf) > 0 and buf[-1] == 0xAA: + buf = bytearray([0xAA]) + else: + buf.clear() + + chunk = self._read_chunk() + if not chunk: + # Quiet port — waiting for data is not a resync failure. + # The device pauses between scans; C#'s findMessageStart + # waits forever. Only actual garbage counts toward the cap. + continue resync_attempts += 1 self.resync_count += 1 if resync_attempts > _MAX_RESYNC_ATTEMPTS: @@ -172,12 +235,6 @@ class StreamReader: TimeoutError("Binary stream resync failed: no sync marker") ) return - if len(buf) > 0 and buf[-1] == 0xAA: - buf = bytearray([0xAA]) - else: - buf.clear() - - chunk = self._read_chunk() buf.extend(chunk) def _run_ascii(self, initial_bytes: bytes) -> None: @@ -276,6 +333,22 @@ class StreamReader: def stop(self) -> None: self._stop.set() + # Close the port out from under any blocked transport.read() call — + # relying on _stop alone means a thread stuck in a longer-than- + # expected read (or one the OS driver is slow to unblock) keeps the + # port open, and a subsequent startStream() then opens a *second* + # handle to the same port, racing the still-live old reader for + # bytes (garbled/duplicated frames on the next Live session). + try: + self._transport.close() + except Exception: + pass if self._thread is not None and self._thread.is_alive(): self._thread.join(timeout = 2.0) + if self._thread.is_alive(): + log.error( + "StreamReader: worker thread still alive %.1fs after " + "stop() — port may not be released for the next session", + 2.0, + ) self._thread = None diff --git a/tests/test_session_controller.py b/tests/test_session_controller.py index 0b69bf3..c577ff5 100644 --- a/tests/test_session_controller.py +++ b/tests/test_session_controller.py @@ -96,6 +96,45 @@ def test_leaving_live_mode_stops_stream(controller): assert not controller._repaint_timer.isActive() +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.""" + fake_transport = MagicMock() + fake_transport.read.return_value = b"" + monkeypatch.setattr( + "pygui.serialcomm.serial_port.SerialPort.open_port", + lambda *a, **k: fake_transport, + ) + + controller.startStream("COM_FAKE", "A") + try: + assert len(controller._sensors) == 48 # sanity: aepwafer.yaml loaded + parse = controller._reader._parse + payload = bytes(244 * 2) # full-width wire payload, well beyond the real 48 + frame = parse(payload, seq=1) + assert len(frame.values) == 48 + finally: + controller.stopStream() + + +def test_live_error_surfaces_message_and_stops_stream(controller): + """A fatal StreamReader error must reach QML with its message (for the + Activity Log) and must tear the stream down cleanly.""" + controller.setMode("live") + controller._reader = MagicMock() + + error_seen = MagicMock() + controller.liveError.connect(error_seen) + + controller._on_live_error("Binary stream resync failed: no sync marker") + + error_seen.assert_called_once_with("Binary stream resync failed: no sync marker") + assert controller._reader is None + + 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 ce8d916..d45ea34 100644 --- a/tests/test_stream_reader.py +++ b/tests/test_stream_reader.py @@ -1,7 +1,7 @@ import time from pygui.backend.models.frame import Frame -from pygui.serialcomm.stream_reader import StreamReader +from pygui.serialcomm.stream_reader import StreamReader, _crc16 class FakeTransport: @@ -47,3 +47,121 @@ def test_reads_frames_and_counts_errors(): 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, + seq (LE), payload len (LE), payload, CRC16/MODBUS (LE) over the rest.""" + body = (b"\xaa\x88\x01" + seq.to_bytes(2, "little") + + len(payload).to_bytes(2, "little") + payload) + return body + _crc16(body).to_bytes(2, "little") + + +def parse_payload(payload: bytes, seq: int) -> Frame: + return Frame(seq=seq, time=0.0, values=[float(b) for b in payload]) + + +def test_crc16_known_vector(): + # CRC16/MODBUS check value for "123456789" is 0x4B37. + assert _crc16(b"123456789") == 0x4B37 + + +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, + 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 + + +def test_corrupt_crc_packet_dropped_and_stream_continues(): + got = [] + 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) + r.start() + time.sleep(0.2) + r.stop() + assert [f.seq for f in got] == [2] + assert r.error_count == 1 + + +def test_bogus_giant_payload_len_does_not_stall_the_stream(): + """A false 0xAA 0x88 inside data yields a garbage payload_len (up to + 64KB). Without a sanity cap the reader waits forever accumulating a + packet that doesn't exist — the 'frames stop, no error' symptom.""" + got = [] + # False sync + header claiming a 0xFFFF-byte payload, then a real packet. + bogus = b"\xaa\x88\x01\x00\x00\xff\xff" + data = bogus + make_packet(b"\x01\x02", seq=7) + r = StreamReader(FakeTransport([data]), 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] == [7] + + +def test_quiet_port_between_scans_is_not_a_resync_failure(): + """Device pauses between measurement scans. Empty reads must not count + toward the resync give-up cap (C#'s findMessageStart waits forever) — + the old code died with TimeoutError after ~50 empty reads.""" + errors = [] + got = [] + 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 + r.stop() + assert [f.seq for f in got] == [1] + assert errors == [] + + +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, + on_frame=lambda f: None, on_error=errors.append) + r.start() + time.sleep(0.05) + r.stop() + time.sleep(0.05) # let the worker thread's except-handler run + assert errors == []