2 Commits

Author SHA1 Message Date
Thomas Ales 844fcd0297 SAW quality check: one middle row per angle, and a viewer that overlays them
A full multi-angle scan takes hours, and a rig whose angles disagree produces
all of them before anyone finds out. This adds a test mode that acquires one
row per angle — the row-wise middle of the ROI — and a viewer that puts every
angle's SAW frequency on one graph. The default 80×50 mm ROI at 5 angles goes
from 1461 rows to 5.

Why the middle row answers an alignment question at all: 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. All the angles
measure the same material, so a spread in their frequencies belongs to the rig
rather than to where each row happened to land. test_every_angles_middle_row_
crosses_the_roi_centre pins that premise, since the whole comparison rests on
it and nothing else in the geometry code would notice it breaking.

core/saw_check.py — both halves of the mode, kept together because neither is
much use alone. middle_row_plan() reduces a ScanPlan to one row per angle
(n_rows // 2, the upper of two centre rows when even); frequency_traces() and
alignment_summary() turn the resulting file back into per-angle frequency
traces and the scalars an operator is actually asking about — the spread of
the per-angle medians, the worst drift along a row, the sparsest row. The
verdict thresholds are labelled as rules of thumb, not physics: an anisotropic
sample genuinely varies with angle, so a wide spread is a prompt to look at
the curves rather than a verdict.

Format v10: byte-identical to v6, one row per angle. The version byte earns
its keep because the two are otherwise indistinguishable — a v6 scan aborted
after its first row is not a check, and a reader guessing from the row count
would read a failed scan as a deliberate measurement. create_scan_file()
enforces the one-row rule at write time, since nothing downstream can recover
from a v10 file that breaks it. ScanEngine gains file_version and is otherwise
untouched: the acquisition, the abort/pause path and the background capture
are the scan's, unchanged.

sras_scan_manager.py now carries the source file's version through an export
instead of stamping v6 on everything, which the wider reader would otherwise
have made a lie.

saw_check_viewer.py — frequency along the row, one curve per angle, over a
common offset axis so the curves lie on the same piece of sample; a summary of
each angle's median ±1σ against angle; and the per-angle numbers in a table.
Analysis parameters (DC threshold, background, time gate) recompute on a
worker thread; display ones (smoothing, axis, MHz↔m/s) only redraw. A full v6
scan opens too — the same middle row is pulled out of it — so a finished scan
can be re-examined with the check's own read-out.

In the app, a check finishes by handing the operator the file and an "Open
Viewer" button rather than shutting the rig down the way a completed scan
does. Burst mode is not offered: one row per angle means every burst would be
a single row, so it buys nothing and still pays for the gate preflight.

137 tests passing, ruff clean.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-09-04 08:13:54 -05:00
Thomas Ales dfd6c9e2b8 Merge dev-angle-inspect: pre-scan angle inspection
Walk a plan's angles before committing to the run, parking the rig at a random
point in each so the SAW response can be judged on the oscilloscope. A weak
angle otherwise produces rows that look structurally fine in the file with no
usable packet in them, which is only discoverable hours later.

The app configures the scope and drives the motion; it never reads a waveform
back. That is the feature's premise rather than an omission, so a test asserts
no transfer path exists.

- core/scope_inspect.py  free-running edge trigger on CH2 at 2.0 V, FastFrame
  and averaging off, CH1 on the acquisition front-end, CH3/CH4 rescaled as
  bias monitors sharing one scale and position
- core/angle_inspect.py  headless AngleInspector; points land on the angle's
  own scan grid, and New Point re-rolls without rotating
- gui/inspect_bridge.py  QtAngleInspector on the QueueWorker base
- sc3_aui_app.py         AngleInspectWindow, driven off the entered plan

The bias scaling (100 mV/div, ground 3.5 divisions low) is derived to fit
0-700 mV on an 8- or 10-division graticule, not measured on the rig; expect to
tune BIAS_POSITION_DIV against the bench-tuned values in SRAS_CHANNELS.

114 tests passing, ruff clean.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-09-04 07:21:56 -05:00
11 changed files with 1541 additions and 28 deletions
+36 -1
View File
@@ -14,6 +14,9 @@ scanengine-3 is a unified platform for scanning acoustic microscopy and precisio
- **Scan Planning**: Automated raster scan generation and execution - **Scan Planning**: Automated raster scan generation and execution
- **Angle Inspection**: Park the rig at random points across a plan's angles - **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 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 - **Real-time Monitoring**: Live status updates and progress tracking
## Hardware Components ## Hardware Components
@@ -60,8 +63,9 @@ scanengine-3/
│ ├── scope_burst.py # Burst-mode FastFrame sizing + row splitting │ ├── scope_burst.py # Burst-mode FastFrame sizing + row splitting
│ ├── scope_inspect.py # Scope setup for pre-scan angle inspection │ ├── scope_inspect.py # Scope setup for pre-scan angle inspection
│ ├── angle_inspect.py # AngleInspector — park on a point per angle │ ├── 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 │ ├── 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 │ ├── sras_analysis.py # Image reducers + SAW matched filter
│ └── config.py # ScanDefaults ⇄ aui_defaults.json │ └── config.py # ScanDefaults ⇄ aui_defaults.json
│ │
@@ -84,6 +88,7 @@ scanengine-3/
│ │
├── sc3_aui_app.py # Main acquisition application ├── sc3_aui_app.py # Main acquisition application
├── sras_viewer.py # Scan data viewer ├── 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 ├── sras_scan_manager.py # CLI: inspect/export/delete angles
├── t3r_control_panel.py # T3R panel (used by the main app) ├── t3r_control_panel.py # T3R panel (used by the main app)
├── helios_test_app.py # Per-device test benches ├── helios_test_app.py # Per-device test benches
@@ -133,6 +138,9 @@ python sc3_aui_app.py
# Scan data viewer # Scan data viewer
python sras_viewer.py 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 # Inspect / export / delete angles in a .sras file
python sras_scan_manager.py path/to/scan.sras 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}") 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 ### Reading a scan file
`SrasFile` memory-maps the data block, so opening a multi-gigabyte scan `SrasFile` memory-maps the data block, so opening a multi-gigabyte scan
+255
View File
@@ -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),
)
+9 -2
View File
@@ -18,7 +18,7 @@ from typing import Callable
from core import scope_burst, scope_sras from core import scope_burst, scope_sras
from core.rotation import RotationAxis from core.rotation import RotationAxis
from core.scan_geometry import ScanPlan, validate_plan 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__) logger = logging.getLogger(__name__)
@@ -96,7 +96,8 @@ class ScanEngine:
plan: ScanPlan, out_path: Path, plan: ScanPlan, out_path: Path,
resume: ResumeState | None = None, resume: ResumeState | None = None,
callbacks: ScanCallbacks | 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._stage = stage
self._scope = scope self._scope = scope
self._rotator = rotator self._rotator = rotator
@@ -110,6 +111,11 @@ class ScanEngine:
# Strict row packing stops the scan on a frame-count mismatch # Strict row packing stops the scan on a frame-count mismatch
# instead of squaring the row up (see _check_frame_delta). # instead of squaring the row up (see _check_frame_delta).
self._strict_rows = strict_rows 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._max_frames = 0
self._preflight_done = False self._preflight_done = False
@@ -336,6 +342,7 @@ class ScanEngine:
return create_scan_file( return create_scan_file(
self._out_path, self._plan, samples_per_frame, self._out_path, self._plan, samples_per_frame,
scope_sras.SAMPLE_RATE_HZ, self._preambles, self._background, 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): def _scan_loop(self, scan_file, samples_per_frame: int, result: ScanResult):
+45 -7
View File
@@ -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: 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 Incomplete files are valid: the data block is one contiguous append-only
stream, so the readable prefix defines a single frontier past which nothing stream, so the readable prefix defines a single frontier past which nothing
has been written yet (see ``SrasFile.angle_status``). 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 from __future__ import annotations
@@ -32,6 +38,9 @@ from core.scan_geometry import AngleGeometry, ScanPlan
MAGIC = b"SRAS" MAGIC = b"SRAS"
VERSION = 6 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_FMT = ">4sBHfffffffIdBB"
HDR_SIZE = struct.calcsize(HDR_FMT) # 49 bytes HDR_SIZE = struct.calcsize(HDR_FMT) # 49 bytes
GEOM_FMT = ">ffIH" GEOM_FMT = ">ffIH"
@@ -80,16 +89,38 @@ class AngleStatus:
def create_scan_file(path: Path, plan: ScanPlan, samples_per_frame: int, def create_scan_file(path: Path, plan: ScanPlan, samples_per_frame: int,
sample_rate: float, preambles: list[str], sample_rate: float, preambles: list[str],
background_waveform: bytes) -> BinaryIO: background_waveform: bytes,
"""Create a new .sras file and write the v6 header + tables. 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; Returns an open binary file positioned at the start of the data block;
the caller appends waveform rows and must close it (try/finally). 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) path.parent.mkdir(parents=True, exist_ok=True)
f = open(path, "wb") f = open(path, "wb")
f.write(struct.pack( f.write(struct.pack(
HDR_FMT, MAGIC, VERSION, HDR_FMT, MAGIC, version,
plan.n_angles, plan.n_angles,
plan.x_start_nominal, plan.y_start_nominal, plan.x_start_nominal, plan.y_start_nominal,
plan.x_delta_nominal, plan.y_delta_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 @dataclass
class SrasFile: 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 Parsing reads only the header/tables — never the waveform block — so
opening a multi-GB file is cheap. ``load_angle``/``load_row`` return opening a multi-GB file is cheap. ``load_angle``/``load_row`` return
@@ -124,6 +155,7 @@ class SrasFile:
the caller computes on it. the caller computes on it.
""" """
path: Path path: Path
version: int = field(init=False)
header: ScanHeader = field(init=False) header: ScanHeader = field(init=False)
per_angle: list[AngleGeometry] = field(init=False) per_angle: list[AngleGeometry] = field(init=False)
preambles: list[str] = field(init=False) preambles: list[str] = field(init=False)
@@ -149,11 +181,12 @@ class SrasFile:
n_channels) = struct.unpack(HDR_FMT, raw) n_channels) = struct.unpack(HDR_FMT, raw)
if magic != MAGIC: if magic != MAGIC:
raise ValueError(f"{self.path.name}: not a valid SRAS file (bad 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( raise ValueError(
f"{self.path.name}: unsupported SRAS format version {version} " 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( self.header = ScanHeader(
n_angles=n_angles, n_angles=n_angles,
x_start_nominal=x_start_nominal, y_start_nominal=y_start_nominal, x_start_nominal=x_start_nominal, y_start_nominal=y_start_nominal,
@@ -187,6 +220,11 @@ class SrasFile:
self.data_start_offset = f.tell() 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 ─────────────────────────────────────── # ── Frontier / truncation analysis ───────────────────────────────────────
def row_bytes(self, angle_idx: int) -> int: def row_bytes(self, angle_idx: int) -> int:
+674
View File
@@ -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()
+11
View File
@@ -1017,6 +1017,16 @@
</property> </property>
</widget> </widget>
</item> </item>
<item>
<widget class="QPushButton" name="saw_check_btn">
<property name="toolTip">
<string>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.</string>
</property>
<property name="text">
<string>SAW Quality Check…</string>
</property>
</widget>
</item>
<item> <item>
<widget class="QPushButton" name="inspect_angles_btn"> <widget class="QPushButton" name="inspect_angles_btn">
<property name="toolTip"> <property name="toolTip">
@@ -1086,6 +1096,7 @@
<tabstop>bbd_set_current_start_btn</tabstop> <tabstop>bbd_set_current_start_btn</tabstop>
<tabstop>bbd_set_delta_current_btn</tabstop> <tabstop>bbd_set_delta_current_btn</tabstop>
<tabstop>show_camera_toggle</tabstop> <tabstop>show_camera_toggle</tabstop>
<tabstop>saw_check_btn</tabstop>
<tabstop>start_scan_btn</tabstop> <tabstop>start_scan_btn</tabstop>
</tabstops> </tabstops>
<resources/> <resources/>
+111 -8
View File
@@ -6,6 +6,7 @@ and wires up T3R, BBD202, oscilloscope, and camera hardware workers.
""" """
import struct import struct
import subprocess
import sys import sys
import time import time
from pathlib import Path from pathlib import Path
@@ -32,10 +33,13 @@ from core.scan_engine import (
ResumeState, ResumeState,
LASER_FREQ_HZ, SCAN_VELOCITY_MM_S, 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_geometry import ScanPlan, EtaEstimator, build_plan, format_eta
from core.scan_resume import is_compatible, plan_resume from core.scan_resume import is_compatible, plan_resume
from core.scope_sras import SAMPLE_RATE_HZ, configure_channels 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_t3r import QtT3RAdapter
from gui.qt_workers import PollingQueueWorker, QueueWorker from gui.qt_workers import PollingQueueWorker, QueueWorker
from gui.inspect_bridge import QtAngleInspector from gui.inspect_bridge import QtAngleInspector
@@ -51,6 +55,10 @@ from t3r_control_panel import T3RControlPanel
DEFAULTS = ScanDefaults.load() DEFAULTS = ScanDefaults.load()
BBD_DEFAULT_JOG_MM = 0.5 # default jog step for BBD202 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): class DCBiasImageWidget(FigureCanvas):
@@ -977,6 +985,11 @@ class MainWindow(QMainWindow):
self._scan_progress = ScanProgressWindow() self._scan_progress = ScanProgressWindow()
self._scan_worker: QtScanController | None = None 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_thread: QThread | None = None
self._inspect_worker: QtAngleInspector | None = None self._inspect_worker: QtAngleInspector | None = None
self._inspect_window: AngleInspectWindow | None = None self._inspect_window: AngleInspectWindow | None = None
@@ -1101,6 +1114,7 @@ class MainWindow(QMainWindow):
# Scan # Scan
self.start_scan_btn.clicked.connect(self._on_start_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.inspect_angles_btn.clicked.connect(self._on_inspect_angles)
self.save_dir_browse_btn.clicked.connect(self._on_browse_save_dir) self.save_dir_browse_btn.clicked.connect(self._on_browse_save_dir)
self._scan_progress.abort_requested.connect(self._on_abort_scan) self._scan_progress.abort_requested.connect(self._on_abort_scan)
@@ -1267,6 +1281,11 @@ class MainWindow(QMainWindow):
# ── Scan ────────────────────────────────────────────────────────────────── # ── 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): def _on_browse_save_dir(self):
d = QFileDialog.getExistingDirectory( d = QFileDialog.getExistingDirectory(
self, "Select Scan Save Directory", self.scan_save_dir_edit.text() 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.scan_save_dir_edit.setText(d)
self._persist_defaults() 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): def _on_inspect_angles(self):
"""Open the pre-scan angle inspector for the plan currently entered.""" """Open the pre-scan angle inspector for the plan currently entered."""
if self._scan_thread is not None and self._scan_thread.isRunning(): 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)) lambda m: QMessageBox.warning(self, "Inspection Error", m))
self._inspect_thread.started.connect(self._inspect_worker.run) 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) self.inspect_angles_btn.setEnabled(False)
window.show() window.show()
self._inspect_thread.start() self._inspect_thread.start()
@@ -1345,7 +1440,7 @@ class MainWindow(QMainWindow):
self._inspect_thread = None self._inspect_thread = None
self._inspect_worker = None self._inspect_worker = None
self._inspect_window = None self._inspect_window = None
self.start_scan_btn.setEnabled(True) self._set_scan_buttons_enabled(True)
self.inspect_angles_btn.setEnabled(True) self.inspect_angles_btn.setEnabled(True)
def _on_start_scan(self): def _on_start_scan(self):
@@ -1424,8 +1519,10 @@ class MainWindow(QMainWindow):
str(path.parent), resume_plan.to_state(sras)) str(path.parent), resume_plan.to_state(sras))
def _launch_scan_worker(self, plan: ScanPlan, prefix: str, save_dir: str, 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) rotator = RotationAxis(self._t3r_driver.driver, DEFAULT_ROTATION)
self._scan_is_saw_check = saw_check
self._scan_thread = QThread(self) self._scan_thread = QThread(self)
self._scan_worker = QtScanController( self._scan_worker = QtScanController(
@@ -1439,8 +1536,9 @@ class MainWindow(QMainWindow):
# a worker concern, not the engine's. # a worker concern, not the engine's.
on_scan_active=lambda active: setattr( on_scan_active=lambda active: setattr(
self._bbd_worker, "scanning_active", active), 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(), 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_worker.moveToThread(self._scan_thread)
self._scan_thread.started.connect(self._scan_worker.run) 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.user_prompt.connect(self._on_scan_user_prompt)
self._scan_worker.paused_changed.connect(self._scan_progress.on_worker_paused) 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() self._scan_progress.reset_pause_btn()
ai0 = 0 if resume is None else resume.targets[0].angle_idx ai0 = 0 if resume is None else resume.targets[0].angle_idx
self._scan_progress.update_progress( self._scan_progress.update_progress(
@@ -1494,7 +1592,11 @@ class MainWindow(QMainWindow):
def _on_scan_complete(self): def _on_scan_complete(self):
self._scan_progress.close() 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( QMessageBox.information(
self, "Scan Complete", self, "Scan Complete",
"All rows and angles have been acquired.\n\n" "All rows and angles have been acquired.\n\n"
@@ -1504,7 +1606,8 @@ class MainWindow(QMainWindow):
def _on_scan_failed(self, msg: str): def _on_scan_failed(self, msg: str):
self._scan_progress.close() 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(): if "aborted" in msg.lower():
QMessageBox.warning(self, "Scan Aborted", msg) QMessageBox.warning(self, "Scan Aborted", msg)
else: else:
+46 -3
View File
@@ -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 Each `.sras` file contains **one complete scan**: all GR rotation angles and all
Y rows. Files are named `{prefix}.sras`. 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 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 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 `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 | | Offset | Size | Type | Field | Description |
|--------|------|-----------|--------------------|--------------------------------------------------| |--------|------|-----------|--------------------|--------------------------------------------------|
| 0 | 4 | `4s` | `magic` | Always `SRAS` (0x53 0x52 0x41 0x53) | | 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 | | 5 | 2 | `uint16` | `n_angles` | Number of GR rotation angles |
| 7 | 4 | `float32` | `x_start_nominal` | Nominal (pre-rotation) X scan start, mm | | 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 | | 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 History
| Version | Change | | 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. | | 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) | | 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). | | 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`. |
+11 -7
View File
@@ -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 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 the file in place — both operations rewrite the angle/geometry/row tables
and stream-copy only the selected angles' waveform data, producing a file 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. (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 import argparse
@@ -24,7 +26,7 @@ from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent)) 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 @dataclass
@@ -48,7 +50,7 @@ class AngleEntry:
class SrasScanFile: 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): def __init__(self, path: Path):
self.path = Path(path) self.path = Path(path)
@@ -57,6 +59,7 @@ class SrasScanFile:
def _parse(self): def _parse(self):
sras = SrasFile(self.path) sras = SrasFile(self.path)
h = sras.header h = sras.header
self.version = sras.version
self.x_start_nominal = h.x_start_nominal self.x_start_nominal = h.x_start_nominal
self.y_start_nominal = h.y_start_nominal self.y_start_nominal = h.y_start_nominal
self.x_delta_nominal = h.x_delta_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: 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 (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 angles that were truncated on disk and thus exported with fewer rows
than declared). 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] selected = [sf.get(i) for i in indices]
header = struct.pack( 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_start_nominal, sf.y_start_nominal,
sf.x_delta_nominal, sf.y_delta_nominal, sf.x_delta_nominal, sf.y_delta_nominal,
sf.row_spacing_mm, sf.velocity_mm_s, sf.laser_freq_hz, 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): def print_summary(sf: SrasScanFile, selected: set):
print() 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} " 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"y_start={sf.y_start_nominal:.4f} y_delta={sf.y_delta_nominal:.4f} mm "
f"row_spacing={sf.row_spacing_mm:.4f} mm") f"row_spacing={sf.row_spacing_mm:.4f} mm")
+332
View File
@@ -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
+11
View File
@@ -49,6 +49,17 @@ def test_sras_viewer_window(qapp):
_pump(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): def test_helios_test_app(qapp):
import helios_test_app import helios_test_app
win = helios_test_app.HeliosTestApp() win = helios_test_app.HeliosTestApp()