From 9e28a25695814283360f63b969cb79cae3dbb7c3 Mon Sep 17 00:00:00 2001 From: jack Date: Thu, 2 Jul 2026 22:33:44 -0700 Subject: [PATCH] refactor: implement thread-safe hardware operations, port caching, and refined state management in the device controller --- .../backend/controllers/device_controller.py | 139 ++++++++++++------ .../backend/controllers/session_controller.py | 27 +++- src/pygui/serialcomm/data_parser.py | 2 +- src/pygui/serialcomm/device_service.py | 13 ++ src/pygui/serialcomm/stream_reader.py | 20 +-- tests/test_data_parser.py | 5 + tests/test_device_controller.py | 55 ++++++- 7 files changed, 193 insertions(+), 68 deletions(-) diff --git a/src/pygui/backend/controllers/device_controller.py b/src/pygui/backend/controllers/device_controller.py index 171bb48..8504b9d 100644 --- a/src/pygui/backend/controllers/device_controller.py +++ b/src/pygui/backend/controllers/device_controller.py @@ -39,6 +39,9 @@ class DeviceController(QObject): # ---- public signals ---- portsUpdated = Signal(list) + connectionStatusChanged = Signal() + operationInProgressChanged = Signal() + selectedPortChanged = Signal() detectResult = Signal(object) # WaferInfo dict or None readResult = Signal(dict) # {"success": bool, "bytes": int} or {"error": str} eraseResult = Signal(dict) # {"success": bool} @@ -51,6 +54,8 @@ class DeviceController(QObject): # ---- private signals (marshal worker-thread results back to main thread) ---- _detectFinished = Signal(object, object) # (port_or_None, WaferInfo_or_None) _readFinished = Signal(str, str, object) # (port, family_code, bytes_or_None) + _eraseFinished = Signal(bool) # (success) + _debugFinished = Signal(dict) # (result_dict) def __init__(self, settings: LocalSettings, data_dir: str, parent: QObject | None = None) -> None: super().__init__(parent) @@ -79,6 +84,8 @@ class DeviceController(QObject): # Marshal worker-thread results onto the Qt main thread. self._detectFinished.connect(self._handle_detect_finished, Qt.ConnectionType.QueuedConnection) self._readFinished.connect(self._handle_read_finished, Qt.ConnectionType.QueuedConnection) + self._eraseFinished.connect(self._handle_erase_finished, Qt.ConnectionType.QueuedConnection) + self._debugFinished.connect(self._handle_debug_finished, Qt.ConnectionType.QueuedConnection) root_logger = logging.getLogger() root_logger.addHandler(QmlActivityLogHandler(self)) @@ -94,14 +101,14 @@ class DeviceController(QObject): """Update connection status and notify QML.""" if self._connection_status != status: self._connection_status = status - self.portsUpdated.emit(self.availablePorts) + self.connectionStatusChanged.emit() - @Property(str, notify=portsUpdated) + @Property(str, notify=connectionStatusChanged) def connectionStatus(self) -> str: """Current connection status string for display.""" return self._connection_status - @Property(bool, notify=portsUpdated) + @Property(bool, notify=operationInProgressChanged) def operationInProgress(self) -> bool: """Whether a hardware operation is currently running.""" return self._operation_in_progress @@ -119,7 +126,7 @@ class DeviceController(QObject): self._append_log(f"Save data dir set to: {path}") self._save_status() - @Property(str, notify=portsUpdated) + @Property(str, notify=selectedPortChanged) def selectedPort(self) -> str: """Currently selected serial port shared across the UI.""" return self._selected_port @@ -128,8 +135,10 @@ class DeviceController(QObject): @slot_error_boundary def setSelectedPort(self, port: str) -> None: """Set the active serial port from the port selector.""" - self._selected_port = port - self._save_status() + if self._selected_port != port: + self._selected_port = port + self.selectedPortChanged.emit() + self._save_status() @Property(int, notify=parsedDataReady) def dataRowCount(self) -> int: @@ -243,14 +252,10 @@ class DeviceController(QObject): def _set_operation_progress(self, in_progress: bool) -> None: - """Set operation progress state and notify QML. - - portsUpdated is the shared notify signal for availablePorts, - connectionStatus, and operationInProgress — emitting it forces - QML to re-evaluate all three bindings. - """ - self._operation_in_progress = in_progress - self.portsUpdated.emit(self._service.enumerate_ports()) + """Set operation progress state and notify QML.""" + if self._operation_in_progress != in_progress: + self._operation_in_progress = in_progress + self.operationInProgressChanged.emit() # ---- Public Slots ---- @@ -258,6 +263,7 @@ class DeviceController(QObject): @slot_error_boundary def refreshPorts(self) -> None: """Scan for available serial ports and emit updated list.""" + self._service.clear_port_scan_cache() ports = self._service.enumerate_ports() self._set_connection_status("Disconnected") self.portsUpdated.emit(ports) @@ -396,12 +402,33 @@ class DeviceController(QObject): @slot_error_boundary def eraseMemory(self, port: str) -> None: """Send p1 erase command.""" + if self._operation_in_progress: + self._append_log("Already busy — ignoring erase request") + return self._set_operation_progress(True) self._set_connection_status("Erasing...") self.portsUpdated.emit(self._service.enumerate_ports()) self._append_log(f"Erase command sent to {port} (wafer takes ~15s) ...") - ok = self._service.erase_wafer(port) + threading.Thread( + target=self._erase_worker, + args=(port,), + daemon=True, + ).start() + + def _erase_worker(self, port: str) -> None: + """Background thread worker for eraseMemory.""" + try: + ok = self._service.erase_wafer(port) + except Exception as exc: + log.exception("Erase worker crashed: %s", exc) + ok = False + self._eraseFinished.emit(ok) + + @Slot(bool) + @slot_error_boundary + def _handle_erase_finished(self, ok: bool) -> None: + """Main thread completion handler for eraseMemory.""" if ok: self._set_connection_status("Connected") self._append_log("Erase command accepted — wait ~15 seconds") @@ -414,45 +441,63 @@ class DeviceController(QObject): @Slot(str) @slot_error_boundary def readDebug(self, port: str) -> None: - """Read both D1 (sensor) and F1 (debug/cold-junction) data. - - Emits debugResult with counts on success. - """ + """Read both D1 (sensor) and F1 (debug/cold-junction) data.""" + if self._operation_in_progress: + self._append_log("Already busy — ignoring debug read request") + return self._set_operation_progress(True) self._set_connection_status("Reading debug...") self.portsUpdated.emit(self._service.enumerate_ports()) self._append_log(f"Reading debug data on {port} ...") - # Step 1: D1 sensor data - sensor_data = self._service.read_wafer_data(port, family_code="") - if sensor_data is None: + threading.Thread( + target=self._debug_worker, + args=(port,), + daemon=True, + ).start() + + def _debug_worker(self, port: str) -> None: + """Background thread worker for readDebug.""" + try: + # Step 1: D1 sensor data + sensor_data = self._service.read_wafer_data(port, family_code="") + if sensor_data is None: + self._debugFinished.emit({"error": "D1 read failed"}) + return + + # Step 2: F1 debug data + debug_data = self._service.read_wafer_data(port, family_code="", cmd="F1") + if debug_data is None: + self._debugFinished.emit({"error": "F1 read failed"}) + return + + self._debugFinished.emit({ + "success": True, + "sensor_bytes": len(sensor_data), + "debug_bytes": len(debug_data), + }) + except Exception as exc: + log.exception("Debug worker crashed: %s", exc) + self._debugFinished.emit({"error": str(exc)}) + + @Slot(dict) + @slot_error_boundary + def _handle_debug_finished(self, result: dict) -> None: + """Main thread completion handler for readDebug.""" + if result.get("success"): + self._set_connection_status("Connected") + sensor_bytes = result.get("sensor_bytes", 0) + debug_bytes = result.get("debug_bytes", 0) + self._append_log( + f"Debug read complete: {sensor_bytes} sensor bytes, " + f"{debug_bytes} debug bytes" + ) + else: self._set_connection_status("Disconnected") - self._append_log("D1 read failed — aborting debug read") - self.debugResult.emit({"error": "D1 read failed"}) - self._set_operation_progress(False) - return + err_msg = result.get("error", "Debug read failed") + self._append_log(err_msg) - self._append_log(f"D1 read: {len(sensor_data)} bytes") - - # Step 2: F1 debug data - debug_data = self._service.read_wafer_data(port, family_code="", cmd="F1") - if debug_data is None: - self._set_connection_status("Disconnected") - self._append_log("F1 read failed") - self.debugResult.emit({"error": "F1 read failed"}) - self._set_operation_progress(False) - return - - self._set_connection_status("Connected") - self._append_log( - f"Debug read complete: {len(sensor_data)} sensor bytes, " - f"{len(debug_data)} debug bytes" - ) - self.debugResult.emit({ - "success": True, - "sensor_bytes": len(sensor_data), - "debug_bytes": len(debug_data), - }) + self.debugResult.emit(result) self._set_operation_progress(False) @Slot(str, str) diff --git a/src/pygui/backend/controllers/session_controller.py b/src/pygui/backend/controllers/session_controller.py index c715c05..c40f70f 100644 --- a/src/pygui/backend/controllers/session_controller.py +++ b/src/pygui/backend/controllers/session_controller.py @@ -232,7 +232,8 @@ class SessionController(QObject): def setMode(self, mode: str) -> None: if mode == self._mode: return - self.stopStream() + if mode != "live": + self.stopStream() self._play_timer.stop() self._mode = mode self.modeChanged.emit() @@ -394,6 +395,9 @@ class SessionController(QObject): @Slot(str, str) @slot_error_boundary def startStream(self, port: str, family_code: str = "") -> None: + if self._reader is not None: + log.warning("startStream: StreamReader is already running.") + return from pygui.backend.wafer.wafer_layouts import load_layout_for_wafer_id from pygui.serialcomm.serial_port import SerialPort # transport open @@ -449,13 +453,24 @@ class SessionController(QObject): cmd = b"D2" + SerialPort.COMMAND_PAD.encode() transport.write(cmd) + import weakref + weak_self = weakref.ref(self) + def on_error(exc: Exception): log.error("Live stream error: %s", exc) - self._liveError.emit() + ref = weak_self() + if ref is not None: + ref._liveError.emit() + ref.stopStream() + + def on_frame(frame: Frame): + ref = weak_self() + if ref is not None: + ref._liveFrame.emit(frame) self._reader = StreamReader( transport, parse_binary_frame, - on_frame=lambda f: self._liveFrame.emit(f), + on_frame=on_frame, on_error=on_error, family_code=family_code or "A") self._reader.start() @@ -474,10 +489,9 @@ class SessionController(QObject): self.liveStatsChanged.emit() if self._reader: transport = self._reader._transport - self._reader.stop() - self._reader = None # 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: try: cmd = b"D2S" + (b"F" * 509) @@ -486,6 +500,9 @@ class SessionController(QObject): except Exception as exc: log.error("Error sending stop command: %s", exc) + self._reader.stop() + self._reader = None + self.stateChanged.emit() self.stopRecording() diff --git a/src/pygui/serialcomm/data_parser.py b/src/pygui/serialcomm/data_parser.py index 4ad9bca..a4ef950 100644 --- a/src/pygui/serialcomm/data_parser.py +++ b/src/pygui/serialcomm/data_parser.py @@ -239,7 +239,7 @@ def remove_trailing_zeros(data: list[list[str]]) -> None: fval = float(val) except ValueError: fval = 99.9 - if fval >= 0.01: + if abs(fval) >= 0.01: is_all_zero = False break if is_all_zero: diff --git a/src/pygui/serialcomm/device_service.py b/src/pygui/serialcomm/device_service.py index b12c3fe..641dee1 100644 --- a/src/pygui/serialcomm/device_service.py +++ b/src/pygui/serialcomm/device_service.py @@ -29,6 +29,12 @@ class DeviceService: def __init__(self, settings: LocalSettings) -> None: self._settings = settings + self._cached_ports: list[str] = [] + self._last_port_scan_time = 0.0 + + def clear_port_scan_cache(self) -> None: + """Force the next scan to bypass the cache.""" + self._last_port_scan_time = 0.0 # ------------------------------------------------------------------ # Public API @@ -36,6 +42,11 @@ class DeviceService: def enumerate_ports(self) -> list[str]: """Return list of available serial port names.""" + import time + now = time.monotonic() + if now - self._last_port_scan_time < 2.0: + return self._cached_ports + hardware_ports = [p.device for p in serial.tools.list_ports.comports()] virtual_ports = [] @@ -86,6 +97,8 @@ class DeviceService: if p not in ordered_ports: ordered_ports.append(p) + self._cached_ports = ordered_ports + self._last_port_scan_time = time.monotonic() return ordered_ports diff --git a/src/pygui/serialcomm/stream_reader.py b/src/pygui/serialcomm/stream_reader.py index 64cd069..74fc39d 100644 --- a/src/pygui/serialcomm/stream_reader.py +++ b/src/pygui/serialcomm/stream_reader.py @@ -36,17 +36,13 @@ class StreamReader: def _read_chunk(self, min_bytes: int = 1) -> bytes: """Read a chunk of available bytes from the transport. - Reads up to 256 bytes in one shot instead of byte-at-a-time, - falling back to single-byte reads with a yield when nothing is - immediately available.""" + Reads up to 256 bytes in one shot. If nothing is immediately + available, yields briefly to avoid a busy spin. + """ chunk = self._transport.read(max(min_bytes, 256)) - if chunk: - return chunk - # Nothing available right now – yield briefly to avoid busy-spin. - b = self._transport.read(1) - if not b: + if not chunk: time.sleep(0.005) - return b + return chunk def _accumulate(self, needed: int) -> bytearray: """Accumulate *needed* bytes using buffered reads.""" @@ -92,6 +88,12 @@ 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 finally: try: self._transport.close() diff --git a/tests/test_data_parser.py b/tests/test_data_parser.py index 9e14617..e89a075 100644 --- a/tests/test_data_parser.py +++ b/tests/test_data_parser.py @@ -242,6 +242,11 @@ class TestRemoveTrailingZeros: remove_trailing_zeros(data) assert len(data) == 2 + def test_preserves_negative_temperatures(self): + data = [["25.0", "24.5"], ["-5.5", "-10.0"]] + remove_trailing_zeros(data) + assert len(data) == 2 + # ── save_to_csv ─────────────────────────────────────────────────────────────── diff --git a/tests/test_device_controller.py b/tests/test_device_controller.py index fe16f12..dac63cc 100644 --- a/tests/test_device_controller.py +++ b/tests/test_device_controller.py @@ -87,31 +87,48 @@ def test_clear_session(controller): assert controller.selectedPort == "" assert controller.lastWaferInfo == ["", "", 0, "", 0, 0] -def test_erase_memory(controller): - controller._service.erase_wafer = MagicMock(return_value=True) +def test_erase_memory_spawns_thread(controller): + with patch("threading.Thread") as mock_thread: + controller.eraseMemory("COM1") + mock_thread.assert_called_once() + assert controller.operationInProgress + +def test_erase_memory_handler(controller): emitted_result = None def on_erase_result(result): nonlocal emitted_result emitted_result = result controller.eraseResult.connect(on_erase_result) - controller.eraseMemory("COM1") + controller._handle_erase_finished(True) assert emitted_result == {"success": True} assert controller.connectionStatus == "Connected" + assert not controller.operationInProgress -def test_read_debug(controller): - controller._service.read_wafer_data = MagicMock(return_value=bytes([12] * 512)) +def test_read_debug_spawns_thread(controller): + with patch("threading.Thread") as mock_thread: + controller.readDebug("COM1") + mock_thread.assert_called_once() + assert controller.operationInProgress + +def test_read_debug_handler(controller): emitted_result = None def on_debug_result(result): nonlocal emitted_result emitted_result = result controller.debugResult.connect(on_debug_result) - controller.readDebug("COM1") + controller._handle_debug_finished({ + "success": True, + "sensor_bytes": 512, + "debug_bytes": 512 + }) assert emitted_result is not None assert emitted_result.get("success") is True assert emitted_result.get("sensor_bytes") == 512 assert emitted_result.get("debug_bytes") == 512 + assert controller.connectionStatus == "Connected" + assert not controller.operationInProgress def test_parse_and_save_data_and_get_chart_data(controller): res = controller.getChartData() @@ -187,3 +204,29 @@ def test_detect_wafer_clears_old_session(controller): assert len(controller._activity_log) == 1 assert "Scanning all ports" in controller._activity_log[0] +def test_device_service_port_caching(controller): + service = controller._service + # Mock comports list + with patch("serial.tools.list_ports.comports", return_value=[]), \ + patch("subprocess.run") as mock_run: + # Mock subprocess to return empty + mock_run.return_value.returncode = 0 + mock_run.return_value.stdout = "" + + # Clear cache first + service.clear_port_scan_cache() + + # First scan should invoke list_ports and subprocess + ports1 = service.enumerate_ports() + assert mock_run.call_count == 1 + + # Second scan within 2 seconds should return cached list without invoking subprocess + ports2 = service.enumerate_ports() + assert mock_run.call_count == 1 + assert ports1 == ports2 + + # Clear cache and scan again should invoke subprocess + service.clear_port_scan_cache() + ports3 = service.enumerate_ports() + assert mock_run.call_count == 2 +