"""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, )