"""QML-facing controller for the live/review wafer dashboard.""" from __future__ import annotations import logging import time from typing import Any, Optional from PySide6.QtCore import QObject, Property, QTimer, Qt, Signal, Slot from pygui.backend.data.csv_recorder import CsvRecorder from pygui.backend.models.frame import Frame from pygui.backend.models.frame_player import FramePlayer, frames_from_wafer_data from pygui.backend.models.session_model import SessionModel from pygui.backend.models.threshold_classifier import ThresholdConfig from pygui.backend.wafer.zwafer_models import Sensor from pygui.backend.wafer.zwafer_parser import ZWaferParser from pygui.serialcomm.stream_reader import StreamReader from pygui.backend.models.sensor_editor import SensorEditor from pygui.backend.cluster_average import average_clusters, group_sensors_by_radius log = logging.getLogger(__name__) MODE_LIVE = "live" MODE_REVIEW = "review" class SessionController(QObject): # public signals frameUpdated = Signal() modeChanged = Signal() stateChanged = Signal() recordingChanged = Signal() sensorsChanged = Signal() loadedFileChanged = Signal() clusterAveragingEnabledChanged = Signal() # private: marshal a worker-thread frame onto the main thread _liveFrame = Signal(object) # Frame def __init__(self, parent: QObject | None = None) -> None: super().__init__(parent) self._mode = MODE_REVIEW self._model = SessionModel() self._player = FramePlayer() self._reader: Optional[StreamReader] = None self._recorder = CsvRecorder() self._sensors: list[Sensor] = [] self._last = None # last SessionUpdate self._elapsed = 0.0 self._play_timer = QTimer(self) self._play_timer.timeout.connect(self._advance) self._speed = 1.0 # Q1: coalesce live repaints to ~20 Hz; data is still processed every frame. self._repaint_timer = QTimer(self) self._repaint_timer.setInterval(50) # ~20 Hz self._repaint_timer.timeout.connect(self._flush_repaint) self._dirty = False self._liveFrame.connect(self._on_live_frame, Qt.ConnectionType.QueuedConnection) self._sensor_editor = SensorEditor() self._last_raw_frame: Frame | None = None self._loaded_file: str = "" self._cluster_averaging_enabled = False self._active_clusters: list[list[int]] = [] # ---- properties QML binds to ---- @Property(str, notify=modeChanged) def mode(self) -> str: return self._mode @Property(int, notify=frameUpdated) def frameIndex(self) -> int: return self._player.index @Property(int, notify=frameUpdated) def frameTotal(self) -> int: return self._player.total @Property(str, notify=stateChanged) 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 @Property(list, notify=frameUpdated) def sensorDots(self) -> list[dict]: """Per-sensor render data for the radial map.""" if not self._last: return [] out = [] for i, s in enumerate(self._sensors): v = self._last.values[i] if i < len(self._last.values) else 0.0 band = self._last.bands[i] if i < len(self._last.bands) else "in_range" out.append({"label": s.label, "x": s.x, "y": s.y, "value": round(v, 2), "band": band, "index": i}) return out @Property("QVariantList", notify=sensorsChanged) def sensorLayout(self) -> list: """[{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: """[float] in sensor order.""" if not self._last: return [] return [round(v, 3) for v in self._last.values] @Property("QVariantList", notify=frameUpdated) def sensorBands(self) -> list: """['in_range'|'high'|'low'] in sensor order.""" if not self._last: return [] return list(self._last.bands) @Property(float, notify=frameUpdated) def target(self) -> float: """Resolved band center (frame mean in auto mode, else set_point).""" return self._last.target if self._last else 149.0 @Property(float, notify=frameUpdated) def margin(self) -> float: """Resolved band half-width (frame 1σ in auto mode, else margin).""" return self._last.margin if self._last else 1.0 @Property(dict, notify=frameUpdated) def stats(self) -> dict: if not self._last: return {} s = self._last.stats return {"min": round(s.min, 2), "minIndex": s.min_index + 1, "max": round(s.max, 2), "maxIndex": s.max_index + 1, "diff": round(s.diff, 2), "avg": round(s.avg, 2), "sigma": round(s.sigma, 2), "threeSigma": round(s.three_sigma, 2)} @Property(str, notify=loadedFileChanged) def loadedFile(self) -> str: return self._loaded_file @Property("QVariantList", notify=frameUpdated) def overriddenSensors(self) -> list[int]: """Indices of sensors that currently have a replacement or offset.""" return self._sensor_editor.active_indices() @Property(bool, notify=clusterAveragingEnabledChanged) def clusterAveragingEnabled(self) -> bool: return self._cluster_averaging_enabled @clusterAveragingEnabled.setter def clusterAveragingEnabled(self, value: bool) -> None: if self._cluster_averaging_enabled == value: return self._cluster_averaging_enabled = value self.clusterAveragingEnabledChanged.emit() self._reprocess_current() # ---- mode + thresholds ---- @Slot(str) def setMode(self, mode: str) -> None: if mode == self._mode: return self.stopStream() self._play_timer.stop() self._mode = mode self.modeChanged.emit() @Slot(float, float, bool) def setThresholds(self, set_point: float, margin: float, auto: bool) -> None: self._model.set_thresholds(ThresholdConfig(set_point, margin, auto)) if self._last: # re-band current frame in place self._reprocess_current() # ---- 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 frames = [] if is_official_csv(file_path): sensors, records = read_official_csv(file_path) frames = frames_from_wafer_data(None, records) else: try: data, _ = ZWaferParser().parse(file_path) except (ValueError, KeyError) as exc: log.warning("Could not parse %s: %s", file_path, exc) return if data is None or not data.sensors: log.warning("Could not parse %s", file_path) return records = read_data_records(file_path) sensors = data.sensors frames = frames_from_wafer_data(data, records) if not sensors or not frames: log.warning("No sensors or data in %s", file_path) return 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._active_clusters = getattr(self._sensors, 'clusters', []) if not self._active_clusters: self._active_clusters = group_sensors_by_radius(self._sensors) self._player.load(frames) self._model.reset() self._loaded_file = file_path self.loadedFileChanged.emit() self.sensorsChanged.emit() self._emit_current() @Slot() def play(self) -> None: self._play_timer.start(self._next_interval_ms()) @Slot() def pause(self) -> None: self._play_timer.stop() @Slot() def stop(self) -> None: self._play_timer.stop() self._player.seek(0) self._emit_current() @Slot(int) def step(self, delta: int) -> None: self._play_timer.stop() self._player.step(delta) self._emit_current() @Slot(int) def seek(self, i: int) -> None: self._player.seek(i) self._emit_current() @Slot(float) def setSpeed(self, x: float) -> None: self._speed = max(0.1, x) if self._play_timer.isActive(): self._play_timer.start(self._next_interval_ms()) def _next_interval_ms(self) -> int: """Timer interval in ms for the next frame, scaled by playback speed. Falls back to 500ms.""" ms = self._player.next_frame_ms() if ms <= 0: ms = 500.0 return max(1, int(ms / self._speed)) def _advance(self) -> None: if self._player.at_end: self._play_timer.stop() return self._player.step(1) self._emit_current() self._play_timer.start(self._next_interval_ms()) def _apply_pipeline(self, raw_values: list[float]) -> list[float]: """Apply override first, then optional cluster averaging.""" edited = self._sensor_editor.apply(raw_values) if getattr(self, "_cluster_averaging_enabled", False): return average_clusters(edited, getattr(self, "_active_clusters", [])) return edited def _emit_current(self) -> None: frame = self._player.current() if frame is None: return self._last_raw_frame = frame edited = Frame(seq=frame.seq, time=frame.time, values=self._apply_pipeline(frame.values)) self._last = self._model.process(edited) self.frameUpdated.emit() self.stateChanged.emit() def _reprocess_current(self) -> None: """Re-run the model on the last raw frame""" frame = self._last_raw_frame if frame is None: return edited = Frame(seq=frame.seq, time=frame.time, values=self._apply_pipeline(frame.values)) self._last = self._model.process(edited) self.frameUpdated.emit() self.stateChanged.emit() # ---- live: stream start/stop ---- @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 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) self._active_clusters = getattr(self._sensors, 'clusters', []) if not self._active_clusters: self._active_clusters = group_sensors_by_radius(self._sensors) # The new binary protocol sends payload of (sensorCount * 2) bytes # Each sensor is a big-endian 16-bit value (Sign-Magnitude). expected_sensors = 80 if (family_code or "") == "X" else 244 def parse_binary_frame(payload: bytes, seq: int) -> Frame: values = [] num_sensors = min(expected_sensors, 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) return Frame(seq=seq, time=time.monotonic(), values=values) # Clear out any old data from the prev sessions self._model.reset() self._last = None self._last_raw_frame = None transport = SerialPort.open_port(port, timeout=1) # Send 'D2' command padded to 512 bytes to start the stream cmd = b"D2" + SerialPort.COMMAND_PAD.encode() transport.write(cmd) def on_error(exc: Exception): log.error("Live stream error: %s", exc) self._reader = StreamReader( 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() self.setMode("live") self.stateChanged.emit() @Slot() def stopStream(self) -> None: self._repaint_timer.stop() 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) def _on_live_frame(self, frame: Frame) -> None: # Main thread (queued). Record raw, process edited; mark dirty for repaint. self._last_raw_frame = frame edited = Frame(seq=frame.seq, time=frame.time, values=self._apply_pipeline(frame.values)) self._last = self._model.process(edited) if self._recorder.is_recording: self._recorder.write(frame) # always record raw values, never edited self._dirty = True def _flush_repaint(self) -> None: if not self._dirty: return self._dirty = False self.frameUpdated.emit() self.stateChanged.emit() # ---- recording ---- @Slot(str, str) def startRecording(self, path: str, serial: str = "") -> None: self._recorder.start(path, self._sensors, serial) self.recordingChanged.emit() @Slot() def stopRecording(self) -> None: if self._recorder.is_recording: self._recorder.stop() self.recordingChanged.emit() @Slot(int, float) def replaceSensor(self, index: int, value: float) -> None: """Override sensor `index` to display `value` every frame.""" self._sensor_editor.set_replacement(index, value) self._reprocess_current() @Slot(int, float) def offsetSensor(self, index: int, delta: float) -> None: """Shift sensor `index` by `delta` every frame.""" self._sensor_editor.set_offset(index, delta) self._reprocess_current() @Slot(int) def clearSensorEdit(self, index: int) -> None: """Remove all overrides for sensor `index`.""" self._sensor_editor.clear(index) self._reprocess_current() @Slot() def clearSensorEdits(self) -> None: """Remove all sensor overrides.""" self._sensor_editor.clear() self._reprocess_current()