jack/pyGUI
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(wafer): introduce layout metadata modeling and shape resolution helpers

Browse Source 
- Define WaferLayout subclass to wrap sensor lists with shape and size attributes.
- Add resolve_shape_and_size resolver to infer wafer shape/size from filename prefix or sensor count.
- Update Sensor dataclass to support dynamic side alignments and coordinates offsets.
- Expose waferShape and waferSize properties from SessionController for QML data-bindings.
This commit is contained in:
jack
2026-06-15 11:19:32 -07:00
parent 5b186df888
commit cea4fb782e
5 changed files with 217 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/pygui/backend/controllers/session_controller.py
+113 -15
View File
@@ -70,7 +70,14 @@ class SessionController(QObject):
def frameTotal(self) -> int: return self._player.total
@Property(str, notify=stateChanged)
def state(self) -> str: return self._last.state if self._last else "idle"
def state(self) -> str:
if self._last:
return self._last.state
# IF we dont have data yet, but the reader is running -> Streaming
if self._mode == MODE_LIVE and self._reader is not None:
return "streaming"
return "idle"
@Property(bool, notify=recordingChanged)
def recording(self) -> bool: return self._recorder.is_recording
@@ -88,11 +95,30 @@ class SessionController(QObject):
"value": round(v, 2), "band": band, "index": i})
return out
# ---- WaferMapItem bindings (split view of sensorDots) ----
@Property("QVariantList", notify=sensorsChanged)
def sensorLayout(self) -> list:
"""[{label, x, y}] for WaferMapItem.sensors."""
return [{"label": s.label, "x": s.x, "y": s.y} for s in self._sensors]
"""[{label, x, y, side, offset_x, offset_y}] for WaferMapItem.sensors."""
return [
{
"label": s.label,
"x": s.x,
"y": s.y,
"side": getattr(s, "side", "right"),
"offset_x": getattr(s, "offset_x", 0.0),
"offset_y": getattr(s, "offset_y", 0.0),
}
for s in self._sensors
]
@Property(str, notify=sensorsChanged)
def waferShape(self) -> str:
"""Wafer shape: 'round' or 'square'."""
return getattr(self._sensors, "shape", "round")
@Property(float, notify=sensorsChanged)
def waferSize(self) -> float:
"""Wafer size in mm."""
return getattr(self._sensors, "size", 300.0)
@Property("QVariantList", notify=frameUpdated)
def sensorValues(self) -> list:
@@ -155,7 +181,9 @@ class SessionController(QObject):
# ---- review: file load + playback ----
@Slot(str)
def loadFile(self, file_path: str) -> None:
from pathlib import Path
from pygui.backend.data.data_records import is_official_csv, read_data_records, read_official_csv
from pygui.backend.wafer.wafer_layouts import WaferLayout, resolve_shape_and_size
sensors: list[Sensor] = []
frames = []
@@ -180,7 +208,15 @@ class SessionController(QObject):
log.warning("No sensors or data in %s", file_path)
return
self._sensors = sensors
wafer_id = ""
if not is_official_csv(file_path):
wafer_id = data.serial if (data and data.serial) else ""
else:
stem = Path(file_path).stem
wafer_id = stem.split("-")[0] if "-" in stem else stem
shape, size = resolve_shape_and_size(sensors, wafer_id)
self._sensors = WaferLayout(sensors, shape=shape, size=size)
self._player.load(frames)
self._model.reset()
self._loaded_file = file_path
@@ -256,28 +292,90 @@ class SessionController(QObject):
self.stateChanged.emit()
# ---- live: stream start/stop ----
@Slot(str)
def startStream(self, port: str) -> None:
@Slot(str, str)
def startStream(self, port: str, family_code: str = "") -> None:
from pygui.serialcomm.serial_port import SerialPort # transport open
import serial as pyserial
from pygui.backend.wafer.wafer_layouts import load_layout_for_wafer_id
import time
def parse_line(raw: str, seq: int) -> Frame:
parts = [float(x) for x in raw.split(",")]
return Frame(seq=seq, time=parts[0], values=parts[1:])
if family_code:
try:
self._sensors = load_layout_for_wafer_id(family_code)
self.sensorsChanged.emit()
except Exception as e:
log.warning("Could not load layout for %s: %s", family_code, e)
# The new binary protocol sends payload of (sensorCount * 2) bytes
# Each sensor is a big-endian 16-bit value (Sign-Magnitude).
def parse_binary_frame(payload: bytes, seq: int) -> Frame:
values = []
num_sensors = min(80, len(payload) // 2)
for i in range(num_sensors):
high_byte = payload[i * 2]
low_byte = payload[i * 2 + 1]
val16 = (high_byte << 8) | low_byte
is_negative = (val16 & 0x8000) != 0
integer_part = (val16 >> 7) & 0xFF
fractional_bits = val16 & 0x7F
fractional_part = fractional_bits / 128.0
result = integer_part + fractional_part
if is_negative:
result = -result
values.append(result)
# Create a mock timestamp or use elapsed time since start.
# We don't have a time value from the binary stream header right now.
t = seq * 0.05 # approx 20Hz if seq increments 1 per frame
return Frame(seq=seq, time=t, values=values)
# Clear out any old data from the prev sessions
self._model.reset()
self._last = None
self._last_raw_frame = None
try:
transport = pyserial.Serial(port, SerialPort.BAUDRATE, timeout=1)
except Exception as exc:
log.warning("Baud rate %d failed on %s, falling back to 115200: %s", SerialPort.BAUDRATE, port, exc)
transport = pyserial.Serial(port, 115200, timeout=1)
# Send 'D2' command padded to 512 bytes to start the stream
cmd = b"D2" + (b"F" * 510)
transport.write(cmd)
def on_error(exc: Exception):
log.error("Live stream error: %s", exc)
transport = pyserial.Serial(port, SerialPort.BAUDRATE, timeout=1)
self._reader = StreamReader(
transport, parse_line,
on_frame=lambda f: self._liveFrame.emit(f))
transport, parse_binary_frame,
on_frame=lambda f: self._liveFrame.emit(f),
on_error=on_error,
family_code=family_code or "A")
self._reader.start()
self._repaint_timer.start() # Q1: begin ~20 Hz repaints
self._repaint_timer.start()
self.setMode("live")
self.stateChanged.emit()
@Slot()
def stopStream(self) -> None:
self._repaint_timer.stop()
if self._reader is not None:
if self._reader:
transport = self._reader._transport
self._reader.stop()
self._reader = None
# Send 'D2S' command padded to 512 bytes to stop the stream
if transport and transport.is_open:
try:
cmd = b"D2S" + (b"F" * 509)
transport.write(cmd)
transport.flush()
except Exception as exc:
log.error("Error sending stop command: %s", exc)
self.stateChanged.emit()
self.stopRecording()
@Slot(object)