Jack.Le
e306db6816
- SettingsTab: chamber ID, master CSV mapping (families A-Z), wafer behavior toggle, light/dark mode toggle, Edit CSV Metadata dialog - StatusTab: connection status card, wafer info, activity log (blank until Detect Wafer fires from the side rail) - DataTab: empty state, ready for parsed data - HomePage: side rail buttons dispatch device actions and jump to Status tab; selectedSideActionIndex defaults to -1 on startup - DeviceController registered as QML context property in main.py - LocalSettings: added status persistence fields - Theme: added subheadingColor and disabledText - Added serialcomm/ package and backend data/graph modules - gitignore: restore clean version
204 lines
7.3 KiB
Python
204 lines
7.3 KiB
Python
"""Low-level serial port communication with the wafer device.
|
||
|
||
Mirrors the C# Form1.cs wafer protocol:
|
||
- Detect : s0 + 510×'F' → 1024 hex chars → WaferInfo
|
||
- ReadMemory: D1 + 510×'F' → N hex chars (retry on bad length)
|
||
- Erase : p1 + 510×'F' → no response (~15s blocking)
|
||
|
||
All commands are 512-char strings: 2-char command + 510×'F' padding.
|
||
"""
|
||
|
||
import logging
|
||
from contextlib import contextmanager
|
||
from dataclasses import dataclass
|
||
from typing import Iterator, Optional
|
||
|
||
import serial as pyserial
|
||
|
||
log = logging.getLogger(__name__)
|
||
|
||
|
||
@dataclass
|
||
class WaferInfo:
|
||
"""Parsed wafer identification from the s0 detect response."""
|
||
|
||
family_code: str # "P", "X", "A", "B", "C", "D", "E", "F"
|
||
serial_number: str # e.g. "P00001"
|
||
sensor_count: int
|
||
mfg_date_hex: str # raw 8 hex chars
|
||
sensor_assigned_hex: str # raw 2 hex chars
|
||
locked_hex: str # raw 2 hex chars
|
||
runtime: int # seconds
|
||
cycle_count: int
|
||
|
||
|
||
class SerialPort:
|
||
"""Low-level serial transport for the temperature-sensing wafer.
|
||
|
||
The serial port is opened per-operation (matching C# behaviour)
|
||
rather than kept open persistently.
|
||
"""
|
||
|
||
BAUDRATE = 888888
|
||
COMMAND_PAD = "F" * 510 # pad to 512 chars total
|
||
|
||
def __init__(self, port_name: str) -> None:
|
||
self._port_name = port_name
|
||
|
||
# ------------------------------------------------------------------
|
||
# Context manager for scoped port access
|
||
# ------------------------------------------------------------------
|
||
|
||
@contextmanager
|
||
def _open(self, timeout: float | None = None) -> Iterator[pyserial.Serial]:
|
||
"""Open the serial port, yield it, then close on exit."""
|
||
port = pyserial.Serial(
|
||
self._port_name,
|
||
self.BAUDRATE,
|
||
parity=pyserial.PARITY_NONE,
|
||
bytesize=8,
|
||
stopbits=pyserial.STOPBITS_ONE,
|
||
xonxoff=False,
|
||
rtscts=False,
|
||
dsrdtr=False,
|
||
)
|
||
if timeout is not None:
|
||
port.timeout = timeout
|
||
try:
|
||
yield port
|
||
finally:
|
||
port.close()
|
||
|
||
# ------------------------------------------------------------------
|
||
# Public API
|
||
# ------------------------------------------------------------------
|
||
|
||
def detect(self, timeout_ms: int = 5000) -> Optional[WaferInfo]:
|
||
"""Try s0 command. Returns WaferInfo if response is 1024 hex chars."""
|
||
with self._open(timeout_ms / 1000) as port:
|
||
port.write(("s0" + self.COMMAND_PAD).encode())
|
||
response = port.readline().decode().strip()
|
||
if len(response) == 1024:
|
||
return self._parse_wafer_info(response)
|
||
return None
|
||
|
||
def read_memory(
|
||
self,
|
||
cmd: str = "D1",
|
||
expected_hex_len: int = 393216,
|
||
timeout_ms: int = 120000,
|
||
retries: int = 8,
|
||
) -> Optional[bytes]:
|
||
"""Send read command, retry on bad length or odd-length string.
|
||
|
||
The wafer sends hex-encoded bytes. A dropped byte corrupts the
|
||
alignment, so the entire buffer must be retried.
|
||
|
||
Returns raw bytes on success, None on failure.
|
||
"""
|
||
with self._open(timeout_ms / 1000) as port:
|
||
port.write((cmd + self.COMMAND_PAD).encode())
|
||
|
||
for attempt in range(retries, 0, -1):
|
||
data_string = port.readline().decode(errors="ignore").strip()
|
||
|
||
# Trim trailing odd nibble — wafer sometimes drops 1 char
|
||
# at the tail; one stray nibble is harmless, the rest is good.
|
||
if len(data_string) % 2 != 0:
|
||
data_string = data_string[:-1]
|
||
|
||
# Defensively cap at expected length if the wafer over-sends.
|
||
if expected_hex_len > 0 and len(data_string) > expected_hex_len:
|
||
data_string = data_string[:expected_hex_len]
|
||
|
||
# Accept if we got at least 99% of expected (matches C# tolerance).
|
||
threshold = int(expected_hex_len * 0.99) if expected_hex_len > 0 else 1
|
||
if len(data_string) >= threshold:
|
||
break
|
||
|
||
log.warning(
|
||
"Short hex string (len=%d, expected~%d), retries left=%d",
|
||
len(data_string),
|
||
expected_hex_len,
|
||
attempt - 1,
|
||
)
|
||
|
||
if attempt > 1:
|
||
port.reset_input_buffer()
|
||
# Re-issue the command — reset_input_buffer alone leaves
|
||
# the wafer waiting for nothing.
|
||
port.write((cmd + self.COMMAND_PAD).encode())
|
||
continue
|
||
|
||
log.error("Retries exhausted after bad read")
|
||
return None
|
||
|
||
try:
|
||
return bytes.fromhex(data_string)
|
||
except ValueError as exc:
|
||
log.error("Failed to decode hex: %s", exc)
|
||
return None
|
||
|
||
def erase_memory(self, timeout_ms: int = 5000) -> bool:
|
||
"""Send p1 erase command. Wafer takes ~15s to erase.
|
||
|
||
Returns True if command was sent successfully.
|
||
"""
|
||
with self._open(timeout_ms / 1000) as port:
|
||
try:
|
||
port.write(("p1" + self.COMMAND_PAD).encode())
|
||
return True
|
||
except Exception as exc:
|
||
log.error("Erase command failed: %s", exc)
|
||
return False
|
||
|
||
# ------------------------------------------------------------------
|
||
# Helpers
|
||
# ------------------------------------------------------------------
|
||
|
||
@staticmethod
|
||
def _hex_to_ascii(hex_str: str) -> str:
|
||
"""Convert hex string (e.g. '50') to ASCII char ('P')."""
|
||
hex_str = hex_str.replace(" ", "")
|
||
result = []
|
||
for i in range(0, len(hex_str), 2):
|
||
hex_char = hex_str[i : i + 2]
|
||
if len(hex_char) == 2:
|
||
result.append(chr(int(hex_char, 16)))
|
||
return "".join(result)
|
||
|
||
def _parse_wafer_info(self, hex_response: str) -> WaferInfo:
|
||
"""Parse the 1024-char hex response into WaferInfo.
|
||
|
||
Layout (from C# checkPort):
|
||
bytes 0-1 : FamilyCode (2 hex chars → ASCII)
|
||
bytes 2-5 : Serial number (3 bytes hex → decimal, 5 digits)
|
||
bytes 6-7 : Sensor count (1 byte hex)
|
||
bytes 8-15 : Mfg date (4 bytes hex)
|
||
bytes 16-17 : Sensor assigned (1 byte hex)
|
||
bytes 18-19 : Locked (1 byte hex)
|
||
bytes 1016-1019 : Runtime (2 bytes hex → decimal seconds)
|
||
bytes 1020-1023 : Cycle time (2 bytes hex → decimal cycles)
|
||
"""
|
||
raw = bytes.fromhex(hex_response)
|
||
|
||
family_code = self._hex_to_ascii(hex_response[0:2])
|
||
serial_num = f"{family_code}{int(hex_response[2:6], 16):05d}"
|
||
sensor_count = int(hex_response[6:8], 16)
|
||
mfg_date_hex = hex_response[8:16]
|
||
sensor_assigned_hex = hex_response[16:18]
|
||
locked_hex = hex_response[18:20]
|
||
runtime = int(hex_response[1016:1020], 16)
|
||
cycle_count = int(hex_response[1020:1024], 16)
|
||
|
||
return WaferInfo(
|
||
family_code=family_code,
|
||
serial_number=serial_num,
|
||
sensor_count=sensor_count,
|
||
mfg_date_hex=mfg_date_hex,
|
||
sensor_assigned_hex=sensor_assigned_hex,
|
||
locked_hex=locked_hex,
|
||
runtime=runtime,
|
||
cycle_count=cycle_count,
|
||
)
|