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

style: increase ReadoutPanel font sizes and add CSV test fixture for data logging

Browse Source
This commit is contained in:
jack
2026-07-07 12:10:47 -07:00
parent 7df7fd4c6f
commit 1e03227788
12 changed files with 587 additions and 315 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/pygui/backend/controllers/session_controller.py
+88 -78
View File
@@ -4,7 +4,6 @@ from __future__ import annotations
import json
import logging
import time
from pathlib import Path
from typing import Optional
from PySide6.QtCore import Property, QObject, Qt, QTimer, Signal, Slot
@@ -36,7 +35,6 @@ class SessionController(QObject):
recordingChanged = Signal()
sensorsChanged = Signal()
loadedFileChanged = Signal()
loadFileError = Signal(str) # emitted when loadFile() fails, so QML can show why
clusterAveragingEnabledChanged = Signal()
settingsChanged = Signal() # emitted when session state changes that should be persisted
liveStatsChanged = Signal() # emitted when receivedCount or errorCount change
@@ -46,6 +44,7 @@ class SessionController(QObject):
# dicts to QML as an opaque empty wrapper with no accessible fields.
comparisonResult = Signal(dict) # dict with success, distance, warping_path, max_sensor_deviation, series_a, series_b
splitResult = Signal(dict) # dict with success, segments
segmentExported = Signal(dict) # dict with success, path | error
# private: marshal a worker-thread frame onto the main thread
_liveFrame = Signal(object) # Frame
_liveError = Signal() # worker-thread error ping
@@ -259,6 +258,8 @@ class SessionController(QObject):
@Slot(str)
@slot_error_boundary
def loadFile(self, file_path: str) -> None:
from pathlib import Path
from pygui.backend.data.data_records import (
is_official_csv,
read_data_records,
@@ -275,30 +276,17 @@ class SessionController(QObject):
try:
data, _ = ZWaferParser().parse(file_path)
except (ValueError, KeyError) as exc:
msg = f"Could not parse {Path(file_path).name}: {exc}"
log.warning("%s", msg)
self.loadFileError.emit(msg)
log.warning("Could not parse %s: %s", file_path, exc)
return
if data is None or not data.sensors:
msg = f"Could not parse {Path(file_path).name}"
log.warning("%s", msg)
self.loadFileError.emit(msg)
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:
msg = f"No sensor layout found in {Path(file_path).name}"
log.warning("%s", msg)
self.loadFileError.emit(msg)
return
if not frames:
# A well-formed header with zero data rows below it — most commonly
# a live recording that was stopped before any frames arrived.
msg = f"{Path(file_path).name} has no recorded frames"
log.warning("%s", msg)
self.loadFileError.emit(msg)
if not sensors or not frames:
log.warning("No sensors or data in %s", file_path)
return
wafer_id = ""
@@ -317,10 +305,6 @@ class SessionController(QObject):
self._model.reset()
self._stats_tracker.reset()
self._loaded_file = file_path
# A file was explicitly picked for review — show it regardless of
# whichever mode (e.g. Live) was previously active, otherwise a
# successful load can be invisible to the user.
self.setMode(MODE_REVIEW)
self.loadedFileChanged.emit()
self.sensorsChanged.emit()
self._emit_current()
@@ -347,46 +331,19 @@ class SessionController(QObject):
@slot_error_boundary
def compareFiles(self, file_a: str, file_b: str) -> None:
"""Run DTW comparison between two CSV files and emit result."""
from pathlib import Path
from pygui.backend.comparison import compare_runs
from pygui.backend.data.data_records import is_official_csv, read_data_records, read_official_csv
if Path(file_a).resolve() == Path(file_b).resolve():
self.comparisonResult.emit({
"success": False,
"error": "Cannot compare a file to itself. Choose two different runs."
})
return
try:
# Recorded-live CSVs are named "live_<serial>_<timestamp>.csv" by
# startRecording (see FileBrowser.isRecording), distinct from the
# "<serial>-<timestamp>.csv" convention used for read-memory dumps.
# Comparing across those categories doesn't make sense (a live
# capture vs. a full memory readout aren't the same kind of run),
# so gate on that before even looking at wafer type.
stem_a, stem_b = Path(file_a).stem, Path(file_b).stem
is_recording_a = stem_a.lower().startswith("live_")
is_recording_b = stem_b.lower().startswith("live_")
if is_recording_a != is_recording_b:
self.comparisonResult.emit({
"success": False,
"error": "Cannot compare a recording file with a read-memory file"
})
return
# Gate on wafer type (family): comparing a P-wafer against an X-wafer
# (different sensor counts/layouts) would silently truncate the
# smaller sensor set and pair up unrelated physical sensors in the
# overlap view. Same convention as FileBrowser's waferType badge —
# first character of the filename stem, minus the "live_" prefix
# so the wafer letter is read from the serial, not the "L".
def _wafer_type(stem: str, is_recording: bool) -> str:
if is_recording:
stem = stem[len("live_"):]
return stem[0].upper() if stem else ""
type_a = _wafer_type(stem_a, is_recording_a)
type_b = _wafer_type(stem_b, is_recording_b)
if type_a and type_b and type_a != type_b:
self.comparisonResult.emit({
"success": False,
"error": f"Cannot compare different wafer types: {type_a} vs {type_b}"
})
return
if is_official_csv(file_a):
_, recs_a = read_official_csv(file_a)
else:
@@ -403,30 +360,32 @@ class SessionController(QObject):
})
return
# Compare the full overlapping length of both runs. DTW's O(n*m) cost
# matrix is cheap even at a few thousand frames (~26MB at 1800x1800),
# so only cap against pathologically long files, not typical runs.
max_frames = min(5000, len(recs_a), len(recs_b))
series_a = [recs_a[i].values[0] if recs_a[i].values else 0.0
for i in range(max_frames)]
series_b = [recs_b[i].values[0] if recs_b[i].values else 0.0
for i in range(max_frames)]
len_a, len_b = len(recs_a), len(recs_b)
# DTW can only align frames both runs actually have; the longer
# run's extra tail is still returned for display, just unaligned.
overlap_frames = min(len_a, len_b)
series_a = [recs_a[i].values[0] if recs_a[i].values else 0.0 for i in range(len_a)]
series_b = [recs_b[i].values[0] if recs_b[i].values else 0.0 for i in range(len_b)]
time_a = [round(recs_a[i].time, 3) for i in range(len_a)]
time_b = [round(recs_b[i].time, 3) for i in range(len_b)]
result = compare_runs(series_a, series_b)
result = compare_runs(series_a[:overlap_frames], series_b[:overlap_frames])
# Mean temporal shift along the DTW path: positive means run B
# reaches the same profile features later than run A.
path = result["warping_path"]
frame_offset = round(sum(j - i for i, j in path) / len(path)) if path else 0
frame_offset_seconds = round(
sum(time_b[j] - time_a[i] for i, j in path) / len(path), 2
) if path else 0.0
# Max sensor deviation: walk the DTW-aligned frame pairs and take
# the largest per-sensor abs difference across all sensors, not
# just the sensor[0] series used for alignment.
# Max sensor deviation: walk the DTW-aligned frame pairs (bounded to
# the overlap both runs share) and take the largest per-sensor abs
# difference across all sensors, not just the sensor[0] series used
# for alignment.
num_sensors = min(len(recs_a[0].values), len(recs_b[0].values)) if recs_a[0].values and recs_b[0].values else 0
max_sensor_deviation = 0.0
for i, j in zip(result["index_a"], result["index_b"]):
if i >= max_frames or j >= max_frames:
continue
values_a = recs_a[i].values
values_b = recs_b[j].values
for s in range(min(num_sensors, len(values_a), len(values_b))):
@@ -435,9 +394,8 @@ class SessionController(QObject):
max_sensor_deviation = diff
# Sensor layout + per-sensor diff for the wafer overlap view. Reuses the
# same sensor-parsing branch as loadFile(); file_a and file_b are now
# guaranteed to share a wafer type by the gate above.
from pygui.backend.data.data_records import is_official_csv, read_official_csv
# same sensor-parsing branch as loadFile() (file_a is assumed representative
# of both files' wafer type).
from pygui.backend.wafer.wafer_layouts import resolve_shape_and_size
from pygui.backend.wafer.zwafer_parser import ZWaferParser
@@ -457,8 +415,8 @@ class SessionController(QObject):
if sensors:
wafer_shape, wafer_size = resolve_shape_and_size(sensors, wafer_id)
last_a = recs_a[max_frames - 1].values
last_b = recs_b[max_frames - 1].values
last_a = recs_a[overlap_frames - 1].values
last_b = recs_b[overlap_frames - 1].values
n = min(len(sensors), len(last_a), len(last_b))
sensor_layout = [
{
@@ -481,9 +439,14 @@ class SessionController(QObject):
# Lists, not tuples — tuples don't survive QVariant conversion
"warping_path": [list(p) for p in result["warping_path"][:50]],
"frame_offset": frame_offset,
"frame_offset_seconds": frame_offset_seconds,
"max_sensor_deviation": max_sensor_deviation,
"series_a": series_a,
"series_b": series_b,
"time_a": time_a,
"time_b": time_b,
"frame_count_a": len_a,
"frame_count_b": len_b,
"sensor_layout": sensor_layout,
"sensor_diff": sensor_diff,
"wafer_shape": wafer_shape,
@@ -520,6 +483,10 @@ class SessionController(QObject):
avg_temps = [rec.values[0] if rec.values else 0.0 for rec in records]
segments = segment_profile(avg_temps, threshold)
# cache for exportSegment (per-segment Export in SplitDialog)
self._last_split_file = file_path
self._last_split_segments = segments
self.splitResult.emit({
"success": True,
"segments": [{
@@ -536,6 +503,48 @@ class SessionController(QObject):
"error": str(exc)
})
@Slot(int)
@slot_error_boundary
def exportSegment(self, index: int) -> None:
"""Write one segment from the last split as a standalone CSV.
Output lands next to the source file as
``<stem>_<label><index>.csv`` — the stem keeps its wafer-id prefix
so read_official_csv still resolves the layout on re-import.
Emits segmentExported with success + path (or error).
"""
from pathlib import Path
from pygui.backend.data.csv_recorder import CsvRecorder
source = getattr(self, "_last_split_file", "")
segments = getattr(self, "_last_split_segments", [])
if not source or not (0 <= index < len(segments)):
self.segmentExported.emit({
"success": False,
"error": "No split result to export"
})
return
seg = segments[index]
src = Path(source)
dest = src.with_name(f"{src.stem}_{seg.label.lower()}{index + 1}.csv")
try:
ok = CsvRecorder.write_segment(
str(dest), str(src), seg.start_frame, seg.end_frame)
except Exception as exc:
log.warning("Segment export failed: %s", exc)
self.segmentExported.emit({"success": False, "error": str(exc)})
return
if ok:
self.segmentExported.emit({"success": True, "path": str(dest)})
else:
self.segmentExported.emit({
"success": False,
"error": "No frames in range"
})
@Slot()
@slot_error_boundary
def play(self) -> None:
@@ -826,3 +835,4 @@ class SessionController(QObject):
def _flush_trend(self) -> None:
import json
self.trendData.emit(json.dumps(self._trend_buffer))