Add SettingsTab, StatusTab, DataTab and wire up DeviceController
- SettingsTab: chamber ID, master CSV mapping, wafer behavior toggle, light/dark mode toggle, Edit CSV Metadata dialog - StatusTab: connection status, wafer info, activity log (blank until Detect Wafer fires) - DataTab: empty state, ready for parsed data - HomePage: side rail buttons dispatch device actions and jump to Status tab; selectedSideActionIndex defaults to -1 (nothing active 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: added build/, dist/, *.spec
This commit is contained in:
Jack.Le
@@ -0,0 +1,6 @@
|
||||
"""Serial port communication layer for the temperature-sensing wafer."""
|
||||
|
||||
from serialcomm.device_service import DeviceService
|
||||
from serialcomm.serial_port import SerialPort, WaferInfo
|
||||
|
||||
__all__ = ["DeviceService", "SerialPort", "WaferInfo"]
|
||||
@@ -0,0 +1,292 @@
|
||||
"""Binary data parser and temperature converter for wafer data.
|
||||
|
||||
Mirrors the C# Form1.cs binary parsing pipeline:
|
||||
1. Read raw bytes → strip per-block overhead → 1D hex array
|
||||
2. Chunk 1D array into rows × sensors → List[List[str]] (hex values)
|
||||
3. Convert hex values → float temperatures (family-dependent)
|
||||
4. Remove trailing zero rows
|
||||
5. Save to CSV with Sensor1, Sensor2, ... headers
|
||||
"""
|
||||
|
||||
import logging
|
||||
import os
|
||||
from typing import Optional
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
# Max DUTs (sensors) per row before overhead bytes are stripped
|
||||
# P wafer: 244 valid readings per 256-block (12 overhead bytes)
|
||||
# X wafer: 80 valid readings per 256-block (14 overhead bytes)
|
||||
MAXDUT_P = 244
|
||||
MAXDUT_X = 80
|
||||
|
||||
|
||||
def csv_column_count(family_code: str) -> int:
|
||||
"""Return the number of columns to display for a family code."""
|
||||
mapping = {
|
||||
"A": 48,
|
||||
"E": 48,
|
||||
"P": 48,
|
||||
"B": 29,
|
||||
"C": 29,
|
||||
"D": 29,
|
||||
"F": 22,
|
||||
"X": 80,
|
||||
}
|
||||
return mapping.get(family_code, 0)
|
||||
|
||||
|
||||
def _hex_to_binary(hex_str: str) -> list[int]:
|
||||
"""Convert a 4-char hex string to a 16-bit binary list (MSB first)."""
|
||||
value = int(hex_str, 16)
|
||||
return [(value >> (15 - i)) & 1 for i in range(16)]
|
||||
|
||||
|
||||
def _twos_complement_excluding_msb(bits: list[int]) -> list[int]:
|
||||
"""Invert bits 1-15, add 1 (2's complement excluding MSB)."""
|
||||
bits = list(bits) # copy
|
||||
for i in range(1, len(bits)):
|
||||
bits[i] = 1 - bits[i]
|
||||
carry = 1
|
||||
for i in range(len(bits) - 1, 0, -1):
|
||||
s = bits[i] + carry
|
||||
bits[i] = s % 2
|
||||
carry = s // 2
|
||||
return bits
|
||||
|
||||
|
||||
def _binary_subsequence_to_int(bits: list[int], start: int, end: int) -> int:
|
||||
"""Convert bits[start:end+1] to integer."""
|
||||
result = 0
|
||||
for i in range(start, end + 1):
|
||||
result = (result << 1) | bits[i]
|
||||
return result
|
||||
|
||||
|
||||
def _binary_fraction_to_double(bits: list[int], start: int) -> float:
|
||||
"""Convert bits[start:] to a fractional value (0 < frac < 1)."""
|
||||
result = 0.0
|
||||
divisor = 2.0
|
||||
for i in range(start, len(bits)):
|
||||
result += bits[i] / divisor
|
||||
divisor *= 2.0
|
||||
return result
|
||||
|
||||
|
||||
def _convert_standard(binary_bits: list[int]) -> float:
|
||||
"""Convert 16-bit binary using standard formula (B/C/D/F families).
|
||||
|
||||
Bit layout:
|
||||
bit 0 : sign
|
||||
bits 1-11 : integer part (11 bits, 2^10 .. 2^0)
|
||||
bits 12-13 : fractional part (2 bits, 2^-1, 2^-2)
|
||||
bits 14-15 : unused
|
||||
"""
|
||||
value = 0.0
|
||||
for i in range(1, 14):
|
||||
if binary_bits[i]:
|
||||
value += 2.0 ** (11 - i)
|
||||
if binary_bits[0]:
|
||||
value = -value
|
||||
return value
|
||||
|
||||
|
||||
def _convert_aep(binary_bits: list[int]) -> float:
|
||||
"""Convert 16-bit binary using AEP formula.
|
||||
|
||||
Bit layout:
|
||||
bit 0 : sign
|
||||
bits 1-8 : integer part (8 bits)
|
||||
bits 9-15 : fractional part (7 bits)
|
||||
"""
|
||||
bits = binary_bits
|
||||
if bits[0] == 1:
|
||||
bits = _twos_complement_excluding_msb(bits)
|
||||
integer_part = _binary_subsequence_to_int(bits, 1, 8)
|
||||
fractional_part = _binary_fraction_to_double(bits, 9)
|
||||
result = integer_part + fractional_part
|
||||
if binary_bits[0] == 1:
|
||||
result = -result
|
||||
return result
|
||||
|
||||
|
||||
def _convert_hex_to_temp(hex_str: str, family_code: str) -> float:
|
||||
"""Convert a singi hale 4-char hex string to a float temperature."""
|
||||
bits = _hex_to_binary(hex_str)
|
||||
if family_code in ("A", "E", "P"):
|
||||
result = _convert_aep(bits)
|
||||
elif family_code in ("B", "C", "D"):
|
||||
result = _convert_standard(bits)
|
||||
if result < -2000:
|
||||
result = 0.0
|
||||
elif family_code == "X":
|
||||
# X wafer uses AEP-style conversion
|
||||
result = _convert_aep(bits)
|
||||
else:
|
||||
# Unknown family — try standard
|
||||
result = _convert_standard(bits)
|
||||
return round(result, 2)
|
||||
|
||||
|
||||
def parse_binary_data(data_bytes: bytes, family_code: str) -> Optional[list[list[str]]]:
|
||||
"""Parse raw wafer bytes into a 2D array of hex strings.
|
||||
|
||||
Strips per-block overhead bytes (12 for P, 14 for X) and chunks
|
||||
the remaining readings into rows.
|
||||
|
||||
Args:
|
||||
data_bytes: Raw binary data from the wafer.
|
||||
family_code: Wafer family code ("P", "X", "A", "B", "C", "D", "F").
|
||||
|
||||
Returns:
|
||||
List of rows, each row is a list of 4-char hex strings.
|
||||
Returns None on failure.
|
||||
"""
|
||||
try:
|
||||
if family_code == "X":
|
||||
return _parse_x_binary(data_bytes)
|
||||
else:
|
||||
return _parse_p_binary(data_bytes)
|
||||
except Exception as exc:
|
||||
log.error("Binary parse failed: %s", exc)
|
||||
return None
|
||||
|
||||
|
||||
def _parse_p_binary(data_bytes: bytes) -> list[list[str]]:
|
||||
"""Parse P-family (and A/B/C/D/E/F) binary data.
|
||||
|
||||
Each block of 256 readings has 244 valid + 12 overhead.
|
||||
Valid readings are chunked into rows of MAXDUT_P (244).
|
||||
"""
|
||||
readings: list[str] = []
|
||||
|
||||
# Read 2 bytes at a time (UInt16 little-endian)
|
||||
# Each 256-word block has MAXDUT_P valid readings (first N words), rest are overhead
|
||||
num_words = len(data_bytes) // 2
|
||||
for i in range(num_words):
|
||||
value = int.from_bytes(data_bytes[i * 2 : i * 2 + 2], byteorder="little")
|
||||
if i % 256 < MAXDUT_P:
|
||||
readings.append(f"{value:04X}")
|
||||
|
||||
# Chunk into rows of MAXDUT_P
|
||||
result: list[list[str]] = []
|
||||
idx = 0
|
||||
while idx + MAXDUT_P <= len(readings):
|
||||
result.append(readings[idx : idx + MAXDUT_P])
|
||||
idx += MAXDUT_P
|
||||
|
||||
log.info("Parsed P-family: %d rows × %d cols", len(result), MAXDUT_P)
|
||||
return result
|
||||
|
||||
|
||||
def _parse_x_binary(data_bytes: bytes) -> list[list[str]]:
|
||||
"""Parse X-family binary data.
|
||||
|
||||
Each block of 256 readings has 80 valid + 14 overhead.
|
||||
Valid readings are chunked into rows of MAXDUT_X (80).
|
||||
"""
|
||||
readings: list[str] = []
|
||||
|
||||
num_words = len(data_bytes) // 2
|
||||
for i in range(num_words):
|
||||
value = int.from_bytes(data_bytes[i * 2 : i * 2 + 2], byteorder="little")
|
||||
if i % 256 < MAXDUT_X:
|
||||
readings.append(f"{value:04X}")
|
||||
|
||||
# Chunk into rows of MAXDUT_X
|
||||
result: list[list[str]] = []
|
||||
idx = 0
|
||||
while idx + MAXDUT_X <= len(readings):
|
||||
result.append(readings[idx : idx + MAXDUT_X])
|
||||
idx += MAXDUT_X
|
||||
|
||||
log.info("Parsed X-family: %d rows × %d cols", len(result), MAXDUT_X)
|
||||
return result
|
||||
|
||||
|
||||
def convert_to_temperatures(
|
||||
hex_data: list[list[str]], family_code: str
|
||||
) -> list[list[str]]:
|
||||
"""Convert hex string values to temperature strings.
|
||||
|
||||
Args:
|
||||
hex_data: 2D array of 4-char hex strings from parse_binary_data.
|
||||
family_code: Wafer family code.
|
||||
|
||||
Returns:
|
||||
Same structure with temperature strings (e.g. "25.37").
|
||||
"""
|
||||
temp_value: list[list[str]] = []
|
||||
for row in hex_data:
|
||||
temp_row: list[str] = []
|
||||
for hex_val in row:
|
||||
temp = _convert_hex_to_temp(hex_val, family_code)
|
||||
temp_row.append(str(temp))
|
||||
temp_value.append(temp_row)
|
||||
return temp_value
|
||||
|
||||
|
||||
def remove_trailing_zeros(data: list[list[str]]) -> None:
|
||||
"""Remove rows of all-zero values from the end of data (in-place).
|
||||
|
||||
A value is considered zero if its float representation is < 0.01.
|
||||
"""
|
||||
while data:
|
||||
last_row = data[-1]
|
||||
is_all_zero = True
|
||||
for val in last_row:
|
||||
try:
|
||||
fval = float(val)
|
||||
except ValueError:
|
||||
fval = 99.9
|
||||
if fval >= 0.01:
|
||||
is_all_zero = False
|
||||
break
|
||||
if is_all_zero:
|
||||
data.pop()
|
||||
else:
|
||||
break
|
||||
|
||||
|
||||
def save_to_csv(
|
||||
data: list[list[str]],
|
||||
family_code: str,
|
||||
serial_number: str,
|
||||
output_dir: str,
|
||||
) -> Optional[str]:
|
||||
"""Save parsed temperature data to a CSV file.
|
||||
|
||||
Args:
|
||||
data: 2D array of temperature strings.
|
||||
family_code: Wafer family code.
|
||||
serial_number: Wafer serial number (e.g. "P00001").
|
||||
output_dir: Directory to save the CSV file.
|
||||
|
||||
Returns:
|
||||
Full file path on success, None on failure.
|
||||
"""
|
||||
try:
|
||||
os.makedirs(output_dir, exist_ok=True)
|
||||
# Build filename: P00001-20260505_133045.csv
|
||||
from datetime import datetime
|
||||
|
||||
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
|
||||
filename = f"{serial_number}-{timestamp}.csv"
|
||||
filepath = os.path.join(output_dir, filename)
|
||||
|
||||
num_cols = csv_column_count(family_code)
|
||||
with open(filepath, "w", encoding="utf-8") as f:
|
||||
# Header row: Sensor1,Sensor2,...
|
||||
headers = [f"Sensor{j + 1}" for j in range(num_cols)]
|
||||
f.write(",".join(headers) + "\n")
|
||||
# Data rows
|
||||
for row in data:
|
||||
# Pad or trim to match header count
|
||||
padded = row[:num_cols] + ["0"] * max(0, num_cols - len(row))
|
||||
f.write(",".join(padded) + "\n")
|
||||
|
||||
log.info("Saved %d rows × %d cols to %s", len(data), num_cols, filepath)
|
||||
return filepath
|
||||
except Exception as exc:
|
||||
log.error("CSV save failed: %s", exc)
|
||||
return None
|
||||
@@ -0,0 +1,152 @@
|
||||
"""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 pathlib import Path
|
||||
from typing import Optional
|
||||
|
||||
import serial.tools.list_ports
|
||||
|
||||
from backend.local_settings import LocalSettings
|
||||
from 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."""
|
||||
return [p.device for p in serial.tools.list_ports.comports()]
|
||||
|
||||
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)
|
||||
@@ -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