feat: Wafer Map integration
## Backend — Session & playback - Add SessionController (QML-facing live/review controller) with loadedFile property for active-file highlight binding - Add SessionModel, FramePlayer, and CsvRecorder for state machine, frame seek/step/speed, and CSV recording - Add SensorEditor for per-sensor value overrides (replace + offset)
This commit is contained in:
jack
@@ -0,0 +1,46 @@
|
||||
"""Append live frames"""
|
||||
from __future__ import annotations
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
from typing import Optional, TextIO
|
||||
|
||||
|
||||
from pygui.backend.frame import Frame
|
||||
from pygui.backend.zwafer_models import Sensor
|
||||
|
||||
class CsvRecorder:
|
||||
def __init__(self) -> None:
|
||||
self._file_handle: Optional[TextIO] = None
|
||||
self._oath: Optional[str] = None
|
||||
|
||||
|
||||
@property
|
||||
def is_recording(self) -> bool:
|
||||
return self._file_handle is not None
|
||||
|
||||
|
||||
def start(self, path: str, sensors:list[Sensor], serial: str = "") -> None:
|
||||
Path(path).parent.mkdir(parents=True, exist_ok=True)
|
||||
file_handle = open(path, "w", encoding="utf-8", newline ="")
|
||||
file_handle.write(f"Wafer ID={serial}\n")
|
||||
file_handle.write(f"Acquisition Date={datetime.now():%m/%d/%Y}\n")
|
||||
file_handle.write("Label," + ",".join(s.label for s in sensors) + "\n")
|
||||
file_handle.write("X (mm)," + ",".join(str(s.x) for s in sensors) + "\n")
|
||||
file_handle.write("Y (mm)," + ",".join(str(s.y) for s in sensors) + "\n")
|
||||
file_handle.write("data\n")
|
||||
file_handle.flush()
|
||||
self._file_handle, self._path = file_handle, path
|
||||
|
||||
def write(self, frame: Frame) -> None:
|
||||
if self._file_handle is None:
|
||||
return
|
||||
row = [f"{frame.time:.3f}"] + [f"{v:.3f}" for v in frame.values]
|
||||
self._file_handle.write(",".join(row) + "\n")
|
||||
self._file_handle.flush()
|
||||
|
||||
def stop(self) -> Optional[str]:
|
||||
if self._file_handle is None:
|
||||
return None
|
||||
self._file_handle.close()
|
||||
path, self._file_handle, self._path = self._path, None, None
|
||||
return path
|
||||
@@ -0,0 +1,66 @@
|
||||
"""Read the data section of a Z-wafer CSV into DataRecords, or the official headerless CSV."""
|
||||
from __future__ import annotations
|
||||
from pathlib import Path
|
||||
|
||||
from pygui.backend.zwafer_models import DataRecord, Sensor
|
||||
|
||||
def read_data_records(file_path: str) -> list[DataRecord]:
|
||||
records: list[DataRecord] = []
|
||||
in_data = False
|
||||
with Path(file_path).open("r", encoding="utf-8") as file_handle:
|
||||
for line in file_handle:
|
||||
line = line.rstrip().rstrip(",").strip()
|
||||
if not line:
|
||||
continue
|
||||
if not in_data:
|
||||
if line.split(",")[0].lower() == "data":
|
||||
in_data = True
|
||||
continue
|
||||
cols = [c.strip() for c in line.split(",") if c.strip() != ""]
|
||||
try:
|
||||
nums = [float(c) for c in cols]
|
||||
except ValueError:
|
||||
continue
|
||||
if not nums:
|
||||
continue
|
||||
records.append(DataRecord(time=nums[0], values=nums[1:]))
|
||||
return records
|
||||
|
||||
|
||||
def is_official_csv(file_path: str) -> bool:
|
||||
"""Return True if the file uses the headerless official format (sensor-name header row, no 'data' sentinel)."""
|
||||
with Path(file_path).open("r", encoding="utf-8") as f:
|
||||
first = f.readline().rstrip().rstrip(",")
|
||||
if not first:
|
||||
return False
|
||||
parts = [p.strip() for p in first.split(",") if p.strip()]
|
||||
# Official format: first cell looks like "SensorN" (starts with letter, not key=value)
|
||||
return bool(parts) and parts[0].lower().startswith("sensor") and "=" not in parts[0]
|
||||
|
||||
|
||||
def read_official_csv(file_path: str) -> tuple[list[Sensor], list[DataRecord]]:
|
||||
"""Parse the headerless official CSV: row 0 = sensor names, rows 1+ = values (no timestamp)."""
|
||||
from pygui.backend.wafer_layouts import load_layout_for_wafer_id
|
||||
|
||||
stem = Path(file_path).stem
|
||||
wafer_id = stem.split("-")[0] if "-" in stem else stem
|
||||
try:
|
||||
sensors = load_layout_for_wafer_id(wafer_id)
|
||||
except KeyError:
|
||||
return [], []
|
||||
|
||||
records: list[DataRecord] = []
|
||||
with Path(file_path).open("r", encoding="utf-8") as f:
|
||||
f.readline() # skip sensor-name header
|
||||
for i, line in enumerate(f):
|
||||
line = line.rstrip().rstrip(",")
|
||||
if not line:
|
||||
continue
|
||||
cols = [c.strip() for c in line.split(",") if c.strip()]
|
||||
try:
|
||||
values = [float(c) for c in cols]
|
||||
except ValueError:
|
||||
continue
|
||||
if values:
|
||||
records.append(DataRecord(time=float(i) * 0.5, values=values))
|
||||
return sensors, records
|
||||
@@ -120,24 +120,37 @@ class FileBrowser(QObject):
|
||||
for csv_path in sorted(self._current_directory.glob("*.csv")):
|
||||
try:
|
||||
metadata = self._load_metadata(csv_path)
|
||||
wafer_type = metadata.get_wafer_type().upper() or (csv_path.stem[0].upper() if csv_path.stem else "")
|
||||
date_str = metadata.string_date_format()
|
||||
serial = metadata.wafer[1:] if len(metadata.wafer) > 1 else ""
|
||||
time_str = date_str[11:] if len(date_str) > 10 else ""
|
||||
self._files.append(
|
||||
{
|
||||
"selected": selected_state.get(str(csv_path), False),
|
||||
"baseName": csv_path.stem,
|
||||
"wafer": metadata.wafer,
|
||||
"date": metadata.string_date_format(),
|
||||
"waferType": wafer_type,
|
||||
"serialNumber": serial,
|
||||
"date": date_str,
|
||||
"timeStr": time_str,
|
||||
"chamber": metadata.chamber,
|
||||
"notes": metadata.notes,
|
||||
"masterType": master_state.get(str(csv_path), "") or metadata.master_type,
|
||||
"fileName": str(csv_path),
|
||||
"highlight": metadata.get_wafer_type() in {"A", "B", "C"},
|
||||
"highlight": wafer_type in {"A", "B", "C"},
|
||||
}
|
||||
)
|
||||
except Exception:
|
||||
stem = csv_path.stem
|
||||
self._files.append(
|
||||
{
|
||||
"selected": selected_state.get(str(csv_path), False),
|
||||
"wafer": csv_path.stem,
|
||||
"baseName": stem,
|
||||
"wafer": stem,
|
||||
"waferType": stem[0].upper() if stem else "",
|
||||
"serialNumber": stem[1:] if len(stem) > 1 else "",
|
||||
"date": "",
|
||||
"timeStr": "",
|
||||
"chamber": "",
|
||||
"notes": "Unable to parse metadata",
|
||||
"masterType": master_state.get(str(csv_path), ""),
|
||||
@@ -186,7 +199,7 @@ class FileBrowser(QObject):
|
||||
|
||||
def _metadata_from_filename(self, csv_path: Path) -> CSVFileMetadata:
|
||||
match = re.match(
|
||||
r"^(?P<wafer>[^-]+)-(?P<date>\d{8})(?:_(?P<run>\d+))?$",
|
||||
r"^(?P<wafer>[^-]+)-(?P<date>\d{8})(?:_(?P<time>\d{6}))?$",
|
||||
csv_path.stem,
|
||||
)
|
||||
if not match:
|
||||
@@ -194,7 +207,9 @@ class FileBrowser(QObject):
|
||||
|
||||
date_text = ""
|
||||
try:
|
||||
parsed = datetime.strptime(match.group("date"), "%Y%m%d")
|
||||
date_part = match.group("date")
|
||||
time_part = match.group("time") or "000000"
|
||||
parsed = datetime.strptime(date_part + time_part, "%Y%m%d%H%M%S")
|
||||
date_text = CSVFileMetadata.format_date(parsed)
|
||||
except ValueError:
|
||||
date_text = ""
|
||||
@@ -236,7 +251,7 @@ class FileBrowser(QObject):
|
||||
def showInfo(self, message: str) -> None:
|
||||
self._show_info(message)
|
||||
|
||||
@Slot(str, result="QVariantMap")
|
||||
@Slot(str, result=dict)
|
||||
def parseCsvMetadata(self, file_path: str) -> dict:
|
||||
from pathlib import Path as _Path
|
||||
|
||||
|
||||
@@ -7,5 +7,5 @@ class Frame:
|
||||
"""One sample across all sensors ata a point in time"""
|
||||
|
||||
seq: int # monotonically increasing
|
||||
t: float # seconds (relative or epoch)
|
||||
time: float # seconds (relative or epoch)
|
||||
values: list[float] # one per sensor, in sensor-layout order
|
||||
|
||||
@@ -0,0 +1,51 @@
|
||||
from __future__ import annotations
|
||||
from pygui.backend.frame import Frame
|
||||
from pygui.backend.zwafer_models import ZWaferData
|
||||
|
||||
def frames_from_wafer_data(data: ZWaferData, records) -> list[Frame]:
|
||||
"""Build Frames from parsed DataRecords (records = list[DataRecord].)"""
|
||||
return [Frame(seq=i, time=r.time, values=list(r.values)) for i, r in enumerate(records)]
|
||||
|
||||
class FramePlayer:
|
||||
def __init__(self) -> None:
|
||||
self._frames: list[Frame] =[]
|
||||
self._i =0
|
||||
|
||||
def load(self, frames: list[Frame]) -> "FramePlayer":
|
||||
self._frames = frames
|
||||
self._i = 0
|
||||
return self
|
||||
|
||||
@property
|
||||
def total(self) -> int:
|
||||
return len(self._frames)
|
||||
|
||||
@property
|
||||
def index(self) -> int:
|
||||
return self._i
|
||||
|
||||
@property
|
||||
def at_end(self) -> bool:
|
||||
return self._i >= len(self._frames) -1
|
||||
|
||||
def current(self) -> Frame | None:
|
||||
if not self._frames:
|
||||
return None
|
||||
return self._frames[self._i]
|
||||
|
||||
def _clamp(self, i:int):
|
||||
return max(0, min(i, len(self._frames) - 1)) if self._frames else 0
|
||||
|
||||
def step(self, delta: int) -> "FramePlayer":
|
||||
self._i = self._clamp(self._i + delta)
|
||||
return self
|
||||
|
||||
def seek(self, i: int) -> "FramePlayer":
|
||||
self._i = self._clamp(i)
|
||||
return self
|
||||
|
||||
def next_frame_ms(self) -> float:
|
||||
"""Wall-clock ms until the next frame based on recorded timestamps; 0 if at end."""
|
||||
if self._i + 1 >= len(self._frames):
|
||||
return 0.0
|
||||
return (self._frames[self._i + 1].time - self._frames[self._i].time) * 1000.0
|
||||
@@ -0,0 +1,48 @@
|
||||
"""Per-session sensor value overrides (replacement and/or offset)."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
|
||||
class SensorEditor:
|
||||
def __init__(self) -> None:
|
||||
self._replacements: dict[int, float] = {}
|
||||
self._offsets: dict[int, float] = {}
|
||||
|
||||
|
||||
def set_replacement(self, index: int, value: float) -> None:
|
||||
"""Force sensor `index` to read `value` for every frame this session."""
|
||||
self._replacements[index] = value
|
||||
|
||||
def set_offset(self, index: int, value: float) -> None:
|
||||
"""Shift sensor `index` by `delta` (applied after any replacement)."""
|
||||
self._offsets[index] = value
|
||||
|
||||
def clear(self, index: int | None = None) -> None:
|
||||
"""Remove overrides. Pass an index to clear one sensor; omit to clear all."""
|
||||
if index is None:
|
||||
self._replacements.clear()
|
||||
self._offsets.clear()
|
||||
else:
|
||||
self._replacements.pop(index, None)
|
||||
self._offsets.pop(index, None)
|
||||
def has_overrides(self) -> bool:
|
||||
return bool(self._replacements or self._offsets)
|
||||
|
||||
def active_indices(self) -> list[int]:
|
||||
"""Return sorted sensor indices that have any override."""
|
||||
return sorted(set(self._replacements) | set(self._offsets))
|
||||
|
||||
|
||||
def apply(self, values: list[float]) -> list[float]:
|
||||
"""Return a new list wit all overrides applied (original unchanged)"""
|
||||
if not self.has_overrides():
|
||||
return list(values)
|
||||
result = list(values)
|
||||
for i in range(len(result)):
|
||||
v = result[i]
|
||||
if i in self._replacements:
|
||||
v = self._replacements[i]
|
||||
if i in self._offsets:
|
||||
v += self._offsets[i]
|
||||
result[i] = v
|
||||
return result
|
||||
@@ -0,0 +1,338 @@
|
||||
"""QML-facing controller for the live/review wafer dashboard."""
|
||||
from __future__ import annotations
|
||||
import logging
|
||||
from typing import Any, Optional
|
||||
|
||||
from PySide6.QtCore import QObject, Property, QTimer, Qt, Signal, Slot
|
||||
|
||||
from pygui.backend.csv_recorder import CsvRecorder
|
||||
from pygui.backend.frame import Frame
|
||||
from pygui.backend.frame_player import FramePlayer, frames_from_wafer_data
|
||||
from pygui.backend.session_model import SessionModel
|
||||
from pygui.backend.threshold_classifier import ThresholdConfig
|
||||
from pygui.backend.zwafer_models import Sensor
|
||||
from pygui.backend.zwafer_parser import ZWaferParser
|
||||
from pygui.serialcomm.stream_reader import StreamReader
|
||||
from pygui.backend.sensor_editor import SensorEditor
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
MODE_LIVE = "live"
|
||||
MODE_REVIEW = "review"
|
||||
|
||||
|
||||
class SessionController(QObject):
|
||||
# public signals
|
||||
frameUpdated = Signal() # any property below may have changed
|
||||
modeChanged = Signal()
|
||||
stateChanged = Signal()
|
||||
recordingChanged = Signal()
|
||||
sensorsChanged = Signal()
|
||||
loadedFileChanged = 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 = ""
|
||||
|
||||
# ---- 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: return self._last.state if self._last else "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
|
||||
|
||||
# ---- 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]
|
||||
|
||||
@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()
|
||||
|
||||
# ---- 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 pygui.backend.data_records import is_official_csv, read_data_records, read_official_csv
|
||||
|
||||
sensors: list[Sensor] = []
|
||||
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
|
||||
|
||||
self._sensors = 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 _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._sensor_editor.apply(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._sensor_editor.apply(frame.values))
|
||||
self._last = self._model.process(edited)
|
||||
self.frameUpdated.emit()
|
||||
self.stateChanged.emit()
|
||||
|
||||
# ---- live: stream start/stop ----
|
||||
@Slot(str)
|
||||
def startStream(self, port: str) -> None:
|
||||
from pygui.serialcomm.serial_port import SerialPort # transport open
|
||||
import serial as pyserial
|
||||
|
||||
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:])
|
||||
|
||||
transport = pyserial.Serial(port, SerialPort.BAUDRATE, timeout=1)
|
||||
self._reader = StreamReader(
|
||||
transport, parse_line,
|
||||
on_frame=lambda f: self._liveFrame.emit(f))
|
||||
self._reader.start()
|
||||
self._repaint_timer.start() # Q1: begin ~20 Hz repaints
|
||||
|
||||
@Slot()
|
||||
def stopStream(self) -> None:
|
||||
self._repaint_timer.stop()
|
||||
if self._reader is not None:
|
||||
self._reader.stop()
|
||||
self._reader = None
|
||||
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._sensor_editor.apply(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()
|
||||
|
||||
|
||||
|
||||
@@ -0,0 +1,39 @@
|
||||
from __future__ import annotations
|
||||
from dataclasses import dataclass
|
||||
|
||||
from pygui.backend.frame import Frame
|
||||
from pygui.backend.frame_stats import Stats, compute_stats
|
||||
from pygui.backend.stability_detector import StabilityDetector
|
||||
from pygui.backend.threshold_classifier import ThresholdConfig, classify, resolve_bounds
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class SessionUpdate:
|
||||
seq: int
|
||||
values: list[float]
|
||||
bands: list[str]
|
||||
stats: Stats
|
||||
state: str
|
||||
target: float
|
||||
margin: float
|
||||
|
||||
class SessionModel:
|
||||
def __init__(self, thresholds: ThresholdConfig | None = None) -> None:
|
||||
self._config = thresholds or ThresholdConfig()
|
||||
self._stability = StabilityDetector()
|
||||
|
||||
def set_thresholds(self, config: ThresholdConfig) -> None:
|
||||
self._config = config
|
||||
|
||||
def reset(self) -> None:
|
||||
self._stability.reset()
|
||||
|
||||
def process(self, frame: Frame) -> SessionUpdate:
|
||||
stats = compute_stats(frame.values)
|
||||
target, margin = resolve_bounds(frame.values, self._config)
|
||||
bands = [classify(v, target, margin)for v in frame.values]
|
||||
# State always track the user's process set_point, even in auto-color mode.
|
||||
state = self._stability.update(stats.avg, frame.time, self._config.set_point)
|
||||
return SessionUpdate(
|
||||
seq=frame.seq, values=frame.values, bands=bands,
|
||||
stats=stats, state=state, target=target, margin=margin,
|
||||
)
|
||||
@@ -32,7 +32,14 @@ class ZWaferParser:
|
||||
|
||||
# ===== Header Parsing =====
|
||||
def _process_header(self, file_obj) -> ZWaferData:
|
||||
"""Parse header and sensor layout from open file object."""
|
||||
"""Parse header and sensor layout from open file object.
|
||||
|
||||
Handles two CSV row orderings:
|
||||
1. Label/X/Y rows BEFORE 'data' (fixture test format)
|
||||
2. 'data' row BEFORE Label/X/Y (some real CSV files)
|
||||
The C# parser handles both by collecting sensor layout rows
|
||||
regardless of whether 'data' has been seen.
|
||||
"""
|
||||
wafer_data = ZWaferData()
|
||||
labels: Optional[list] = None
|
||||
x_coords: Optional[list] = None
|
||||
@@ -47,26 +54,25 @@ class ZWaferParser:
|
||||
parts = [p.strip() for p in line.split(",")]
|
||||
first_part = parts[0].lower()
|
||||
|
||||
# Collect sensor layout rows (Label, X (mm), Y (mm))
|
||||
# regardless of whether 'data' has been seen.
|
||||
# This matches the C# parser behavior.
|
||||
if first_part == "label":
|
||||
labels = parts[1:]
|
||||
elif first_part == "x (mm)":
|
||||
x_coords = parts[1:]
|
||||
elif first_part == "y (mm)":
|
||||
y_coords = parts[1:]
|
||||
# Parse metadata (key=value pairs) — only when not a sensor layout row
|
||||
elif first_part != "data":
|
||||
self._parse_header_line(wafer_data, parts)
|
||||
|
||||
# Detect end of header section
|
||||
if first_part == "data":
|
||||
wafer_data.csv_headers = labels or []
|
||||
self._build_sensor_layout(wafer_data, labels, x_coords, y_coords)
|
||||
return wafer_data
|
||||
|
||||
# Detect switch from metadata to sensor layout
|
||||
# Parse metadata (key=value pairs)
|
||||
if not self._parse_header_line(wafer_data, parts):
|
||||
# If not metadata, it's part of sensor layout
|
||||
label = parts[0].lower()
|
||||
values = parts[1:]
|
||||
|
||||
if label == "label":
|
||||
labels = values
|
||||
elif label == "x (mm)":
|
||||
x_coords = values
|
||||
elif label == "y (mm)":
|
||||
y_coords = values
|
||||
|
||||
return wafer_data # Incomplete file — return partial data
|
||||
|
||||
# ===== Metadata Parsing =====
|
||||
|
||||
Reference in New Issue
Block a user