diff --git a/README.md b/README.md index fef01a6..3a3683d 100755 --- a/README.md +++ b/README.md @@ -14,6 +14,9 @@ scanengine-3 is a unified platform for scanning acoustic microscopy and precisio - **Scan Planning**: Automated raster scan generation and execution - **Angle Inspection**: Park the rig at random points across a plan's angles to check the SAW response on the scope before committing to a long scan +- **SAW Quality Check**: Acquire one row per angle — the row-wise middle of + the ROI — as a v10 `.sras`, then compare every angle's SAW frequency on one + graph to judge the alignment before a full run - **Real-time Monitoring**: Live status updates and progress tracking ## Hardware Components @@ -60,8 +63,9 @@ scanengine-3/ │ ├── scope_burst.py # Burst-mode FastFrame sizing + row splitting │ ├── scope_inspect.py # Scope setup for pre-scan angle inspection │ ├── angle_inspect.py # AngleInspector — park on a point per angle +│ ├── saw_check.py # Middle-row SAW check: plan + alignment read-out │ ├── rotation.py # GR rotation axis settings + moves -│ ├── sras_format.py # v6 .sras writer/reader (memory-mapped) +│ ├── sras_format.py # v6/v10 .sras writer/reader (memory-mapped) │ ├── sras_analysis.py # Image reducers + SAW matched filter │ └── config.py # ScanDefaults ⇄ aui_defaults.json │ @@ -84,6 +88,7 @@ scanengine-3/ │ ├── sc3_aui_app.py # Main acquisition application ├── sras_viewer.py # Scan data viewer +├── saw_check_viewer.py # SAW check viewer: every angle's frequency, one graph ├── sras_scan_manager.py # CLI: inspect/export/delete angles ├── t3r_control_panel.py # T3R panel (used by the main app) ├── helios_test_app.py # Per-device test benches @@ -133,6 +138,9 @@ python sc3_aui_app.py # Scan data viewer python sras_viewer.py +# SAW quality check viewer (every angle's frequency on one graph) +python saw_check_viewer.py path/to/scan-sawcheck.sras + # Inspect / export / delete angles in a .sras file python sras_scan_manager.py path/to/scan.sras @@ -208,6 +216,33 @@ result = engine.run() # blocking; engine.abort() is thread-safe print(f"wrote {result.rows_written} rows to {result.path}") ``` +### Running a SAW quality check + +Same engine, same hardware sequence — the plan is reduced to one row per +angle and the result is tagged v10 so the viewer knows it is a check rather +than a scan cut short: + +```python +from core.saw_check import alignment_summary, frequency_traces, middle_row_plan +from core.sras_format import VERSION_SAW_CHECK, SrasFile + +check = middle_row_plan(plan) # the plan above: 163 rows → 3 +engine = ScanEngine(stage, scope, RotationAxis(t3r), check, + Path("/data/SRAS/demo-sawcheck.sras"), + callbacks=ScanCallbacks(on_status=print), + file_version=VERSION_SAW_CHECK) +engine.run() + +with SrasFile("/data/SRAS/demo-sawcheck.sras") as sras: + traces = frequency_traces(sras, dc_threshold_mv=50.0) + for t in traces: + print(f"{t.angle_deg:+7.1f}° {t.median_mhz:.2f} MHz " + f"drift {t.drift_mhz_per_mm:+.3f} MHz/mm") + print(alignment_summary(traces).describe()) +``` + +`saw_check_viewer.py` is the same read-out with the curves drawn. + ### Reading a scan file `SrasFile` memory-maps the data block, so opening a multi-gigabyte scan diff --git a/core/saw_check.py b/core/saw_check.py new file mode 100644 index 0000000..a3e93f5 --- /dev/null +++ b/core/saw_check.py @@ -0,0 +1,255 @@ +"""Middle-row SAW quality check: acquire one row per angle, then read the +alignment off the frequencies it produces. + +Two halves of one test mode, kept together because neither is much use +without the other: + +*Acquisition* — ``middle_row_plan`` reduces a full ScanPlan to a single row +per angle, the row-wise middle of the ROI. ScanEngine runs the result +exactly like any other scan and writes it as a v10 .sras file +(``sras_format.VERSION_SAW_CHECK``), so a check costs one row-time per angle +instead of the hours a full multi-angle scan takes. + +*Analysis* — ``frequency_traces`` turns such a file back into one peak-SAW- +frequency trace per angle, and ``alignment_summary`` reduces those to the +numbers the operator is actually asking about. Both are Qt-free; the plotting +lives in saw_check_viewer.py. + +Why the middle row answers an alignment question: ``scan_geometry.build_plan`` +centres every angle's rotated bounding box on the same nominal ROI centre, so +each angle's middle row crosses that one point on the sample. Every angle +therefore measures the same material, and a spread in the per-angle +frequencies is a property of the rig (or of a genuinely anisotropic sample), +not of where each row happened to land. +""" +from __future__ import annotations + +from dataclasses import dataclass, field, replace + +import numpy as np + +from core.scan_geometry import ScanGeometryError, ScanPlan +from core.sras_analysis import ChannelCalibration, compute_rf_image +from core.sras_format import SrasFile + +# Rules of thumb for the read-out, not physics. A well-aligned rig on an +# isotropic sample reads the same frequency at every angle, so the spread of +# the per-angle medians is the alignment signal — but an anisotropic sample +# genuinely varies with angle, so a wide spread is a prompt to look at the +# curves, never a verdict on its own. +SPREAD_GOOD_PCT = 1.0 +SPREAD_MARGINAL_PCT = 3.0 +# Below this fraction of unmasked pixels a trace is too sparse to read. +VALID_FRACTION_FLOOR = 0.5 + + +# ── Acquisition side ───────────────────────────────────────────────────────── + +def middle_row_plan(plan: ScanPlan) -> ScanPlan: + """Reduce a scan plan to its row-wise middle row at every angle. + + Each angle keeps the geometry the full scan would have used — same + x_start, x_delta and n_frames from its own rotated bounding box — and + scans only the middle entry of its row list, so the check samples exactly + what the scan would along that row. + + An even row count has no exact middle; the upper of the two central rows + is taken (``n_rows // 2``), which is also the row the viewer picks when it + reads the middle row out of a full v6 scan. + """ + if plan.n_angles == 0: + raise ScanGeometryError("Cannot build a SAW check from a plan with no angles") + + per_angle = [] + for pa in plan.per_angle: + if not pa.y_positions: + raise ScanGeometryError( + f"Angle {pa.angle_deg:.1f}° has no rows, so it has no middle row to check" + ) + per_angle.append(replace(pa, n_rows=1, + y_positions=[pa.y_positions[middle_row_index(pa.n_rows)]])) + return replace(plan, per_angle=per_angle) + + +def middle_row_index(n_rows: int) -> int: + """The row this check calls the middle one. One rule, two callers.""" + return max(0, n_rows // 2) + + +# ── Analysis side ──────────────────────────────────────────────────────────── + +@dataclass +class AngleTrace: + """One angle's peak SAW frequency along its middle row. + + ``freq_mhz`` is NaN wherever the pixel was masked out (CH4 DC below the + threshold), so the gaps stay gaps instead of reading as 0 MHz. + """ + angle_idx: int + angle_deg: float + row_idx: int + y_mm: float + x_mm: np.ndarray # absolute stage X of each frame + freq_mhz: np.ndarray # NaN where masked + _valid: np.ndarray = field(init=False, repr=False) + + def __post_init__(self): + self._valid = np.isfinite(self.freq_mhz) + + @property + def offset_mm(self) -> np.ndarray: + """X relative to the centre of this row. + + Every angle's row is centred on the same ROI centre, so plotting + against this puts all the angles' curves over the same piece of + sample — which is the whole point of the comparison. + """ + if len(self.x_mm) == 0: + return self.x_mm + return self.x_mm - 0.5 * (self.x_mm[0] + self.x_mm[-1]) + + @property + def n_valid(self) -> int: + return int(self._valid.sum()) + + @property + def valid_fraction(self) -> float: + return self.n_valid / len(self.freq_mhz) if len(self.freq_mhz) else 0.0 + + @property + def median_mhz(self) -> float: + return float(np.median(self.freq_mhz[self._valid])) if self.n_valid else float("nan") + + @property + def std_mhz(self) -> float: + return float(np.std(self.freq_mhz[self._valid])) if self.n_valid > 1 else float("nan") + + @property + def drift_mhz_per_mm(self) -> float: + """Least-squares slope of frequency along the row. + + A flat trace means the response did not change across the ROI; a + sloped one is the signature of a tilt or a defocus the angle spread + alone would not show. + """ + if self.n_valid < 2: + return float("nan") + x = self.offset_mm[self._valid] + if np.ptp(x) == 0: + return float("nan") + return float(np.polyfit(x, self.freq_mhz[self._valid], 1)[0]) + + +@dataclass +class AlignmentSummary: + """What the per-angle traces say about the alignment, in scalars.""" + n_angles: int + median_mhz: float + spread_mhz: float # max − min of the per-angle medians + spread_pct: float # that spread as a % of the overall median + best_angle_deg: float # angle reading the highest median + worst_angle_deg: float # angle reading the lowest median + worst_drift_mhz_per_mm: float + worst_drift_angle_deg: float + min_valid_fraction: float + + @property + def level(self) -> str: + """"good" / "marginal" / "poor" — see the module's threshold note.""" + if self.n_angles == 0 or not np.isfinite(self.spread_pct): + return "poor" + if self.min_valid_fraction < VALID_FRACTION_FLOOR: + return "poor" + if self.spread_pct <= SPREAD_GOOD_PCT: + return "good" + if self.spread_pct <= SPREAD_MARGINAL_PCT: + return "marginal" + return "poor" + + def describe(self) -> str: + if self.n_angles == 0: + return "No angle produced a usable frequency trace." + if self.min_valid_fraction < VALID_FRACTION_FLOOR: + return (f"Only {self.min_valid_fraction * 100:.0f} % of the worst angle's row " + f"is above the DC threshold — check the detection beam and the " + f"threshold before reading the spread.") + return (f"Per-angle medians span {self.spread_mhz:.3f} MHz " + f"({self.spread_pct:.2f} % of {self.median_mhz:.3f} MHz), " + f"lowest at {self.worst_angle_deg:.1f}°, highest at {self.best_angle_deg:.1f}°. " + f"Largest drift along a row: {self.worst_drift_mhz_per_mm:+.3f} MHz/mm " + f"at {self.worst_drift_angle_deg:.1f}°.") + + +def frequency_traces(sras: SrasFile, *, dc_threshold_mv: float = 0.0, + background: np.ndarray | None = None, + gate_start_ns: float | None = None, + gate_end_ns: float | None = None, + calib: ChannelCalibration | None = None, + on_progress=lambda done, total: None) -> list[AngleTrace]: + """Peak SAW frequency along the middle row of every angle in ``sras``. + + Works on a v10 check (one row per angle, so the middle row is the only + row) and on a full v6 scan alike — the same middle row the check would + have acquired is pulled out of the scan, which is what lets a finished + scan be re-examined with the check's own read-out. + + Angles with nothing on disk (an aborted file) are skipped rather than + reported as flat zero. + """ + calib = calib if calib is not None else ChannelCalibration.from_preambles(sras.preambles) + freq_axis = sras.freq_axis_mhz(sras.header.samples_per_frame) + time_axis = sras.time_axis_ns() + statuses = sras.angle_status() + + traces: list[AngleTrace] = [] + for st in statuses: + on_progress(st.index, len(statuses)) + if st.n_rows_available < 1: + continue + pa = sras.per_angle[st.index] + row = middle_row_index(st.n_rows_available) + view = sras.load_angle(st.index, n_rows=st.n_rows_available)[row:row + 1] + + img = compute_rf_image(view, calib, freq_axis, dc_threshold_mv, + background=background, + gate_start_ns=gate_start_ns, gate_end_ns=gate_end_ns, + time_axis_ns=time_axis) + # compute_rf_image zeroes masked pixels and its FFT never peaks in the + # suppressed DC bin, so 0 MHz means "no reading" and nothing else. + freq = img[0].astype(np.float64) + freq[freq <= 0.0] = np.nan + + traces.append(AngleTrace( + angle_idx=st.index, angle_deg=pa.angle_deg, row_idx=row, + y_mm=pa.y_positions[row] if row < len(pa.y_positions) else float("nan"), + x_mm=sras.x_axis_mm(st.index), freq_mhz=freq, + )) + on_progress(len(statuses), len(statuses)) + return traces + + +def alignment_summary(traces: list[AngleTrace]) -> AlignmentSummary: + """Reduce per-angle traces to the alignment read-out.""" + usable = [t for t in traces if t.n_valid > 0] + if not usable: + nan = float("nan") + return AlignmentSummary(0, nan, nan, nan, nan, nan, nan, nan, 0.0) + + medians = np.array([t.median_mhz for t in usable]) + overall = float(np.median(medians)) + spread = float(medians.max() - medians.min()) + drifts = [(abs(t.drift_mhz_per_mm), t) for t in usable + if np.isfinite(t.drift_mhz_per_mm)] + worst_drift = max(drifts, key=lambda d: d[0])[1] if drifts else None + + return AlignmentSummary( + n_angles=len(usable), + median_mhz=overall, + spread_mhz=spread, + spread_pct=spread / overall * 100.0 if overall else float("nan"), + best_angle_deg=usable[int(np.argmax(medians))].angle_deg, + worst_angle_deg=usable[int(np.argmin(medians))].angle_deg, + worst_drift_mhz_per_mm=worst_drift.drift_mhz_per_mm if worst_drift else float("nan"), + worst_drift_angle_deg=worst_drift.angle_deg if worst_drift else float("nan"), + min_valid_fraction=min(t.valid_fraction for t in usable), + ) diff --git a/core/scan_engine.py b/core/scan_engine.py index aac953e..715baf4 100644 --- a/core/scan_engine.py +++ b/core/scan_engine.py @@ -18,7 +18,7 @@ from typing import Callable from core import scope_burst, scope_sras from core.rotation import RotationAxis from core.scan_geometry import ScanPlan, validate_plan -from core.sras_format import SCAN_CHANNELS, create_scan_file +from core.sras_format import SCAN_CHANNELS, VERSION, create_scan_file logger = logging.getLogger(__name__) @@ -96,7 +96,8 @@ class ScanEngine: plan: ScanPlan, out_path: Path, resume: ResumeState | None = None, callbacks: ScanCallbacks | None = None, - burst_mode: bool = False, strict_rows: bool = False): + burst_mode: bool = False, strict_rows: bool = False, + file_version: int = VERSION): self._stage = stage self._scope = scope self._rotator = rotator @@ -110,6 +111,11 @@ class ScanEngine: # Strict row packing stops the scan on a frame-count mismatch # instead of squaring the row up (see _check_frame_delta). self._strict_rows = strict_rows + # Which kind of file this run produces. The acquisition is identical + # either way; VERSION_SAW_CHECK only marks a one-row-per-angle plan + # (core.saw_check) as the quality check it is, so a reader does not + # mistake it for a scan that aborted after its first row. + self._file_version = file_version self._max_frames = 0 self._preflight_done = False @@ -336,6 +342,7 @@ class ScanEngine: return create_scan_file( self._out_path, self._plan, samples_per_frame, scope_sras.SAMPLE_RATE_HZ, self._preambles, self._background, + version=self._file_version, ) def _scan_loop(self, scan_file, samples_per_frame: int, result: ScanResult): diff --git a/core/sras_format.py b/core/sras_format.py index 5dfbe7e..8977c68 100644 --- a/core/sras_format.py +++ b/core/sras_format.py @@ -1,4 +1,4 @@ -"""SRAS v6 binary scan-file format — the single implementation. +"""SRAS binary scan-file format (v6 and v10) — the single implementation. Full byte-level spec: scan_format.md. Summary: @@ -17,6 +17,12 @@ Full byte-level spec: scan_format.md. Summary: Incomplete files are valid: the data block is one contiguous append-only stream, so the readable prefix defines a single frontier past which nothing has been written yet (see ``SrasFile.angle_status``). + +Version 10 is the SAW quality check (core.saw_check): byte layout identical +to v6, but every angle declares exactly one row — the row-wise middle of the +ROI. The version byte is the whole difference, and it exists so a reader can +tell a one-row-per-angle check from a full scan that was aborted after its +first row. ``create_scan_file`` enforces the one-row rule at write time. """ from __future__ import annotations @@ -32,6 +38,9 @@ from core.scan_geometry import AngleGeometry, ScanPlan MAGIC = b"SRAS" VERSION = 6 +# One row per angle, taken from the middle of the ROI — see core.saw_check. +VERSION_SAW_CHECK = 10 +SUPPORTED_VERSIONS = (VERSION, VERSION_SAW_CHECK) HDR_FMT = ">4sBHfffffffIdBB" HDR_SIZE = struct.calcsize(HDR_FMT) # 49 bytes GEOM_FMT = ">ffIH" @@ -80,16 +89,38 @@ class AngleStatus: def create_scan_file(path: Path, plan: ScanPlan, samples_per_frame: int, sample_rate: float, preambles: list[str], - background_waveform: bytes) -> BinaryIO: - """Create a new .sras file and write the v6 header + tables. + background_waveform: bytes, + version: int = VERSION) -> BinaryIO: + """Create a new .sras file and write the header + tables. + + ``version`` selects which kind of file this is — VERSION for a full scan, + VERSION_SAW_CHECK for a middle-row quality check. The layout is the same + either way; the one-row-per-angle rule that gives v10 its meaning is + checked here, since nothing downstream can recover from a v10 file that + breaks it. Returns an open binary file positioned at the start of the data block; the caller appends waveform rows and must close it (try/finally). """ + if version not in SUPPORTED_VERSIONS: + raise ValueError( + f"Cannot write SRAS format version {version} " + f"(supported: {', '.join(str(v) for v in SUPPORTED_VERSIONS)})" + ) + if version == VERSION_SAW_CHECK: + bad = [f"{pa.angle_deg:.1f}° has {pa.n_rows}" + for pa in plan.per_angle if pa.n_rows != 1] + if bad: + raise ValueError( + "A v10 SAW-check file holds exactly one row per angle, but " + + ", ".join(bad) + " — build the plan with " + "core.saw_check.middle_row_plan()." + ) + path.parent.mkdir(parents=True, exist_ok=True) f = open(path, "wb") f.write(struct.pack( - HDR_FMT, MAGIC, VERSION, + HDR_FMT, MAGIC, version, plan.n_angles, plan.x_start_nominal, plan.y_start_nominal, plan.x_delta_nominal, plan.y_delta_nominal, @@ -116,7 +147,7 @@ def create_scan_file(path: Path, plan: ScanPlan, samples_per_frame: int, @dataclass class SrasFile: - """Parsed v6 .sras file: header, tables, and lazy (memmap) data access. + """Parsed .sras file (v6 or v10): header, tables, and lazy (memmap) access. Parsing reads only the header/tables — never the waveform block — so opening a multi-GB file is cheap. ``load_angle``/``load_row`` return @@ -124,6 +155,7 @@ class SrasFile: the caller computes on it. """ path: Path + version: int = field(init=False) header: ScanHeader = field(init=False) per_angle: list[AngleGeometry] = field(init=False) preambles: list[str] = field(init=False) @@ -149,11 +181,12 @@ class SrasFile: n_channels) = struct.unpack(HDR_FMT, raw) if magic != MAGIC: raise ValueError(f"{self.path.name}: not a valid SRAS file (bad magic)") - if version != VERSION: + if version not in SUPPORTED_VERSIONS: raise ValueError( f"{self.path.name}: unsupported SRAS format version {version} " - f"(only version {VERSION} is supported)" + f"(supported: {', '.join(str(v) for v in SUPPORTED_VERSIONS)})" ) + self.version = version self.header = ScanHeader( n_angles=n_angles, x_start_nominal=x_start_nominal, y_start_nominal=y_start_nominal, @@ -187,6 +220,11 @@ class SrasFile: self.data_start_offset = f.tell() + @property + def is_saw_check(self) -> bool: + """True for a v10 middle-row SAW quality check rather than a scan.""" + return self.version == VERSION_SAW_CHECK + # ── Frontier / truncation analysis ─────────────────────────────────────── def row_bytes(self, angle_idx: int) -> int: diff --git a/saw_check_viewer.py b/saw_check_viewer.py new file mode 100755 index 0000000..1994894 --- /dev/null +++ b/saw_check_viewer.py @@ -0,0 +1,674 @@ +#!/usr/bin/env python3 +""" +SAW Check Viewer — every angle's frequency on one graph. + +Opens a v10 middle-row SAW check (written by the main app's "SAW Quality +Check") and plots the peak SAW frequency along each angle's row, all angles +on the same axes. ``core.saw_check`` explains why that answers an alignment +question: every angle's middle row crosses the same ROI centre, so the angles +all measure the same material and a spread between them belongs to the rig. + +Two readings share the window: + + * the main graph — frequency along the row, one curve per angle. Curves + that lie on top of each other and run flat are what a well-aligned rig + looks like; a curve offset from the rest indicts its angle, and a sloped + curve indicts the ROI (tilt or defocus across it, at that angle). + * the summary — each angle's median with ±1σ, plotted against angle, plus + the same numbers per angle in a table. + +A full v6 scan opens too: the same middle row is pulled out of it, so a scan +can be re-examined with the check's own read-out after the fact. +""" + +import sys +from pathlib import Path + +import numpy as np + +from PyQt6.QtCore import Qt, QThread, QTimer, pyqtSignal, QObject +from PyQt6.QtGui import QColor +from PyQt6.QtWidgets import ( + QApplication, QCheckBox, QComboBox, QDoubleSpinBox, QFileDialog, QFrame, + QGroupBox, QHBoxLayout, QHeaderView, QLabel, QListWidget, QListWidgetItem, + QMainWindow, QMessageBox, QPushButton, QSizePolicy, QSpinBox, QSplitter, + QTabWidget, QTableWidget, QTableWidgetItem, QVBoxLayout, QWidget, +) +from matplotlib import colormaps +from matplotlib.backends.backend_qtagg import FigureCanvasQTAgg, NavigationToolbar2QT +from matplotlib.figure import Figure + +sys.path.insert(0, str(Path(__file__).resolve().parent)) + +from core.saw_check import alignment_summary, frequency_traces +from core.sras_analysis import ChannelCalibration +from core.sras_format import SrasFile + +RECOMPUTE_DEBOUNCE_MS = 250 + +# Angle curve colours, sampled across the sequence so the legend reads as the +# progression 0° → 180° rather than as an arbitrary set. +ANGLE_CMAP = "viridis" + +VERDICT_STYLE = { + "good": ("#1b5e20", "#c8e6c9", "Alignment looks good"), + "marginal": ("#7a4f01", "#ffe0b2", "Alignment is marginal"), + "poor": ("#7f1d1d", "#ffcdd2", "Alignment needs attention"), +} + +X_AXIS_MODES = [ + ("Offset from row centre", "offset"), + ("Absolute stage X", "absolute"), +] + + +def angle_colors(n: int) -> list: + cmap = colormaps[ANGLE_CMAP] + if n <= 1: + return [cmap(0.5)] + return [cmap(i / (n - 1)) for i in range(n)] + + +def nan_moving_mean(y: np.ndarray, window: int) -> np.ndarray: + """Moving mean over `window` frames that steps over masked pixels. + + A plain convolution would let one NaN swallow a whole window, which on a + sparsely-masked row erases most of the trace; this divides by the number + of samples that actually contributed instead. + """ + if window <= 1: + return y + valid = np.isfinite(y) + kernel = np.ones(int(window)) + num = np.convolve(np.where(valid, y, 0.0), kernel, mode="same") + den = np.convolve(valid.astype(float), kernel, mode="same") + return np.divide(num, den, out=np.full(num.shape, np.nan), where=den > 0) + + +class LoadedCheck: + """A parsed check file plus the traces currently computed from it.""" + + def __init__(self, path: Path): + self.sras = SrasFile(path) + self.calib = ChannelCalibration.from_preambles(self.sras.preambles) + bg = np.frombuffer(self.sras.background, dtype=np.int8) + self.background = bg.astype(np.float32) if len(bg) else None + self.traces = [] + self.summary = None + + def describe(self) -> str: + h = self.sras.header + kind = ("v10 SAW check" if self.sras.is_saw_check + else f"v{self.sras.version} scan — middle row of each angle") + return (f"{self.sras.path.name}\n{kind}\n" + f"{h.n_angles} angle(s) · {h.samples_per_frame} samples/frame · " + f"{h.sample_rate / 1e9:.2f} GS/s") + + def close(self): + self.sras.close() + + +class FnWorker(QObject): + """Runs a callable on a QThread; emits its return value or the error.""" + finished = pyqtSignal(object) + error = pyqtSignal(str) + + def __init__(self, fn): + super().__init__() + self._fn = fn + + def run(self): + try: + self.finished.emit(self._fn()) + except Exception as exc: + self.error.emit(str(exc)) + + +class TraceCanvas(FigureCanvasQTAgg): + """Frequency along the row, one curve per angle, all on one axes.""" + + def __init__(self, parent=None): + fig = Figure(figsize=(8, 5), tight_layout=True) + self.ax = fig.add_subplot(111) + super().__init__(fig) + self.setParent(parent) + self.setSizePolicy(QSizePolicy.Policy.Expanding, QSizePolicy.Policy.Expanding) + self.clear("Open a SAW check file to begin.") + + def clear(self, message: str): + self.ax.clear() + self.ax.text(0.5, 0.5, message, ha="center", va="center", + transform=self.ax.transAxes, color="#888888") + self.ax.set_xticks([]) + self.ax.set_yticks([]) + self.draw_idle() + + def plot(self, traces, colors, visible, x_mode, scale, unit, y_label, + smoothing, show_median): + self.ax.clear() + shown = 0 + for trace, color in zip(traces, colors, strict=True): + if not visible.get(trace.angle_idx, True): + continue + x = trace.offset_mm if x_mode == "offset" else trace.x_mm + y = nan_moving_mean(trace.freq_mhz, smoothing) * scale + self.ax.plot(x, y, color=color, linewidth=1.0, + label=f"{trace.angle_deg:+.1f}° " + f"med {trace.median_mhz * scale:.2f}") + shown += 1 + + if shown == 0: + self.clear("No angle selected.") + return + + if show_median: + medians = [t.median_mhz for t in traces + if visible.get(t.angle_idx, True) and t.n_valid] + if medians: + self.ax.axhline(float(np.median(medians)) * scale, color="#555555", + linestyle="--", linewidth=1.0, + label="median of shown angles") + + self.ax.set_xlabel("Offset from row centre (mm)" if x_mode == "offset" + else "Stage X (mm)") + self.ax.set_ylabel(y_label) + self.ax.grid(True, alpha=0.25) + self.ax.legend(fontsize=7, ncol=2, loc="best", framealpha=0.85) + self.draw_idle() + + +class SummaryCanvas(FigureCanvasQTAgg): + """Each angle's median frequency, ±1σ, against the GR angle.""" + + def __init__(self, parent=None): + fig = Figure(figsize=(8, 2.6), tight_layout=True) + self.ax = fig.add_subplot(111) + super().__init__(fig) + self.setParent(parent) + self.setSizePolicy(QSizePolicy.Policy.Expanding, QSizePolicy.Policy.Expanding) + + def plot(self, traces, colors, scale, unit): + self.ax.clear() + usable = [(t, c) for t, c in zip(traces, colors, strict=True) if t.n_valid] + if not usable: + self.ax.set_xticks([]) + self.ax.set_yticks([]) + self.draw_idle() + return + + order = sorted(usable, key=lambda tc: tc[0].angle_deg) + angles = [t.angle_deg for t, _ in order] + medians = np.array([t.median_mhz for t, _ in order]) * scale + sigmas = np.array([0.0 if not np.isfinite(t.std_mhz) else t.std_mhz + for t, _ in order]) * scale + + self.ax.plot(angles, medians, color="#999999", linewidth=1.0, zorder=1) + self.ax.errorbar(angles, medians, yerr=sigmas, fmt="none", + ecolor="#999999", capsize=3, zorder=2) + for (_, color), angle, median in zip(order, angles, medians, strict=True): + self.ax.plot([angle], [median], marker="o", markersize=6, + color=color, zorder=3) + self.ax.axhline(float(np.median(medians)), color="#555555", + linestyle="--", linewidth=1.0) + + self.ax.set_xlabel("GR angle (deg)") + self.ax.set_ylabel(f"Median ({unit})") + self.ax.grid(True, alpha=0.25) + self.draw_idle() + + +class SawCheckWindow(QMainWindow): + """Left: what to compute and what to show. Right: the graphs.""" + + TABLE_COLUMNS = ["Angle (°)", "Y (mm)", "Median", "σ", "Drift (/mm)", "Valid (%)"] + + def __init__(self, initial_path: str | None = None): + super().__init__() + self.setWindowTitle("SAW Check Viewer") + self.resize(1280, 860) + + self._check: LoadedCheck | None = None + self._colors: list = [] + self._visible: dict[int, bool] = {} + self._compute_thread: QThread | None = None + self._compute_worker: FnWorker | None = None + self._pending_recompute = False + + self._debounce = QTimer(self) + self._debounce.setSingleShot(True) + self._debounce.setInterval(RECOMPUTE_DEBOUNCE_MS) + self._debounce.timeout.connect(self._recompute) + + self._build_ui() + if initial_path: + self._load(Path(initial_path)) + + # ── Layout ──────────────────────────────────────────────────────────────── + + def _build_ui(self): + splitter = QSplitter(Qt.Orientation.Horizontal, self) + splitter.addWidget(self._build_controls()) + splitter.addWidget(self._build_plots()) + splitter.setStretchFactor(0, 0) + splitter.setStretchFactor(1, 1) + splitter.setSizes([340, 940]) + self.setCentralWidget(splitter) + + def _build_controls(self) -> QWidget: + panel = QWidget(self) + layout = QVBoxLayout(panel) + + # File + grp_file = QGroupBox("File") + fl = QVBoxLayout(grp_file) + self.btn_open = QPushButton("Open SAW Check…") + self.btn_open.clicked.connect(self._on_open) + fl.addWidget(self.btn_open) + self.lbl_file = QLabel("No file loaded.") + self.lbl_file.setWordWrap(True) + self.lbl_file.setStyleSheet("color: #666; font-size: 11px;") + fl.addWidget(self.lbl_file) + layout.addWidget(grp_file) + + # Analysis — anything here changes the numbers, so it recomputes. + self.grp_analysis = QGroupBox("Analysis") + al = QVBoxLayout(self.grp_analysis) + + thr_row = QHBoxLayout() + thr_row.addWidget(QLabel("CH4 DC threshold:")) + self.spin_threshold_mv = QDoubleSpinBox() + self.spin_threshold_mv.setRange(-500.0, 500.0) + self.spin_threshold_mv.setDecimals(1) + self.spin_threshold_mv.setSingleStep(5.0) + self.spin_threshold_mv.setSuffix(" mV") + self.spin_threshold_mv.setValue(50.0) + self.spin_threshold_mv.setToolTip( + "Pixels whose CH4 DC mean falls below this are dropped from the " + "trace — the detection beam was off the sample or out of focus there." + ) + self.spin_threshold_mv.valueChanged.connect(self._queue_recompute) + thr_row.addWidget(self.spin_threshold_mv) + al.addLayout(thr_row) + + self.chk_bg_sub = QCheckBox("Subtract background waveform") + self.chk_bg_sub.setChecked(True) + self.chk_bg_sub.toggled.connect(self._queue_recompute) + al.addWidget(self.chk_bg_sub) + + self.chk_gate = QCheckBox("Time gate before FFT") + self.chk_gate.toggled.connect(self._on_gate_toggled) + al.addWidget(self.chk_gate) + + gate_row = QHBoxLayout() + gate_row.addWidget(QLabel("Start:")) + self.spin_gate_start = QDoubleSpinBox() + self.spin_gate_start.setRange(0.0, 100000.0) + self.spin_gate_start.setDecimals(1) + self.spin_gate_start.setSingleStep(10.0) + self.spin_gate_start.setSuffix(" ns") + self.spin_gate_start.setValue(50.0) + self.spin_gate_start.setEnabled(False) + self.spin_gate_start.valueChanged.connect(self._queue_recompute) + gate_row.addWidget(self.spin_gate_start) + gate_row.addWidget(QLabel("End:")) + self.spin_gate_end = QDoubleSpinBox() + self.spin_gate_end.setRange(0.0, 100000.0) + self.spin_gate_end.setDecimals(1) + self.spin_gate_end.setSingleStep(10.0) + self.spin_gate_end.setSuffix(" ns") + self.spin_gate_end.setValue(200.0) + self.spin_gate_end.setEnabled(False) + self.spin_gate_end.valueChanged.connect(self._queue_recompute) + gate_row.addWidget(self.spin_gate_end) + al.addLayout(gate_row) + layout.addWidget(self.grp_analysis) + + # Display — cheap, so these only redraw. + grp_display = QGroupBox("Display") + dl = QVBoxLayout(grp_display) + + x_row = QHBoxLayout() + x_row.addWidget(QLabel("X axis:")) + self.combo_x = QComboBox() + for label, _ in X_AXIS_MODES: + self.combo_x.addItem(label) + self.combo_x.setToolTip( + "Every angle's row is centred on the same ROI centre, so offset " + "puts the angles over the same piece of sample; absolute shows " + "where each rotated bounding box actually sat on the stage." + ) + self.combo_x.currentIndexChanged.connect(self._redraw) + x_row.addWidget(self.combo_x) + dl.addLayout(x_row) + + y_row = QHBoxLayout() + y_row.addWidget(QLabel("Y axis:")) + self.combo_y = QComboBox() + self.combo_y.addItems(["Frequency (MHz)", "Velocity (m/s)"]) + self.combo_y.currentIndexChanged.connect(self._on_y_mode_changed) + y_row.addWidget(self.combo_y) + dl.addLayout(y_row) + + grat_row = QHBoxLayout() + grat_row.addWidget(QLabel("Grating:")) + self.spin_grating_um = QDoubleSpinBox() + self.spin_grating_um.setRange(0.1, 1000.0) + self.spin_grating_um.setDecimals(2) + self.spin_grating_um.setSingleStep(0.5) + self.spin_grating_um.setSuffix(" µm") + self.spin_grating_um.setValue(12.5) + self.spin_grating_um.setEnabled(False) + self.spin_grating_um.setToolTip("v (m/s) = freq (MHz) × grating (µm)") + self.spin_grating_um.valueChanged.connect(self._redraw) + grat_row.addWidget(self.spin_grating_um) + dl.addLayout(grat_row) + + smooth_row = QHBoxLayout() + smooth_row.addWidget(QLabel("Smoothing:")) + self.spin_smoothing = QSpinBox() + self.spin_smoothing.setRange(1, 2001) + self.spin_smoothing.setSingleStep(10) + self.spin_smoothing.setSuffix(" frames") + self.spin_smoothing.setValue(1) + self.spin_smoothing.setToolTip( + "Moving average along the row, masked pixels skipped. Display " + "only — the table's statistics always use the unsmoothed trace." + ) + self.spin_smoothing.valueChanged.connect(self._redraw) + smooth_row.addWidget(self.spin_smoothing) + dl.addLayout(smooth_row) + + self.chk_median_line = QCheckBox("Show median of shown angles") + self.chk_median_line.setChecked(True) + self.chk_median_line.toggled.connect(self._redraw) + dl.addWidget(self.chk_median_line) + layout.addWidget(grp_display) + + # Angles + grp_angles = QGroupBox("Angles") + gl = QVBoxLayout(grp_angles) + self.list_angles = QListWidget() + self.list_angles.setMaximumHeight(190) + self.list_angles.itemChanged.connect(self._on_angle_toggled) + gl.addWidget(self.list_angles) + btn_row = QHBoxLayout() + btn_all = QPushButton("All") + btn_all.clicked.connect(lambda: self._set_all_angles(True)) + btn_none = QPushButton("None") + btn_none.clicked.connect(lambda: self._set_all_angles(False)) + btn_row.addWidget(btn_all) + btn_row.addWidget(btn_none) + gl.addLayout(btn_row) + layout.addWidget(grp_angles) + + # Verdict + self.lbl_verdict = QLabel("—") + self.lbl_verdict.setWordWrap(True) + self.lbl_verdict.setFrameShape(QFrame.Shape.StyledPanel) + self.lbl_verdict.setMinimumHeight(92) + self.lbl_verdict.setAlignment(Qt.AlignmentFlag.AlignTop) + layout.addWidget(self.lbl_verdict) + + self.lbl_status = QLabel("") + self.lbl_status.setStyleSheet("color: #666; font-size: 11px;") + layout.addWidget(self.lbl_status) + + layout.addStretch(1) + return panel + + def _build_plots(self) -> QWidget: + splitter = QSplitter(Qt.Orientation.Vertical, self) + + top = QWidget() + tl = QVBoxLayout(top) + tl.setContentsMargins(0, 0, 0, 0) + self.trace_canvas = TraceCanvas(top) + tl.addWidget(NavigationToolbar2QT(self.trace_canvas, top)) + tl.addWidget(self.trace_canvas) + splitter.addWidget(top) + + tabs = QTabWidget() + self.summary_canvas = SummaryCanvas(tabs) + tabs.addTab(self.summary_canvas, "Frequency vs angle") + + self.table = QTableWidget(0, len(self.TABLE_COLUMNS)) + self.table.setHorizontalHeaderLabels(self.TABLE_COLUMNS) + self.table.horizontalHeader().setSectionResizeMode( + QHeaderView.ResizeMode.Stretch) + self.table.setEditTriggers(QTableWidget.EditTrigger.NoEditTriggers) + tabs.addTab(self.table, "Per-angle statistics") + splitter.addWidget(tabs) + + splitter.setStretchFactor(0, 3) + splitter.setStretchFactor(1, 1) + # Stretch factors alone leave the summary too short to fit its own + # axis label on first show; give it a real starting height. + splitter.setSizes([540, 300]) + return splitter + + # ── Loading ─────────────────────────────────────────────────────────────── + + def _on_open(self): + start = str(self._check.sras.path.parent) if self._check else "" + path, _ = QFileDialog.getOpenFileName( + self, "Open SAW Check File", start, "SRAS Files (*.sras)") + if path: + self._load(Path(path)) + + def _load(self, path: Path): + try: + check = LoadedCheck(path) + except Exception as exc: + QMessageBox.critical(self, "Cannot Open File", + f"Could not read {path.name}:\n\n{exc}") + return + + if self._check is not None: + self._check.close() + self._check = check + self.setWindowTitle(f"SAW Check Viewer — {path.name}") + self.lbl_file.setText(check.describe()) + + if not check.sras.is_saw_check: + self.lbl_status.setText( + "Not a v10 check — reading the middle row of each angle " + "out of this scan instead.") + else: + self.lbl_status.setText("") + + self._colors = angle_colors(check.sras.header.n_angles) + self._visible = {i: True for i in range(check.sras.header.n_angles)} + self._recompute() + + # ── Compute ─────────────────────────────────────────────────────────────── + + def _queue_recompute(self): + if self._check is not None: + self._debounce.start() + + def _on_gate_toggled(self, enabled: bool): + self.spin_gate_start.setEnabled(enabled) + self.spin_gate_end.setEnabled(enabled) + self._queue_recompute() + + def _recompute(self): + if self._check is None: + return + if self._compute_thread is not None and self._compute_thread.isRunning(): + # One worker owns the mmap at a time; fold this request into the + # one already in flight rather than racing it. + self._pending_recompute = True + return + + check = self._check + gated = self.chk_gate.isChecked() + kwargs = dict( + dc_threshold_mv=self.spin_threshold_mv.value(), + background=check.background if self.chk_bg_sub.isChecked() else None, + gate_start_ns=self.spin_gate_start.value() if gated else None, + gate_end_ns=self.spin_gate_end.value() if gated else None, + calib=check.calib, + ) + + self.grp_analysis.setEnabled(False) + self.lbl_status.setText("Computing frequency traces …") + + self._compute_thread = QThread(self) + self._compute_worker = FnWorker( + lambda: frequency_traces(check.sras, **kwargs)) + self._compute_worker.moveToThread(self._compute_thread) + self._compute_thread.started.connect(self._compute_worker.run) + self._compute_worker.finished.connect(self._on_traces_ready) + self._compute_worker.error.connect(self._on_compute_error) + self._compute_thread.start() + + def _finish_compute(self): + if self._compute_thread is not None: + self._compute_thread.quit() + self._compute_thread.wait(5000) + self._compute_thread = None + self._compute_worker = None + self.grp_analysis.setEnabled(True) + if self._pending_recompute: + self._pending_recompute = False + self._queue_recompute() + + def _on_compute_error(self, message: str): + self._finish_compute() + self.lbl_status.setText("") + QMessageBox.critical(self, "Analysis Failed", message) + + def _on_traces_ready(self, traces): + self._finish_compute() + if self._check is None: + return + self._check.traces = traces + self._check.summary = alignment_summary(traces) + self.lbl_status.setText( + f"{len(traces)} of {self._check.sras.header.n_angles} angle(s) " + f"produced a trace.") + self._rebuild_angle_list() + self._redraw() + + # ── Display ─────────────────────────────────────────────────────────────── + + def _scale(self) -> tuple[float, str, str]: + """Display factor, unit and axis label. + + The file only ever holds a frequency; velocity is that frequency times + the grating period, applied at display time so switching units never + costs a recompute. + """ + if self.combo_y.currentIndex() == 1: + return self.spin_grating_um.value(), "m/s", "SAW velocity (m/s)" + return 1.0, "MHz", "Peak SAW frequency (MHz)" + + def _on_y_mode_changed(self): + self.spin_grating_um.setEnabled(self.combo_y.currentIndex() == 1) + self._redraw() + + def _rebuild_angle_list(self): + self.list_angles.blockSignals(True) + self.list_angles.clear() + for trace in self._check.traces: + item = QListWidgetItem( + f"{trace.angle_deg:+7.2f}° Y={trace.y_mm:.3f} mm") + item.setFlags(item.flags() | Qt.ItemFlag.ItemIsUserCheckable) + item.setCheckState( + Qt.CheckState.Checked if self._visible.get(trace.angle_idx, True) + else Qt.CheckState.Unchecked) + item.setData(Qt.ItemDataRole.UserRole, trace.angle_idx) + r, g, b, _ = self._colors[trace.angle_idx] + item.setForeground(QColor(int(r * 255), int(g * 255), int(b * 255))) + self.list_angles.addItem(item) + self.list_angles.blockSignals(False) + + def _on_angle_toggled(self, item: QListWidgetItem): + self._visible[item.data(Qt.ItemDataRole.UserRole)] = ( + item.checkState() == Qt.CheckState.Checked) + self._redraw() + + def _set_all_angles(self, visible: bool): + self.list_angles.blockSignals(True) + for row in range(self.list_angles.count()): + item = self.list_angles.item(row) + item.setCheckState(Qt.CheckState.Checked if visible + else Qt.CheckState.Unchecked) + self._visible[item.data(Qt.ItemDataRole.UserRole)] = visible + self.list_angles.blockSignals(False) + self._redraw() + + def _redraw(self): + if self._check is None or not self._check.traces: + self.trace_canvas.clear("No angle in this file has data on disk.") + return + traces = self._check.traces + colors = [self._colors[t.angle_idx] for t in traces] + scale, unit, y_label = self._scale() + + self.trace_canvas.plot( + traces, colors, self._visible, + X_AXIS_MODES[self.combo_x.currentIndex()][1], scale, unit, y_label, + self.spin_smoothing.value(), self.chk_median_line.isChecked()) + self.summary_canvas.plot(traces, colors, scale, unit) + self._fill_table(traces, scale, unit) + self._show_verdict(scale, unit) + + def _fill_table(self, traces, scale: float, unit: str): + headers = list(self.TABLE_COLUMNS) + headers[2] = f"Median ({unit})" + headers[3] = f"σ ({unit})" + headers[4] = f"Drift ({unit}/mm)" + self.table.setHorizontalHeaderLabels(headers) + + self.table.setRowCount(len(traces)) + for row, trace in enumerate(traces): + values = [ + f"{trace.angle_deg:+.2f}", + f"{trace.y_mm:.3f}", + f"{trace.median_mhz * scale:.3f}", + f"{trace.std_mhz * scale:.3f}", + f"{trace.drift_mhz_per_mm * scale:+.4f}", + f"{trace.valid_fraction * 100:.1f}", + ] + for col, text in enumerate(values): + item = QTableWidgetItem(text) + item.setTextAlignment(Qt.AlignmentFlag.AlignRight + | Qt.AlignmentFlag.AlignVCenter) + if col == 0: + r, g, b, _ = self._colors[trace.angle_idx] + item.setForeground(QColor(int(r * 255), int(g * 255), int(b * 255))) + self.table.setItem(row, col, item) + + def _show_verdict(self, scale: float, unit: str): + summary = self._check.summary + fg, bg, headline = VERDICT_STYLE[summary.level] + detail = summary.describe() + if scale != 1.0 and summary.n_angles: + detail += (f"\nIn {unit}: spread {summary.spread_mhz * scale:.3f} " + f"about {summary.median_mhz * scale:.1f}.") + self.lbl_verdict.setText(f"{headline}\n\n{detail}") + self.lbl_verdict.setStyleSheet( + f"color: {fg}; background: {bg}; padding: 8px; font-size: 11px;") + + # ── Teardown ────────────────────────────────────────────────────────────── + + def closeEvent(self, event): + self._debounce.stop() + if self._compute_thread is not None: + self._compute_thread.quit() + self._compute_thread.wait(5000) + if self._check is not None: + self._check.close() + super().closeEvent(event) + + +def main(): + app = QApplication(sys.argv) + window = SawCheckWindow(sys.argv[1] if len(sys.argv) > 1 else None) + window.show() + sys.exit(app.exec()) + + +if __name__ == "__main__": + main() diff --git a/sc3-aui-main.ui b/sc3-aui-main.ui index 91dd41d..feb959d 100755 --- a/sc3-aui-main.ui +++ b/sc3-aui-main.ui @@ -1017,6 +1017,16 @@ + + + + Acquire one row per angle — the row-wise middle of the ROI — and save it as a v10 .sras SAW check. Costs one row-time per angle instead of a full scan, and every angle's row crosses the same ROI centre, so the per-angle frequencies can be compared in the SAW Check Viewer to judge the alignment. + + + SAW Quality Check… + + + @@ -1086,6 +1096,7 @@ bbd_set_current_start_btn bbd_set_delta_current_btn show_camera_toggle + saw_check_btn start_scan_btn diff --git a/sc3_aui_app.py b/sc3_aui_app.py index c5cf7cc..285faa2 100755 --- a/sc3_aui_app.py +++ b/sc3_aui_app.py @@ -6,6 +6,7 @@ and wires up T3R, BBD202, oscilloscope, and camera hardware workers. """ import struct +import subprocess import sys import time from pathlib import Path @@ -32,10 +33,13 @@ from core.scan_engine import ( ResumeState, LASER_FREQ_HZ, SCAN_VELOCITY_MM_S, ) +from core.saw_check import middle_row_plan from core.scan_geometry import ScanPlan, EtaEstimator, build_plan, format_eta from core.scan_resume import is_compatible, plan_resume from core.scope_sras import SAMPLE_RATE_HZ, configure_channels -from core.sras_format import SCAN_CHANNELS, SrasFile, plan_from_header +from core.sras_format import ( + SCAN_CHANNELS, VERSION, VERSION_SAW_CHECK, SrasFile, plan_from_header, +) from gui.qt_t3r import QtT3RAdapter from gui.qt_workers import PollingQueueWorker, QueueWorker from gui.inspect_bridge import QtAngleInspector @@ -51,6 +55,10 @@ from t3r_control_panel import T3RControlPanel DEFAULTS = ScanDefaults.load() BBD_DEFAULT_JOG_MM = 0.5 # default jog step for BBD202 +# A SAW check is written beside the scan it belongs to, under the same prefix. +# The suffix keeps it from overwriting the scan itself, which is the one file +# in the directory that cost hours to acquire. +SAW_CHECK_SUFFIX = "-sawcheck" class DCBiasImageWidget(FigureCanvas): @@ -977,6 +985,11 @@ class MainWindow(QMainWindow): self._scan_progress = ScanProgressWindow() self._scan_worker: QtScanController | None = None + # A SAW check runs through the same worker as a scan; this says which, + # since the two finish very differently (a check hands the operator a + # file to look at; a scan shuts the rig down). + self._scan_is_saw_check = False + self._saw_check_path: Path | None = None self._inspect_thread: QThread | None = None self._inspect_worker: QtAngleInspector | None = None self._inspect_window: AngleInspectWindow | None = None @@ -1101,6 +1114,7 @@ class MainWindow(QMainWindow): # Scan self.start_scan_btn.clicked.connect(self._on_start_scan) + self.saw_check_btn.clicked.connect(self._on_saw_check) self.inspect_angles_btn.clicked.connect(self._on_inspect_angles) self.save_dir_browse_btn.clicked.connect(self._on_browse_save_dir) self._scan_progress.abort_requested.connect(self._on_abort_scan) @@ -1267,6 +1281,11 @@ class MainWindow(QMainWindow): # ── Scan ────────────────────────────────────────────────────────────────── + def _set_scan_buttons_enabled(self, enabled: bool): + """Both entry points drive the same rig, so they lock and unlock together.""" + self.start_scan_btn.setEnabled(enabled) + self.saw_check_btn.setEnabled(enabled) + def _on_browse_save_dir(self): d = QFileDialog.getExistingDirectory( self, "Select Scan Save Directory", self.scan_save_dir_edit.text() @@ -1275,6 +1294,82 @@ class MainWindow(QMainWindow): self.scan_save_dir_edit.setText(d) self._persist_defaults() + def _on_saw_check(self): + """Acquire the middle row of the current ROI at every angle. + + Same engine, same hardware sequence, same file format as a scan — the + plan is just reduced to one row per angle and the result is tagged v10 + so the viewer knows it is a check rather than a scan cut short. + """ + if self._scan_thread is not None and self._scan_thread.isRunning(): + QMessageBox.warning( + self, "Scan In Progress", + "A scan is running — abort it before starting a SAW check." + ) + return + try: + plan, prefix, save_dir = self._build_scan_plan() + check_plan = middle_row_plan(plan) # ScanGeometryError is a ValueError + except ValueError as e: + QMessageBox.warning(self, "Invalid Scan Parameters", str(e)) + return + + check_prefix = f"{prefix}{SAW_CHECK_SUFFIX}" + out_path = Path(save_dir) / f"{check_prefix}.sras" + rows = ", ".join(f"{pa.angle_deg:.1f}°: Y={pa.y_positions[0]:.3f} mm" + for pa in check_plan.per_angle) + overwrite = ("\n\nThis will overwrite the existing file." + if out_path.exists() else "") + reply = QMessageBox.question( + self, "SAW Quality Check", + f"Acquire the middle row of the ROI at {check_plan.n_angles} angle(s)?\n\n" + f"{rows}\n\n" + f"Save → {out_path.name}{overwrite}", + QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.No, + ) + if reply != QMessageBox.StandardButton.Yes: + return + + self._saw_check_path = out_path + # Burst mode is deliberately not offered here: one row per angle means + # every burst would be a single row, so it buys nothing and still pays + # for the gate preflight. + self._launch_scan_worker(check_plan, check_prefix, save_dir, saw_check=True) + + def _on_saw_check_complete(self): + """A check is a thing to look at, not a run to shut down after.""" + path = self._saw_check_path + box = QMessageBox(self) + box.setIcon(QMessageBox.Icon.Information) + box.setWindowTitle("SAW Check Complete") + box.setText( + f"Middle-row SAW check written to:\n{path}\n\n" + "Open it in the SAW Check Viewer to compare each angle's " + "frequency and judge the alignment." + ) + open_btn = box.addButton("Open Viewer", QMessageBox.ButtonRole.AcceptRole) + box.addButton(QMessageBox.StandardButton.Close) + box.exec() + if box.clickedButton() is open_btn: + self._launch_saw_check_viewer(path) + + def _launch_saw_check_viewer(self, path: Path): + """Open the viewer as its own process. + + Deliberately not in-process: the acquisition app owns the hardware and + must stay responsive, and the viewer is a separate entry point that + outlives any one scan session. + """ + try: + subprocess.Popen([sys.executable, + str(ROOT / "saw_check_viewer.py"), str(path)]) + except OSError as e: + QMessageBox.warning( + self, "Could Not Open Viewer", + f"Could not start the SAW Check Viewer:\n\n{e}\n\n" + f"Run it manually: python saw_check_viewer.py {path}" + ) + def _on_inspect_angles(self): """Open the pre-scan angle inspector for the plan currently entered.""" if self._scan_thread is not None and self._scan_thread.isRunning(): @@ -1323,7 +1418,7 @@ class MainWindow(QMainWindow): lambda m: QMessageBox.warning(self, "Inspection Error", m)) self._inspect_thread.started.connect(self._inspect_worker.run) - self.start_scan_btn.setEnabled(False) + self._set_scan_buttons_enabled(False) self.inspect_angles_btn.setEnabled(False) window.show() self._inspect_thread.start() @@ -1345,7 +1440,7 @@ class MainWindow(QMainWindow): self._inspect_thread = None self._inspect_worker = None self._inspect_window = None - self.start_scan_btn.setEnabled(True) + self._set_scan_buttons_enabled(True) self.inspect_angles_btn.setEnabled(True) def _on_start_scan(self): @@ -1424,8 +1519,10 @@ class MainWindow(QMainWindow): str(path.parent), resume_plan.to_state(sras)) def _launch_scan_worker(self, plan: ScanPlan, prefix: str, save_dir: str, - resume: ResumeState | None = None): + resume: ResumeState | None = None, + saw_check: bool = False): rotator = RotationAxis(self._t3r_driver.driver, DEFAULT_ROTATION) + self._scan_is_saw_check = saw_check self._scan_thread = QThread(self) self._scan_worker = QtScanController( @@ -1439,8 +1536,9 @@ class MainWindow(QMainWindow): # a worker concern, not the engine's. on_scan_active=lambda active: setattr( self._bbd_worker, "scanning_active", active), - burst_mode=self.burst_mode_check.isChecked(), + burst_mode=self.burst_mode_check.isChecked() and not saw_check, strict_rows=self.strict_rows_check.isChecked(), + file_version=VERSION_SAW_CHECK if saw_check else VERSION, ) self._scan_worker.moveToThread(self._scan_thread) self._scan_thread.started.connect(self._scan_worker.run) @@ -1453,7 +1551,7 @@ class MainWindow(QMainWindow): self._scan_worker.user_prompt.connect(self._on_scan_user_prompt) self._scan_worker.paused_changed.connect(self._scan_progress.on_worker_paused) - self.start_scan_btn.setEnabled(False) + self._set_scan_buttons_enabled(False) self._scan_progress.reset_pause_btn() ai0 = 0 if resume is None else resume.targets[0].angle_idx self._scan_progress.update_progress( @@ -1494,7 +1592,11 @@ class MainWindow(QMainWindow): def _on_scan_complete(self): self._scan_progress.close() - self.start_scan_btn.setEnabled(True) + self._set_scan_buttons_enabled(True) + if self._scan_is_saw_check: + self._scan_is_saw_check = False + self._on_saw_check_complete() + return QMessageBox.information( self, "Scan Complete", "All rows and angles have been acquired.\n\n" @@ -1504,7 +1606,8 @@ class MainWindow(QMainWindow): def _on_scan_failed(self, msg: str): self._scan_progress.close() - self.start_scan_btn.setEnabled(True) + self._set_scan_buttons_enabled(True) + self._scan_is_saw_check = False if "aborted" in msg.lower(): QMessageBox.warning(self, "Scan Aborted", msg) else: diff --git a/scan_format.md b/scan_format.md index ad69ab8..8c7e2d2 100755 --- a/scan_format.md +++ b/scan_format.md @@ -1,8 +1,18 @@ -# SRAS Scan Binary Format — Version 6 +# SRAS Scan Binary Format — Versions 6 and 10 Each `.sras` file contains **one complete scan**: all GR rotation angles and all Y rows. Files are named `{prefix}.sras`. +Two versions share this layout byte for byte — only the version field differs, +and with it what the file means: + +| Version | Meaning | Rows per angle | +|---------|---------|----------------| +| 6 | A full scan. | Whatever the ROI needs. | +| 10 | A middle-row SAW quality check (`{prefix}-sawcheck.sras`). | Exactly 1. | + +See [SAW Quality Check (v10)](#saw-quality-check-v10) below. + Starting in v6, each angle only scans the **bounding box of the nominal ROI rotated by that specific angle** — not the worst case across all angles — so `x_start`, `x_delta` (and therefore `n_frames`, the points/row count) and @@ -34,7 +44,7 @@ All multi-byte integers and floats use **big-endian** byte order | Offset | Size | Type | Field | Description | |--------|------|-----------|--------------------|--------------------------------------------------| | 0 | 4 | `4s` | `magic` | Always `SRAS` (0x53 0x52 0x41 0x53) | -| 4 | 1 | `uint8` | `version` | Format version — `6` | +| 4 | 1 | `uint8` | `version` | Format version — `6` (scan) or `10` (SAW check) | | 5 | 2 | `uint16` | `n_angles` | Number of GR rotation angles | | 7 | 4 | `float32` | `x_start_nominal` | Nominal (pre-rotation) X scan start, mm | | 11 | 4 | `float32` | `y_start_nominal` | Nominal (pre-rotation) Y scan start, mm | @@ -242,6 +252,38 @@ the file always ends on a whole-row boundary. --- +## SAW Quality Check (v10) + +A full multi-angle scan takes hours, and a rig whose angles disagree produces +all of them before anyone finds out. The SAW quality check acquires **one row +per angle — the row-wise middle of the ROI** — and writes it as a v10 file. +The cost is one row-time per angle instead of `n_rows` of them. + +Nothing about the byte layout changes. A v10 file is a v6 file in which every +angle's Per-Angle Geometry Table entry declares `n_rows = 1`, and its Row Table +holds that angle's single middle Y position. Every v6 reader that works from +the geometry table (rather than assuming a uniform shape) reads a v10 file +unchanged. + +The version byte earns its keep because the two are otherwise +indistinguishable: **a v6 scan aborted after its first row is not a check**, +even though both hold one row per angle. A reader that guessed from the row +count would treat a failed scan as a deliberate measurement. + +Why the middle row in particular: `core/scan_geometry.py` centres every +angle's rotated bounding box on the same nominal ROI centre, so each angle's +middle row crosses that one point on the sample. All the angles therefore +measure the same material, and a spread in their SAW frequencies is a property +of the rig — which is what makes it an alignment check. `saw_check_viewer.py` +plots every angle's frequency on one graph for exactly that comparison. + +Writers must honour the one-row rule; `core.sras_format.create_scan_file` +refuses a v10 write for any plan that breaks it. Producing the plan is +`core.saw_check.middle_row_plan(plan)`, and `n_rows // 2` is the middle-row +rule (the upper of the two central rows when the count is even). + +--- + ## Version History | Version | Change | @@ -252,4 +294,5 @@ the file always ends on a whole-row boundary. | 4 | Added background waveform block (CH1, Helios ON / Genesis OFF) after the preamble blocks; stored as `uint32` sample count followed by raw `int8` ADC bytes. | | 5 | (skipped) | | 6 | Each angle now scans only the bounding box of the nominal ROI rotated by that angle instead of the AABB-expanded worst case across all angles. Header no longer carries a single global `x_start`/`x_delta`/`n_rows` — replaced with `*_nominal` reference fields plus a new Per-Angle Geometry Table (`x_start`, `x_delta`, `n_frames`, `n_rows` per angle) and a ragged Row Table / Waveform Data block sized per angle. **Not compatible with v4 readers** (e.g. `sras_viewer.py`, which has not yet been updated for v6). | - +| 7–9 | (skipped) | +| 10 | Middle-row SAW quality check. Byte layout identical to v6, with every angle declaring exactly one row — the row-wise middle of the ROI. A v6 reader that derives its shape from the Per-Angle Geometry Table reads these unchanged; the version byte exists so a check is not confused with a scan aborted after its first row. Written by the main app's *SAW Quality Check*, read by `saw_check_viewer.py`. | diff --git a/sras_scan_manager.py b/sras_scan_manager.py index 9274a01..901c947 100755 --- a/sras_scan_manager.py +++ b/sras_scan_manager.py @@ -9,10 +9,12 @@ n_rows) and waveform data block. This tool lists those per-angle sub-scans and lets you export a subset to a new .sras file, or delete a subset from the file in place — both operations rewrite the angle/geometry/row tables and stream-copy only the selected angles' waveform data, producing a file -that is itself a valid v6 .sras readable by sras_viewer.py-style tools +that is itself a valid .sras readable by sras_viewer.py-style tools (once updated for v6) or sc3_aui_app.py. -Only format version 6 is supported. +Format versions 6 (full scan) and 10 (middle-row SAW check) are supported. +A subset keeps the version of the file it came from — a v10 check exports as +a v10 check, since dropping angles from one leaves it one row per angle. """ import argparse @@ -24,7 +26,7 @@ from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parent)) -from core.sras_format import GEOM_FMT, HDR_FMT, MAGIC, VERSION as BLOB_VERSION, SrasFile +from core.sras_format import GEOM_FMT, HDR_FMT, MAGIC, VERSION_SAW_CHECK, SrasFile @dataclass @@ -48,7 +50,7 @@ class AngleEntry: class SrasScanFile: - """Parsed view of a v6 .sras file's header/tables plus per-angle data offsets.""" + """Parsed view of a .sras file's header/tables plus per-angle data offsets.""" def __init__(self, path: Path): self.path = Path(path) @@ -57,6 +59,7 @@ class SrasScanFile: def _parse(self): sras = SrasFile(self.path) h = sras.header + self.version = sras.version self.x_start_nominal = h.x_start_nominal self.y_start_nominal = h.y_start_nominal self.x_delta_nominal = h.x_delta_nominal @@ -96,7 +99,7 @@ class SrasScanFile: # --------------------------------------------------------------------------- def _write_subset(sf: SrasScanFile, indices: list, dst_path: Path) -> list: - """Write a new v6 .sras file containing only the given angle indices + """Write a new .sras file containing only the given angle indices (in the given order). Returns a list of warning strings (e.g. for angles that were truncated on disk and thus exported with fewer rows than declared). @@ -105,7 +108,7 @@ def _write_subset(sf: SrasScanFile, indices: list, dst_path: Path) -> list: selected = [sf.get(i) for i in indices] header = struct.pack( - HDR_FMT, MAGIC, BLOB_VERSION, len(selected), + HDR_FMT, MAGIC, sf.version, len(selected), sf.x_start_nominal, sf.y_start_nominal, sf.x_delta_nominal, sf.y_delta_nominal, sf.row_spacing_mm, sf.velocity_mm_s, sf.laser_freq_hz, @@ -221,7 +224,8 @@ def parse_index_spec(spec: str, max_index: int) -> list: def print_summary(sf: SrasScanFile, selected: set): print() - print(f"File: {sf.path} (v{BLOB_VERSION}, {_human_size(sf.file_size)})") + kind = " SAW check" if sf.version == VERSION_SAW_CHECK else "" + print(f"File: {sf.path} (v{sf.version}{kind}, {_human_size(sf.file_size)})") print(f"Nominal ROI: x_start={sf.x_start_nominal:.4f} x_delta={sf.x_delta_nominal:.4f} " f"y_start={sf.y_start_nominal:.4f} y_delta={sf.y_delta_nominal:.4f} mm " f"row_spacing={sf.row_spacing_mm:.4f} mm") diff --git a/tests/test_saw_check.py b/tests/test_saw_check.py new file mode 100644 index 0000000..b764da5 --- /dev/null +++ b/tests/test_saw_check.py @@ -0,0 +1,332 @@ +"""Middle-row SAW quality check: plan reduction, the v10 file, and the read-out. + +The acquisition half runs on the same fake rig as the scan tests; the +analysis half runs on a synthetic v10 file whose CH1 is a pure sine at a +known FFT bin, so the frequency a trace reports is a number the test knows +in advance rather than one it copies from the implementation. +""" +import math + +import numpy as np +import pytest + +from core.rotation import RotationAxis, RotationSettings +from core.saw_check import ( + SPREAD_GOOD_PCT, alignment_summary, frequency_traces, middle_row_index, + middle_row_plan, +) +from core.scan_engine import ScanCallbacks, ScanEngine +from core.scan_geometry import ScanGeometryError, build_plan +from core.sras_format import ( + SCAN_CHANNELS, VERSION, VERSION_SAW_CHECK, SrasFile, create_scan_file, +) +from fakes import FakeScope, FakeStage, FakeT3R, Trace + +SAMPLE_RATE = 6.25e9 +SPF = 256 +LASER_FREQ_HZ = 20000.0 +VELOCITY_MM_S = 100.0 +PREAMBLES = [f"WFMOUTPRE:CH{ch};YMULT 1.5625E-3;YOFF -87.04;YZERO 0.0" + for ch in SCAN_CHANNELS] +# adc_to_mv with those constants maps 0 → +136 mV and -120 → -51 mV, so a +# frame of zeros passes a 50 mV CH4 gate and a frame of -120 does not. +DC_THRESHOLD_MV = 50.0 +CH4_PASS = bytes(SPF) +CH4_FAIL = bytes([256 - 120]) * SPF + + +def full_plan(num_angles=3, y_delta=0.05): + """A small ROI, well inside the stage limits, with several rows per angle.""" + return build_plan(40.0, 30.0, 0.02, y_delta, num_angles, 0.01, + laser_freq_hz=LASER_FREQ_HZ, velocity_mm_s=VELOCITY_MM_S) + + +def bin_mhz(k: int) -> float: + return k * SAMPLE_RATE / SPF / 1e6 + + +def sine_frame(k: int) -> bytes: + """One frame holding a pure sine at FFT bin `k`.""" + n = np.arange(SPF) + return np.round(100 * np.sin(2 * math.pi * k * n / SPF)).astype(np.int8).tobytes() + + +def write_check(path, bins, n_masked_frames=0, plan=None): + """A synthetic v10 file: angle `i`'s CH1 is a sine at FFT bin `bins[i]`.""" + plan = plan if plan is not None else middle_row_plan(full_plan(len(bins))) + f = create_scan_file(path, plan, SPF, SAMPLE_RATE, PREAMBLES, bytes(SPF), + version=VERSION_SAW_CHECK) + try: + for ai, pa in enumerate(plan.per_angle): + wave = sine_frame(bins[ai]) + for ch in SCAN_CHANNELS: + for fi in range(pa.n_frames): + if ch == 1: + f.write(wave) + elif ch == 3: + f.write(bytes(SPF)) + else: + f.write(CH4_FAIL if fi < n_masked_frames else CH4_PASS) + finally: + f.close() + return plan + + +# ── Plan reduction ─────────────────────────────────────────────────────────── + +def test_middle_row_plan_keeps_one_middle_row_per_angle(): + plan = full_plan(num_angles=3) + check = middle_row_plan(plan) + + assert check.n_angles == plan.n_angles + assert [pa.n_rows for pa in check.per_angle] == [1] * plan.n_angles + for original, reduced in zip(plan.per_angle, check.per_angle, strict=True): + mid = original.n_rows // 2 + assert reduced.y_positions == [original.y_positions[mid]] + # The row is scanned exactly as the full scan would have scanned it. + assert reduced.angle_deg == original.angle_deg + assert reduced.x_start == original.x_start + assert reduced.x_delta == original.x_delta + assert reduced.n_frames == original.n_frames + + +def test_middle_row_plan_does_not_mutate_its_input(): + plan = full_plan(num_angles=3) + before = [(pa.n_rows, list(pa.y_positions)) for pa in plan.per_angle] + middle_row_plan(plan) + assert [(pa.n_rows, pa.y_positions) for pa in plan.per_angle] == before + + +def test_every_angles_middle_row_crosses_the_roi_centre(): + """The premise the whole comparison rests on: one shared point on the sample.""" + plan = full_plan(num_angles=5) + check = middle_row_plan(plan) + cx = plan.x_start_nominal + plan.x_delta_nominal / 2 + cy = plan.y_start_nominal + plan.y_delta_nominal / 2 + for pa in check.per_angle: + assert pa.x_start + pa.x_delta / 2 == pytest.approx(cx, abs=1e-6) + # Within one row spacing — the middle row is a grid point, not exact. + assert abs(pa.y_positions[0] - cy) <= plan.row_spacing + + +def test_middle_row_index_rule(): + assert [middle_row_index(n) for n in (1, 2, 3, 4, 6)] == [0, 1, 1, 2, 3] + + +def test_middle_row_plan_rejects_an_empty_plan(): + plan = full_plan(num_angles=1) + plan.per_angle = [] + with pytest.raises(ScanGeometryError, match="no angles"): + middle_row_plan(plan) + + +def test_middle_row_plan_rejects_an_angle_with_no_rows(): + plan = full_plan(num_angles=1) + plan.per_angle[0].y_positions = [] + with pytest.raises(ScanGeometryError, match="no middle row"): + middle_row_plan(plan) + + +# ── The v10 file ───────────────────────────────────────────────────────────── + +def test_v10_write_read_roundtrip(tmp_path): + out = tmp_path / "check.sras" + plan = write_check(out, bins=(8, 8, 8)) + + sras = SrasFile(out) + assert sras.version == VERSION_SAW_CHECK + assert sras.is_saw_check + assert [s.status for s in sras.angle_status()] == ["OK"] * plan.n_angles + assert [pa.n_rows for pa in sras.per_angle] == [1] * plan.n_angles + sras.close() + + +def test_v10_rejects_a_multi_row_plan(tmp_path): + plan = full_plan(num_angles=2) + assert any(pa.n_rows > 1 for pa in plan.per_angle) + with pytest.raises(ValueError, match="exactly one row per angle"): + create_scan_file(tmp_path / "bad.sras", plan, SPF, SAMPLE_RATE, + PREAMBLES, bytes(SPF), version=VERSION_SAW_CHECK) + assert not (tmp_path / "bad.sras").exists() + + +def test_unknown_version_rejected_at_write(tmp_path): + with pytest.raises(ValueError, match="version 7"): + create_scan_file(tmp_path / "bad.sras", middle_row_plan(full_plan(1)), + SPF, SAMPLE_RATE, PREAMBLES, bytes(SPF), version=7) + + +def test_v6_file_is_not_a_saw_check(): + sras = SrasFile("tests/golden/complete.sras") + assert sras.version == VERSION and not sras.is_saw_check + + +# ── Acquisition through the engine ─────────────────────────────────────────── + +def run_engine(tmp_path, num_angles=3): + trace = Trace() + scope = FakeScope(trace, samples_per_frame=SPF) + stage = FakeStage(trace, scope=scope) + rotator = RotationAxis(FakeT3R(trace), RotationSettings()) + plan = full_plan(num_angles) + check = middle_row_plan(plan) + engine = ScanEngine(stage, scope, rotator, check, tmp_path / "check.sras", + callbacks=ScanCallbacks(), + file_version=VERSION_SAW_CHECK) + return engine.run(), plan, check, trace + + +def test_engine_writes_a_complete_v10_check(tmp_path): + result, plan, check, _ = run_engine(tmp_path) + + assert not result.aborted + assert result.rows_written == check.n_angles # exactly one row per angle + assert result.angles_acquired == list(range(check.n_angles)) + + sras = SrasFile(result.path) + assert sras.is_saw_check + assert [s.status for s in sras.angle_status()] == ["OK"] * check.n_angles + assert [pa.y_positions for pa in sras.per_angle] == [ + [pytest.approx(original.y_positions[original.n_rows // 2], abs=1e-4)] + for original in plan.per_angle + ] + sras.close() + + +def test_engine_visits_each_middle_row_once(tmp_path): + _, _, check, trace = run_engine(tmp_path) + y_moves = [round(c[2], 4) for c in trace.of("move_axis_absolute") + if c[1] == 0x22] + assert y_moves == [round(pa.y_positions[0], 4) for pa in check.per_angle] + + +def test_engine_still_writes_v6_by_default(tmp_path): + trace = Trace() + scope = FakeScope(trace, samples_per_frame=SPF) + stage = FakeStage(trace, scope=scope) + rotator = RotationAxis(FakeT3R(trace), RotationSettings()) + engine = ScanEngine(stage, scope, rotator, full_plan(1), + tmp_path / "scan.sras", callbacks=ScanCallbacks()) + result = engine.run() + assert SrasFile(result.path).version == VERSION + + +# ── Analysis ───────────────────────────────────────────────────────────────── + +def test_traces_report_the_injected_frequency(tmp_path): + out = tmp_path / "check.sras" + bins = (8, 9, 10) + write_check(out, bins=bins) + + with SrasFile(out) as sras: + traces = frequency_traces(sras, dc_threshold_mv=DC_THRESHOLD_MV) + + assert len(traces) == len(bins) + for trace, k in zip(traces, bins, strict=True): + assert np.allclose(trace.freq_mhz, bin_mhz(k)) + assert trace.median_mhz == pytest.approx(bin_mhz(k)) + assert trace.valid_fraction == 1.0 + assert trace.drift_mhz_per_mm == pytest.approx(0.0, abs=1e-6) + + +def test_masked_pixels_become_nan_not_zero(tmp_path): + out = tmp_path / "check.sras" + write_check(out, bins=(8, 8, 8), n_masked_frames=2) + + with SrasFile(out) as sras: + traces = frequency_traces(sras, dc_threshold_mv=DC_THRESHOLD_MV) + + for trace in traces: + assert np.isnan(trace.freq_mhz[:2]).all() + assert np.isfinite(trace.freq_mhz[2:]).all() + # A masked pixel must not drag the median toward 0 MHz. + assert trace.median_mhz == pytest.approx(bin_mhz(8)) + assert trace.valid_fraction < 1.0 + + +def test_traces_are_centred_on_a_common_offset(tmp_path): + out = tmp_path / "check.sras" + write_check(out, bins=(8, 9, 10)) + + with SrasFile(out) as sras: + traces = frequency_traces(sras, dc_threshold_mv=DC_THRESHOLD_MV) + + # Absolute X differs per angle (different bounding boxes); the offset the + # viewer plots against does not, which is what puts the curves together. + assert len({round(t.x_mm[0], 6) for t in traces}) > 1 + for trace in traces: + assert trace.offset_mm[0] == pytest.approx(-trace.offset_mm[-1]) + + +def test_angles_with_no_data_are_skipped(tmp_path): + out = tmp_path / "check.sras" + write_check(out, bins=(8, 8, 8)) + full = out.read_bytes() + with SrasFile(out) as sras: + last_offset = sras.angle_data_offset(2) + out.write_bytes(full[:last_offset]) # angle 3 never acquired + + with SrasFile(out) as sras: + traces = frequency_traces(sras, dc_threshold_mv=DC_THRESHOLD_MV) + assert [t.angle_idx for t in traces] == [0, 1] + + +def test_summary_flags_agreeing_angles_as_good(tmp_path): + out = tmp_path / "check.sras" + write_check(out, bins=(8, 8, 8)) + + with SrasFile(out) as sras: + summary = alignment_summary( + frequency_traces(sras, dc_threshold_mv=DC_THRESHOLD_MV)) + + assert summary.n_angles == 3 + assert summary.median_mhz == pytest.approx(bin_mhz(8)) + assert summary.spread_mhz == pytest.approx(0.0) + assert summary.spread_pct <= SPREAD_GOOD_PCT + assert summary.level == "good" + + +def test_summary_flags_disagreeing_angles(tmp_path): + out = tmp_path / "check.sras" + write_check(out, bins=(8, 9, 10)) + + with SrasFile(out) as sras: + traces = frequency_traces(sras, dc_threshold_mv=DC_THRESHOLD_MV) + summary = alignment_summary(traces) + + assert summary.spread_mhz == pytest.approx(bin_mhz(10) - bin_mhz(8)) + assert summary.level == "poor" + assert summary.worst_angle_deg == traces[0].angle_deg # lowest median + assert summary.best_angle_deg == traces[2].angle_deg # highest median + assert f"{summary.spread_mhz:.3f} MHz" in summary.describe() + + +def test_summary_calls_out_a_mostly_masked_row(tmp_path): + out = tmp_path / "check.sras" + plan = middle_row_plan(full_plan(3)) + # Mask nearly every frame of every angle: the spread is meaningless then. + write_check(out, bins=(8, 8, 8), plan=plan, + n_masked_frames=max(pa.n_frames for pa in plan.per_angle) - 1) + + with SrasFile(out) as sras: + summary = alignment_summary( + frequency_traces(sras, dc_threshold_mv=DC_THRESHOLD_MV)) + + assert summary.level == "poor" + assert "DC threshold" in summary.describe() + + +def test_summary_of_nothing_is_not_a_crash(): + summary = alignment_summary([]) + assert summary.n_angles == 0 and summary.level == "poor" + assert "No angle" in summary.describe() + + +def test_middle_row_of_a_full_v6_scan_is_readable(): + """The check's read-out applied to a finished scan, after the fact.""" + with SrasFile("tests/golden/complete.sras") as sras: + traces = frequency_traces(sras, dc_threshold_mv=-1e6) + assert len(traces) == sras.header.n_angles + for trace, pa in zip(traces, sras.per_angle, strict=True): + assert trace.row_idx == pa.n_rows // 2 + assert len(trace.freq_mhz) == pa.n_frames diff --git a/tests/test_smoke_apps.py b/tests/test_smoke_apps.py index 452be72..b995501 100644 --- a/tests/test_smoke_apps.py +++ b/tests/test_smoke_apps.py @@ -49,6 +49,17 @@ def test_sras_viewer_window(qapp): _pump(qapp) +def test_saw_check_viewer_window(qapp): + import saw_check_viewer + win = saw_check_viewer.SawCheckWindow() + _pump(qapp) + try: + assert win.windowTitle() + finally: + win.deleteLater() + _pump(qapp) + + def test_helios_test_app(qapp): import helios_test_app win = helios_test_app.HeliosTestApp()