jack/pyGUI
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: implement thread-safe hardware operations, port caching, and refined state management in the device controller

Browse Source
This commit is contained in:
jack
2026-07-02 22:33:44 -07:00
parent da43aeb1f6
commit 9e28a25695
7 changed files with 193 additions and 68 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/pygui/backend/controllers/device_controller.py
+92 -47
View File
@@ -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)
src/pygui/backend/controllers/session_controller.py
+22 -5
View File
@@ -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()
src/pygui/serialcomm/data_parser.py
+1 -1
View File
@@ -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:
src/pygui/serialcomm/device_service.py
+13
View File
@@ -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
src/pygui/serialcomm/stream_reader.py
+11 -9
View File
@@ -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()
tests/test_data_parser.py
+5
View File
@@ -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 ───────────────────────────────────────────────────────────────
tests/test_device_controller.py
+49 -6
View File
@@ -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