jack
203 lines
7.5 KiB
Python
203 lines
7.5 KiB
Python
"""High-level wafer device communication service.
|
||
|
||
Coordinates port enumeration, detect, read, and erase operations
|
||
using settings from LocalSettingsModel. Designed to be called from
|
||
QML via @Slot methods on a QObject controller.
|
||
"""
|
||
|
||
import logging
|
||
from typing import Optional
|
||
|
||
import serial.tools.list_ports
|
||
|
||
from pygui.backend.data.local_settings import LocalSettings
|
||
from pygui.serialcomm.serial_port import SerialPort, WaferInfo
|
||
|
||
log = logging.getLogger(__name__)
|
||
|
||
|
||
class DeviceService:
|
||
"""High-level wafer device communication."""
|
||
|
||
# Expected hex string lengths for known wafer families
|
||
# P wafer: 393216 hex chars (196608 bytes)
|
||
# X wafer: 1310720 hex chars (655360 bytes)
|
||
EXPECTED_HEX_LENGTHS = {
|
||
"P": 393216,
|
||
"X": 1310720,
|
||
}
|
||
|
||
def __init__(self, settings: LocalSettings) -> None:
|
||
self._settings = settings
|
||
|
||
# ------------------------------------------------------------------
|
||
# Public API
|
||
# ------------------------------------------------------------------
|
||
|
||
def enumerate_ports(self) -> list[str]:
|
||
"""Return list of available serial port names."""
|
||
hardware_ports = [p.device for p in serial.tools.list_ports.comports()]
|
||
virtual_ports = []
|
||
|
||
# On macOS/Linux, also detect virtual loopback serial ports (PTYs) opened by socat
|
||
import platform
|
||
import re
|
||
import subprocess
|
||
|
||
if platform.system() in ("Darwin", "Linux"):
|
||
try:
|
||
# Find PTYs opened by socat
|
||
result = subprocess.run(
|
||
["lsof", "-c", "socat"],
|
||
capture_output=True,
|
||
text=True,
|
||
check=False
|
||
)
|
||
if result.returncode == 0:
|
||
for line in result.stdout.splitlines():
|
||
parts = line.strip().split()
|
||
if len(parts) >= 9:
|
||
fd = parts[3]
|
||
# Exclude standard FDs (0, 1, 2) which are typically the controlling terminal
|
||
if fd.startswith(('0', '1', '2')):
|
||
continue
|
||
name = parts[-1]
|
||
match = re.search(r'(/dev/(?:ttys\d+|pts/\d+))\b', name)
|
||
if match:
|
||
port = match.group(1)
|
||
if port not in virtual_ports:
|
||
virtual_ports.append(port)
|
||
except Exception as e:
|
||
log.debug("Error listing virtual ports via lsof: %s", e)
|
||
|
||
# Prioritize physical USB serial interfaces first -> generic hardware -> virtual ports
|
||
# Virtual ports (PTYs detected via lsof) should NOT appear in hardware_ports,
|
||
# so we filter them out explicitly.
|
||
virtual_set = set(virtual_ports)
|
||
ordered_ports = []
|
||
for p in hardware_ports:
|
||
if "usbserial" in p.lower() or "usb" in p.lower() or "ttyusb" in p.lower():
|
||
if p not in ordered_ports:
|
||
ordered_ports.append(p)
|
||
for p in hardware_ports:
|
||
if p not in ordered_ports and p not in virtual_set:
|
||
ordered_ports.append(p)
|
||
for p in virtual_ports:
|
||
if p not in ordered_ports:
|
||
ordered_ports.append(p)
|
||
|
||
return ordered_ports
|
||
|
||
|
||
def detect_wafer(self, port: str) -> Optional[WaferInfo]:
|
||
"""Try to detect a wafer on the given port.
|
||
|
||
Opens the port, sends s0 command, parses response.
|
||
Returns WaferInfo if detected, None otherwise.
|
||
"""
|
||
try:
|
||
sp = SerialPort(port)
|
||
info = sp.detect(self._settings.wafer_detect_timeout)
|
||
return info
|
||
except Exception as exc:
|
||
log.error("Detect failed on %s: %s", port, exc)
|
||
return None
|
||
|
||
SCAN_PER_PORT_TIMEOUT_MS = 1500
|
||
"""Per-port timeout while scanning. Wafer responds fast or not at all."""
|
||
|
||
def detect_all_ports(self) -> tuple[Optional[str], Optional[WaferInfo]]:
|
||
"""Scan every available serial port until one responds to s0.
|
||
|
||
Mirrors C# chkConnection(): enumerates all ports, tries each one,
|
||
returns the first that responds with a valid 1024-byte payload.
|
||
|
||
Uses a short per-port timeout (SCAN_PER_PORT_TIMEOUT_MS) so a
|
||
machine with many serial devices (Bluetooth, debug consoles, etc.)
|
||
doesn't take 30+ seconds to scan. The configured wafer_detect_timeout
|
||
is reserved for confirmed-port operations.
|
||
|
||
Returns:
|
||
(port_name, WaferInfo) on success, (None, None) if nothing found.
|
||
"""
|
||
ports = self.enumerate_ports()
|
||
log.info("Scanning %d port(s) for wafer ...", len(ports))
|
||
for port in ports:
|
||
log.info(" trying %s ...", port)
|
||
try:
|
||
sp = SerialPort(port)
|
||
info = sp.detect(self.SCAN_PER_PORT_TIMEOUT_MS)
|
||
except Exception as exc:
|
||
log.warning(" %s: %s", port, exc)
|
||
info = None
|
||
if info is not None:
|
||
log.info(" found wafer on %s (family=%s)", port, info.family_code)
|
||
return port, info
|
||
log.info(" no wafer detected on any port")
|
||
return None, None
|
||
|
||
def read_wafer_data(
|
||
self,
|
||
port: str,
|
||
family_code: str = "",
|
||
cmd: str = "D1",
|
||
timeout_ms: int | None = None,
|
||
retries: int | None = None,
|
||
) -> Optional[bytes]:
|
||
"""Full read workflow: open → send command → retry → close.
|
||
|
||
Args:
|
||
port: Serial port name (e.g. "/dev/ttyUSB0").
|
||
family_code: Wafer family ("P", "X", etc.) for size validation.
|
||
cmd: Command string ("D1" for sensor data, "F1" for debug data).
|
||
timeout_ms: Read timeout in ms (uses settings default if None).
|
||
retries: Max retries on bad read (uses settings default if None).
|
||
|
||
Returns:
|
||
Raw bytes from the wafer, or None on failure.
|
||
"""
|
||
if timeout_ms is None:
|
||
timeout_ms = self._settings.wafer_read_timeout
|
||
if retries is None:
|
||
retries = self._settings.wafer_read_retries
|
||
|
||
# Determine expected hex string length.
|
||
# get_wafer_data_size returns BYTES; hex string is 2× that.
|
||
if family_code:
|
||
expected_hex_len = self._settings.get_wafer_data_size(family_code) * 2
|
||
else:
|
||
expected_hex_len = 0 # no validation
|
||
|
||
try:
|
||
sp = SerialPort(port)
|
||
data = sp.read_memory(
|
||
cmd=cmd,
|
||
expected_hex_len=expected_hex_len,
|
||
timeout_ms=timeout_ms,
|
||
retries=retries,
|
||
)
|
||
if data is not None:
|
||
log.info("Read %d bytes from wafer on %s", len(data), port)
|
||
else:
|
||
log.warning("Read returned no data from %s", port)
|
||
return data
|
||
except Exception as exc:
|
||
log.error("Read failed on %s: %s", port, exc)
|
||
return None
|
||
|
||
def erase_wafer(self, port: str) -> bool:
|
||
"""Send p1 erase command. Wafer takes ~15s to erase.
|
||
|
||
Returns True if command was sent successfully.
|
||
"""
|
||
try:
|
||
sp = SerialPort(port)
|
||
return sp.erase_memory()
|
||
except Exception as exc:
|
||
log.error("Erase failed on %s: %s", port, exc)
|
||
return False
|
||
|
||
def get_expected_hex_length(self, family_code: str) -> int:
|
||
"""Return expected hex string length for a wafer family."""
|
||
return self._settings.get_wafer_data_size(family_code)
|