Restructure into src/ layout under pygui package
Move all application source under src/pygui/ and rewire imports, build config, and QML module path to match. - Relocate backend/, serialcomm/, and the ISC QML module into src/pygui/; convert main.py into pygui/__main__.py with a main() entry point (run via `python -m pygui` or the new `isc` script) - Rewrite absolute imports: backend.* -> pygui.backend.*, serialcomm.* -> pygui.serialcomm.* (source + tests) - Move app icons (isc.ico/icns) into packaging/ - Update README and ISC.qmlproject to the new paths
This commit is contained in:
jack
@@ -0,0 +1,203 @@
|
||||
"""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,
|
||||
)
|
||||
Reference in New Issue
Block a user