Compare commits
4 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 844fcd0297 | |||
| dfd6c9e2b8 | |||
| 23546a03f7 | |||
| 817da0160c |
@@ -12,6 +12,11 @@ scanengine-3 is a unified platform for scanning acoustic microscopy and precisio
|
||||
- **Laser Systems**: Helios pulsed laser and Genesis CW laser control
|
||||
- **Data Acquisition**: Tektronix oscilloscope integration with fast-frame support
|
||||
- **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
|
||||
@@ -55,8 +60,12 @@ scanengine-3/
|
||||
│ ├── scan_geometry.py # ScanPlan, rotated-bbox planning, limits
|
||||
│ ├── scan_resume.py # Resume planning (frontier rule)
|
||||
│ ├── scope_sras.py # Oscilloscope SCPI policy for SRAS
|
||||
│ ├── 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
|
||||
│
|
||||
@@ -72,12 +81,14 @@ scanengine-3/
|
||||
│
|
||||
├── gui/ # Shared PyQt6 layer
|
||||
│ ├── scan_bridge.py # QtScanController over core.scan_engine
|
||||
│ ├── inspect_bridge.py # QtAngleInspector over core.angle_inspect
|
||||
│ ├── qt_t3r.py # Qt adapter over the T3R driver
|
||||
│ ├── qt_workers.py # QueueWorker / PollingQueueWorker bases
|
||||
│ └── widgets.py # ConnectionBar, LogConsole, PortSelector…
|
||||
│
|
||||
├── 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
|
||||
@@ -127,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
|
||||
|
||||
@@ -202,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
|
||||
|
||||
@@ -0,0 +1,251 @@
|
||||
"""Pre-scan angle inspection: park the rig on a point and let the operator look.
|
||||
|
||||
A multi-angle scan can take hours, and an angle that responds poorly produces
|
||||
rows that look fine in the file but carry no usable SAW packet. This drives
|
||||
the rig through the same angles the scan will use, parking at a random point
|
||||
inside each angle's own bounding box so the response can be judged on the
|
||||
oscilloscope before committing to the run.
|
||||
|
||||
Headless and Qt-free, like ScanEngine: gui/inspect_bridge.py wraps it.
|
||||
|
||||
No waveform ever crosses this boundary. The operator reads the scope screen
|
||||
directly; this module's job is only to put the hardware in the right place and
|
||||
the scope in a state worth looking at (see core.scope_inspect).
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import random
|
||||
from dataclasses import dataclass
|
||||
from typing import Callable
|
||||
|
||||
from core import scope_inspect
|
||||
from core.rotation import RotationAxis
|
||||
from core.scan_engine import (
|
||||
AXIS_X, AXIS_Y, SCAN_ACCEL_MM_S2, SCAN_VELOCITY_MM_S,
|
||||
)
|
||||
from core.scan_geometry import DEFAULT_STAGE_LIMITS, ScanPlan, StageLimits
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# Positioning moves only — no data is taken while moving, so there is no
|
||||
# reason to cross the tray at full scan velocity.
|
||||
INSPECT_VELOCITY_MM_S = SCAN_VELOCITY_MM_S / 2.0
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class InspectionPoint:
|
||||
"""Where the rig is parked, and which angle it is parked for."""
|
||||
angle_idx: int
|
||||
angle_deg: float
|
||||
x_mm: float
|
||||
y_mm: float
|
||||
|
||||
def describe(self) -> str:
|
||||
return (f"Angle {self.angle_idx + 1} ({self.angle_deg:.1f}°) "
|
||||
f"X={self.x_mm:.3f} mm Y={self.y_mm:.3f} mm")
|
||||
|
||||
|
||||
@dataclass
|
||||
class InspectCallbacks:
|
||||
"""Progress reporting. Defaults are no-ops so the core needs no front end."""
|
||||
on_status: Callable[[str], None] = lambda msg: None
|
||||
on_point: Callable[[InspectionPoint], None] = lambda pt: None
|
||||
on_busy: Callable[[bool], None] = lambda busy: None
|
||||
|
||||
|
||||
@dataclass
|
||||
class _State:
|
||||
angle_idx: int = 0
|
||||
point: InspectionPoint | None = None
|
||||
started: bool = False
|
||||
rotator_ready: bool = False
|
||||
|
||||
|
||||
class AngleInspector:
|
||||
"""Drives stage + rotator to inspection points across a plan's angles."""
|
||||
|
||||
def __init__(self, stage, scope, rotator: RotationAxis | None,
|
||||
plan: ScanPlan,
|
||||
callbacks: InspectCallbacks | None = None,
|
||||
limits: StageLimits = DEFAULT_STAGE_LIMITS,
|
||||
rng: random.Random | None = None):
|
||||
self._stage = stage
|
||||
self._scope = scope
|
||||
self._rotator = rotator
|
||||
self._plan = plan
|
||||
self._cb = callbacks if callbacks is not None else InspectCallbacks()
|
||||
self._limits = limits
|
||||
# Injectable so tests can pin the point selection.
|
||||
self._rng = rng if rng is not None else random.Random()
|
||||
self._st = _State()
|
||||
|
||||
# ── Introspection ─────────────────────────────────────────────────────────
|
||||
|
||||
@property
|
||||
def n_angles(self) -> int:
|
||||
return self._plan.n_angles
|
||||
|
||||
@property
|
||||
def angle_idx(self) -> int:
|
||||
return self._st.angle_idx
|
||||
|
||||
@property
|
||||
def current_point(self) -> InspectionPoint | None:
|
||||
return self._st.point
|
||||
|
||||
def angle_labels(self) -> list[str]:
|
||||
return [f"Angle {i + 1}/{self.n_angles} — {pa.angle_deg:.2f}°"
|
||||
for i, pa in enumerate(self._plan.per_angle)]
|
||||
|
||||
# ── Lifecycle ─────────────────────────────────────────────────────────────
|
||||
|
||||
def start(self) -> InspectionPoint:
|
||||
"""Configure the hardware and park on the first angle."""
|
||||
if self._stage is None:
|
||||
raise RuntimeError("BBD202 not connected")
|
||||
if self._scope is None:
|
||||
raise RuntimeError("Oscilloscope not connected")
|
||||
|
||||
self._st.rotator_ready = (self._rotator is not None
|
||||
and self._rotator.is_available)
|
||||
if self.n_angles > 1 and not self._st.rotator_ready:
|
||||
raise RuntimeError(
|
||||
f"Inspecting {self.n_angles} angles requires the T3R rotation "
|
||||
"stage (GR-axis), but it is not connected. Connect T3R from "
|
||||
"the T3R panel, or inspect a single-angle plan."
|
||||
)
|
||||
|
||||
self._cb.on_busy(True)
|
||||
try:
|
||||
self._cb.on_status("Configuring stage for inspection …")
|
||||
ctrl = self._stage
|
||||
for axis in (AXIS_X, AXIS_Y):
|
||||
ctrl.set_velocity_params(axis,
|
||||
max_velocity=INSPECT_VELOCITY_MM_S,
|
||||
acceleration=SCAN_ACCEL_MM_S2)
|
||||
# Nothing here is gated, and an armed trigger output would keep
|
||||
# driving the gate line on every positioning move.
|
||||
ctrl.set_trigger_gate_off(AXIS_X)
|
||||
|
||||
if self._st.rotator_ready:
|
||||
self._cb.on_status("Configuring GR axis …")
|
||||
self._rotator.configure()
|
||||
|
||||
self._cb.on_status("Configuring oscilloscope for inspection …")
|
||||
scope_inspect.configure_inspection(self._scope)
|
||||
|
||||
self._st.started = True
|
||||
return self._goto(0, new_point=True)
|
||||
finally:
|
||||
self._cb.on_busy(False)
|
||||
|
||||
def stop(self) -> None:
|
||||
"""Stop the sweep and send the rotator home. Safe to call twice."""
|
||||
if not self._st.started:
|
||||
return
|
||||
self._st.started = False
|
||||
self._cb.on_busy(True)
|
||||
try:
|
||||
try:
|
||||
scope_inspect.stop_inspection(self._scope)
|
||||
except Exception:
|
||||
logger.exception("Could not stop the inspection acquisition")
|
||||
if self._st.rotator_ready and abs(self._rotator.current_deg) > 0.001:
|
||||
self._cb.on_status("Returning GR to home …")
|
||||
try:
|
||||
self._rotator.return_to_zero()
|
||||
except Exception:
|
||||
logger.exception("GR return-to-home failed")
|
||||
self._cb.on_status("Inspection finished.")
|
||||
finally:
|
||||
self._cb.on_busy(False)
|
||||
|
||||
# ── Navigation ────────────────────────────────────────────────────────────
|
||||
|
||||
def goto_angle(self, angle_idx: int) -> InspectionPoint:
|
||||
"""Rotate to `angle_idx` and park on a fresh random point there."""
|
||||
self._require_started()
|
||||
self._cb.on_busy(True)
|
||||
try:
|
||||
return self._goto(angle_idx, new_point=True)
|
||||
finally:
|
||||
self._cb.on_busy(False)
|
||||
|
||||
def next_angle(self) -> InspectionPoint:
|
||||
"""Advance one angle, wrapping at the end."""
|
||||
return self.goto_angle((self._st.angle_idx + 1) % self.n_angles)
|
||||
|
||||
def prev_angle(self) -> InspectionPoint:
|
||||
return self.goto_angle((self._st.angle_idx - 1) % self.n_angles)
|
||||
|
||||
def new_point(self) -> InspectionPoint:
|
||||
"""Re-roll the point within the current angle, without rotating.
|
||||
|
||||
One point can be unrepresentative — a bad spot on the sample looks the
|
||||
same as a bad angle. Re-rolling a few times is how you tell them
|
||||
apart, so this deliberately skips the rotation.
|
||||
"""
|
||||
self._require_started()
|
||||
self._cb.on_busy(True)
|
||||
try:
|
||||
return self._goto(self._st.angle_idx, new_point=True, rotate=False)
|
||||
finally:
|
||||
self._cb.on_busy(False)
|
||||
|
||||
# ── Internals ─────────────────────────────────────────────────────────────
|
||||
|
||||
def _require_started(self):
|
||||
if not self._st.started:
|
||||
raise RuntimeError("Inspection has not been started")
|
||||
|
||||
def _goto(self, angle_idx: int, new_point: bool,
|
||||
rotate: bool = True) -> InspectionPoint:
|
||||
if not 0 <= angle_idx < self.n_angles:
|
||||
raise IndexError(
|
||||
f"Angle {angle_idx} out of range (plan has {self.n_angles})")
|
||||
|
||||
pa = self._plan.per_angle[angle_idx]
|
||||
self._st.angle_idx = angle_idx
|
||||
|
||||
if rotate and self._st.rotator_ready:
|
||||
delta = pa.angle_deg - self._rotator.current_deg
|
||||
if abs(delta) > 0.001:
|
||||
self._cb.on_status(
|
||||
f"Rotating GR to {pa.angle_deg:.1f}° (Δ{delta:+.1f}°) …")
|
||||
self._rotator.rotate_to(pa.angle_deg)
|
||||
|
||||
point = self._pick_point(angle_idx) if new_point else self._st.point
|
||||
|
||||
self._cb.on_status(f"Moving to {point.describe()} …")
|
||||
# Y first, then X — the same order the scan uses to reach a row.
|
||||
self._stage.move_axis_absolute(AXIS_Y, point.y_mm, timeout=60.0)
|
||||
self._stage.move_axis_absolute(AXIS_X, point.x_mm, timeout=60.0)
|
||||
|
||||
self._st.point = point
|
||||
self._cb.on_point(point)
|
||||
self._cb.on_status(f"Parked at {point.describe()}")
|
||||
return point
|
||||
|
||||
def _pick_point(self, angle_idx: int) -> InspectionPoint:
|
||||
"""A random point on this angle's scan grid.
|
||||
|
||||
Y is drawn from the angle's actual row positions and X uniformly from
|
||||
its data window, so the point is somewhere the scan would really
|
||||
sample — not merely inside the bounding box.
|
||||
"""
|
||||
pa = self._plan.per_angle[angle_idx]
|
||||
if not pa.y_positions:
|
||||
raise ValueError(f"Angle {angle_idx + 1} has no rows to inspect")
|
||||
|
||||
y = self._rng.choice(pa.y_positions)
|
||||
x = self._rng.uniform(pa.x_start, pa.x_start + pa.x_delta)
|
||||
|
||||
lim = self._limits
|
||||
if not (lim.x_min <= x <= lim.x_max and lim.y_min <= y <= lim.y_max):
|
||||
raise ValueError(
|
||||
f"Inspection point X={x:.3f} Y={y:.3f} is outside the stage "
|
||||
f"travel ({lim.x_min}–{lim.x_max} × {lim.y_min}–{lim.y_max} mm)"
|
||||
)
|
||||
return InspectionPoint(angle_idx=angle_idx, angle_deg=pa.angle_deg,
|
||||
x_mm=x, y_mm=y)
|
||||
@@ -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
@@ -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):
|
||||
|
||||
@@ -0,0 +1,104 @@
|
||||
"""Oscilloscope configuration for pre-scan angle inspection.
|
||||
|
||||
Inspection is read-on-the-instrument: nothing in this module transfers or
|
||||
plots waveform data. The app puts the scope into a free-running, edge-
|
||||
triggered state and drives the stage to the point being inspected; the
|
||||
operator judges the SAW response and the bias levels on the scope screen.
|
||||
|
||||
That split is deliberate. A scan's acquisition trigger is the logic AND of
|
||||
the laser pulse and the stage's max-velocity gate, and its transfers are
|
||||
FastFrame blocks — neither is useful for looking at one point by eye. Here
|
||||
the trigger is a plain edge on the laser pulse, FastFrame is off, and the
|
||||
acquisition free-runs, so the display updates continuously while the stage
|
||||
sits still.
|
||||
|
||||
CH1 keeps the acquisition front-end so what is on screen is what a scan would
|
||||
record. CH3 and CH4 are rescaled as DC bias monitors (see BIAS_* below).
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from dataclasses import replace
|
||||
|
||||
from core.scope_sras import SAMPLE_RATE_HZ, SRAS_CHANNELS, configure_channels
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# CH2 carries the laser pulse. The scan triggers it at 0.5 V as one term of a
|
||||
# logic AND; inspection triggers well above that so a slow edge or a noisy
|
||||
# baseline cannot free-run the display.
|
||||
INSPECT_TRIG_LEVEL_V = 2.0
|
||||
|
||||
# CH3/CH4 are the DC bias monitors during inspection. The signal never goes
|
||||
# negative and spans roughly 0–700 mV, so both channels get the *same* scale
|
||||
# and position — the point of inspecting them is comparing the two by eye, and
|
||||
# that only works if a division means the same thing on each.
|
||||
#
|
||||
# Ground sits BIAS_POSITION_DIV divisions below centre, which puts the whole
|
||||
# 0–700 mV range above the centre line with a little room underneath for
|
||||
# undershoot. With 100 mV/div and ground 3.5 divisions low, the visible window
|
||||
# runs from about -50 mV to +750 mV on an 8-division display and wider on a
|
||||
# 10-division one, so 0–700 mV sits comfortably inside either.
|
||||
BIAS_CHANNELS = (3, 4)
|
||||
BIAS_WINDOW_V = 0.700
|
||||
BIAS_SCALE_V_DIV = 0.100
|
||||
BIAS_POSITION_DIV = -3.5
|
||||
|
||||
BIAS_LABELS = {3: "Bias - A", 4: "Bias - B"}
|
||||
|
||||
|
||||
def inspect_channel_profiles() -> dict:
|
||||
"""Channel front-end config for inspection.
|
||||
|
||||
CH1 and CH2 are the acquisition profiles verbatim. CH3 and CH4 differ
|
||||
only in label, scale and position — termination, coupling and bandwidth
|
||||
stay as the scan sets them, so the bias reading is the same measurement
|
||||
the scan records, just displayed usefully.
|
||||
"""
|
||||
profiles = dict(SRAS_CHANNELS)
|
||||
for ch in BIAS_CHANNELS:
|
||||
profiles[ch] = replace(
|
||||
SRAS_CHANNELS[ch],
|
||||
label=BIAS_LABELS[ch],
|
||||
scale_v_div=BIAS_SCALE_V_DIV,
|
||||
position_div=BIAS_POSITION_DIV,
|
||||
)
|
||||
return profiles
|
||||
|
||||
|
||||
def configure_inspection(scope) -> None:
|
||||
"""Put the scope into free-running inspection mode.
|
||||
|
||||
Leaves the acquisition running, so the display stays live while the
|
||||
operator moves between angles and points.
|
||||
"""
|
||||
configure_channels(scope, inspect_channel_profiles())
|
||||
|
||||
# Plain edge trigger on the laser pulse — no logic pattern, so the stage
|
||||
# gate plays no part and a stationary stage still triggers.
|
||||
scope.write("TRIGger:A:TYPe EDGE")
|
||||
scope.set_trigger_source(2)
|
||||
scope.set_trigger_slope("RISE")
|
||||
scope.set_trigger_level(2, INSPECT_TRIG_LEVEL_V)
|
||||
scope.set_trigger_mode("NORMAL")
|
||||
|
||||
# No averaging: a weak or intermittent SAW response is exactly what the
|
||||
# operator is looking for, and averaging would hide it.
|
||||
scope.set_acquire_mode("SAMPLE")
|
||||
scope.set_fastframe_state(False)
|
||||
|
||||
scope.set_sample_rate(SAMPLE_RATE_HZ)
|
||||
scope.write("HORizontal:POSition 30")
|
||||
|
||||
# Free-run rather than single-sequence, so the trace keeps updating.
|
||||
scope.write("ACQuire:STOPAfter RUNSTop")
|
||||
scope.write("ACQuire:STATE RUN")
|
||||
|
||||
|
||||
def stop_inspection(scope) -> None:
|
||||
"""Halt the free-running acquisition.
|
||||
|
||||
The next scan reconfigures the scope from scratch, so this only needs to
|
||||
stop the sweep — it does not try to restore the acquisition profile.
|
||||
"""
|
||||
scope.write("ACQuire:STATE STOP")
|
||||
+45
-7
@@ -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:
|
||||
|
||||
@@ -0,0 +1,113 @@
|
||||
"""Qt bridge over the headless AngleInspector.
|
||||
|
||||
Inspection is command-driven rather than one long run: the operator clicks an
|
||||
angle, waits for the stage to park, looks at the scope, clicks again. That is
|
||||
exactly the shape QueueWorker exists for — it blocks on the queue between
|
||||
commands instead of polling, so an inspection window left open costs nothing.
|
||||
|
||||
Every stage move and rotation blocks for seconds, so all of it runs on this
|
||||
worker's thread; the window only ever enqueues and reacts to signals.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import traceback
|
||||
|
||||
from PyQt6.QtCore import pyqtSignal
|
||||
|
||||
from core.angle_inspect import AngleInspector, InspectCallbacks
|
||||
from gui.qt_workers import QueueWorker
|
||||
|
||||
|
||||
class QtAngleInspector(QueueWorker):
|
||||
"""Runs an AngleInspector on its own QThread and republishes its events."""
|
||||
|
||||
ready = pyqtSignal(object) # InspectionPoint — start() succeeded
|
||||
start_failed = pyqtSignal(str)
|
||||
point_changed = pyqtSignal(object) # InspectionPoint
|
||||
status_msg = pyqtSignal(str)
|
||||
busy_changed = pyqtSignal(bool) # True while a move is in flight
|
||||
stopped = pyqtSignal()
|
||||
|
||||
def __init__(self, stage, scope, rotator, plan, on_inspect_active=None):
|
||||
super().__init__()
|
||||
self._on_inspect_active = on_inspect_active
|
||||
|
||||
callbacks = InspectCallbacks(
|
||||
on_status=self.status_msg.emit,
|
||||
on_point=self.point_changed.emit,
|
||||
on_busy=self.busy_changed.emit,
|
||||
)
|
||||
self._inspector = AngleInspector(stage, scope, rotator, plan,
|
||||
callbacks=callbacks)
|
||||
self._handlers = {
|
||||
"start": self._do_start,
|
||||
"goto": self._do_goto,
|
||||
"new_point": self._do_new_point,
|
||||
"stop": self._do_stop,
|
||||
}
|
||||
|
||||
# ── Introspection (safe from the GUI thread: reads the plan, not the rig) ──
|
||||
|
||||
def angle_labels(self) -> list[str]:
|
||||
return self._inspector.angle_labels()
|
||||
|
||||
@property
|
||||
def n_angles(self) -> int:
|
||||
return self._inspector.n_angles
|
||||
|
||||
# ── Command submission (GUI thread) ───────────────────────────────────────
|
||||
|
||||
def request_start(self):
|
||||
self._enqueue("start")
|
||||
|
||||
def request_goto(self, angle_idx: int):
|
||||
self._enqueue("goto", angle_idx=angle_idx)
|
||||
|
||||
def request_new_point(self):
|
||||
self._enqueue("new_point")
|
||||
|
||||
def request_stop(self):
|
||||
self._enqueue("stop")
|
||||
|
||||
# ── Handlers (worker thread) ──────────────────────────────────────────────
|
||||
|
||||
def _do_start(self):
|
||||
if self._on_inspect_active is not None:
|
||||
self._on_inspect_active(True)
|
||||
try:
|
||||
point = self._inspector.start()
|
||||
except Exception as exc:
|
||||
traceback.print_exc()
|
||||
if self._on_inspect_active is not None:
|
||||
self._on_inspect_active(False)
|
||||
self.start_failed.emit(str(exc))
|
||||
return
|
||||
self.ready.emit(point)
|
||||
|
||||
def _do_goto(self, angle_idx: int):
|
||||
self._inspector.goto_angle(angle_idx)
|
||||
|
||||
def _do_new_point(self):
|
||||
self._inspector.new_point()
|
||||
|
||||
def _do_stop(self):
|
||||
try:
|
||||
self._inspector.stop()
|
||||
finally:
|
||||
if self._on_inspect_active is not None:
|
||||
self._on_inspect_active(False)
|
||||
self.stopped.emit()
|
||||
|
||||
def _on_stop(self):
|
||||
"""Worker loop exiting — make sure the rig is left in a safe state.
|
||||
|
||||
Covers the case where the window is closed without a clean stop
|
||||
command reaching the queue.
|
||||
"""
|
||||
try:
|
||||
self._inspector.stop()
|
||||
except Exception:
|
||||
traceback.print_exc()
|
||||
finally:
|
||||
if self._on_inspect_active is not None:
|
||||
self._on_inspect_active(False)
|
||||
Executable
+674
@@ -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()
|
||||
@@ -1017,6 +1017,26 @@
|
||||
</property>
|
||||
</widget>
|
||||
</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>
|
||||
<widget class="QPushButton" name="inspect_angles_btn">
|
||||
<property name="toolTip">
|
||||
<string>Rotate through the angles of the scan currently entered, parking at a random point in each so the SAW response can be checked on the oscilloscope before committing to the run.</string>
|
||||
</property>
|
||||
<property name="text">
|
||||
<string>Inspect Angles…</string>
|
||||
</property>
|
||||
</widget>
|
||||
</item>
|
||||
<item>
|
||||
<widget class="QPushButton" name="start_scan_btn">
|
||||
<property name="text">
|
||||
@@ -1076,6 +1096,7 @@
|
||||
<tabstop>bbd_set_current_start_btn</tabstop>
|
||||
<tabstop>bbd_set_delta_current_btn</tabstop>
|
||||
<tabstop>show_camera_toggle</tabstop>
|
||||
<tabstop>saw_check_btn</tabstop>
|
||||
<tabstop>start_scan_btn</tabstop>
|
||||
</tabstops>
|
||||
<resources/>
|
||||
|
||||
+285
-6
@@ -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,12 +33,16 @@ 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
|
||||
from gui.scan_bridge import QtScanController
|
||||
from hardware.helios_laser import HeliosLaser
|
||||
from hardware.pybbd202 import AXIS_X, AXIS_Y, ThorlabsServoDriver
|
||||
@@ -50,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):
|
||||
@@ -844,6 +853,104 @@ class ScanProgressWindow(QWidget):
|
||||
|
||||
# ── Main window ───────────────────────────────────────────────────────────────
|
||||
|
||||
class AngleInspectWindow(QWidget):
|
||||
"""Click through a plan's angles, parking the rig at a point in each.
|
||||
|
||||
Deliberately shows no waveform. The operator reads the SAW response and
|
||||
the bias levels off the oscilloscope itself — this window only says where
|
||||
the rig is and lets them move it somewhere else.
|
||||
"""
|
||||
|
||||
goto_requested = pyqtSignal(int)
|
||||
new_point_requested = pyqtSignal()
|
||||
stop_requested = pyqtSignal()
|
||||
|
||||
def __init__(self, angle_labels: list[str], parent: QWidget | None = None):
|
||||
super().__init__(parent, Qt.WindowType.Window)
|
||||
self.setWindowTitle("Inspect Angles")
|
||||
self.resize(420, 460)
|
||||
|
||||
layout = QVBoxLayout(self)
|
||||
layout.addWidget(QLabel(
|
||||
"Select an angle to rotate to it and park at a random point in "
|
||||
"its scan area.\nRead the SAW response on the oscilloscope."
|
||||
))
|
||||
|
||||
self.angle_list = QListWidget(self)
|
||||
for i, label in enumerate(angle_labels):
|
||||
item = QListWidgetItem(label)
|
||||
item.setData(Qt.ItemDataRole.UserRole, i)
|
||||
self.angle_list.addItem(item)
|
||||
self.angle_list.setCurrentRow(0)
|
||||
# currentRowChanged would also fire when the code syncs the highlight
|
||||
# back after a move, re-triggering the move it is reporting.
|
||||
self.angle_list.itemClicked.connect(self._on_item_clicked)
|
||||
layout.addWidget(self.angle_list)
|
||||
|
||||
nav_row = QHBoxLayout()
|
||||
self.prev_btn = QPushButton("◀ Previous")
|
||||
self.next_btn = QPushButton("Next ▶")
|
||||
self.new_point_btn = QPushButton("New Point")
|
||||
self.new_point_btn.setToolTip(
|
||||
"Pick another random point at this angle without rotating — a bad "
|
||||
"spot on the sample looks the same as a bad angle until you move."
|
||||
)
|
||||
self.prev_btn.clicked.connect(self._on_prev)
|
||||
self.next_btn.clicked.connect(self._on_next)
|
||||
self.new_point_btn.clicked.connect(self.new_point_requested.emit)
|
||||
nav_row.addWidget(self.prev_btn)
|
||||
nav_row.addWidget(self.next_btn)
|
||||
nav_row.addWidget(self.new_point_btn)
|
||||
layout.addLayout(nav_row)
|
||||
|
||||
self.point_label = QLabel("—")
|
||||
self.point_label.setStyleSheet("font-weight: bold;")
|
||||
layout.addWidget(self.point_label)
|
||||
|
||||
self.status_label = QLabel("Starting …")
|
||||
self.status_label.setWordWrap(True)
|
||||
layout.addWidget(self.status_label)
|
||||
|
||||
self.close_btn = QPushButton("Close")
|
||||
self.close_btn.clicked.connect(self.close)
|
||||
layout.addWidget(self.close_btn)
|
||||
|
||||
self._n_angles = len(angle_labels)
|
||||
self._angle_idx = 0
|
||||
self.set_busy(True)
|
||||
|
||||
# ── Worker → window ───────────────────────────────────────────────────────
|
||||
|
||||
def set_busy(self, busy: bool):
|
||||
"""Lock navigation while the stage is moving; the rig is not re-entrant."""
|
||||
for w in (self.angle_list, self.prev_btn, self.next_btn,
|
||||
self.new_point_btn):
|
||||
w.setEnabled(not busy)
|
||||
|
||||
def on_status(self, msg: str):
|
||||
self.status_label.setText(msg)
|
||||
|
||||
def on_point(self, point):
|
||||
self._angle_idx = point.angle_idx
|
||||
self.angle_list.setCurrentRow(point.angle_idx)
|
||||
self.point_label.setText(point.describe())
|
||||
|
||||
# ── Window → worker ───────────────────────────────────────────────────────
|
||||
|
||||
def _on_item_clicked(self, item):
|
||||
self.goto_requested.emit(item.data(Qt.ItemDataRole.UserRole))
|
||||
|
||||
def _on_prev(self):
|
||||
self.goto_requested.emit((self._angle_idx - 1) % self._n_angles)
|
||||
|
||||
def _on_next(self):
|
||||
self.goto_requested.emit((self._angle_idx + 1) % self._n_angles)
|
||||
|
||||
def closeEvent(self, event):
|
||||
self.stop_requested.emit()
|
||||
super().closeEvent(event)
|
||||
|
||||
|
||||
class MainWindow(QMainWindow):
|
||||
def __init__(self):
|
||||
super().__init__()
|
||||
@@ -878,6 +985,14 @@ 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
|
||||
self._scan_thread: QThread | None = None
|
||||
|
||||
self._bbd_active_jog: tuple[str, int] | None = None # (axis, direction)
|
||||
@@ -999,6 +1114,8 @@ 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)
|
||||
self._scan_progress.pause_toggled.connect(self._on_pause_scan)
|
||||
@@ -1164,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()
|
||||
@@ -1172,6 +1294,155 @@ 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():
|
||||
QMessageBox.warning(
|
||||
self, "Scan In Progress",
|
||||
"A scan is running — abort it before inspecting angles."
|
||||
)
|
||||
return
|
||||
if self._inspect_thread is not None and self._inspect_thread.isRunning():
|
||||
self._inspect_window.raise_()
|
||||
self._inspect_window.activateWindow()
|
||||
return
|
||||
|
||||
try:
|
||||
plan, _, _ = self._build_scan_plan()
|
||||
except ValueError as e:
|
||||
QMessageBox.warning(self, "Invalid Scan Parameters", str(e))
|
||||
return
|
||||
|
||||
rotator = RotationAxis(self._t3r_driver.driver, DEFAULT_ROTATION)
|
||||
self._inspect_thread = QThread(self)
|
||||
self._inspect_worker = QtAngleInspector(
|
||||
stage=self._bbd_worker.controller,
|
||||
scope=self._oscope_worker.scope,
|
||||
rotator=rotator,
|
||||
plan=plan,
|
||||
# Same reason the scan does it: the inspector drives the stage from
|
||||
# its own thread, and the position poll shares the BBD TX queue.
|
||||
on_inspect_active=lambda active: setattr(
|
||||
self._bbd_worker, "scanning_active", active),
|
||||
)
|
||||
self._inspect_worker.moveToThread(self._inspect_thread)
|
||||
|
||||
window = AngleInspectWindow(self._inspect_worker.angle_labels(), self)
|
||||
self._inspect_window = window
|
||||
|
||||
window.goto_requested.connect(self._inspect_worker.request_goto)
|
||||
window.new_point_requested.connect(self._inspect_worker.request_new_point)
|
||||
window.stop_requested.connect(self._on_inspect_window_closed)
|
||||
|
||||
self._inspect_worker.status_msg.connect(window.on_status)
|
||||
self._inspect_worker.point_changed.connect(window.on_point)
|
||||
self._inspect_worker.busy_changed.connect(window.set_busy)
|
||||
self._inspect_worker.start_failed.connect(self._on_inspect_failed)
|
||||
self._inspect_worker.error_occurred.connect(
|
||||
lambda m: QMessageBox.warning(self, "Inspection Error", m))
|
||||
|
||||
self._inspect_thread.started.connect(self._inspect_worker.run)
|
||||
self._set_scan_buttons_enabled(False)
|
||||
self.inspect_angles_btn.setEnabled(False)
|
||||
window.show()
|
||||
self._inspect_thread.start()
|
||||
self._inspect_worker.request_start()
|
||||
|
||||
def _on_inspect_failed(self, message: str):
|
||||
QMessageBox.warning(self, "Cannot Inspect Angles", message)
|
||||
if self._inspect_window is not None:
|
||||
self._inspect_window.close()
|
||||
|
||||
def _on_inspect_window_closed(self):
|
||||
"""Tear the worker down and hand the hardware back to the scan panel."""
|
||||
if self._inspect_worker is not None:
|
||||
self._inspect_worker.request_stop()
|
||||
self._inspect_worker.stop_worker()
|
||||
if self._inspect_thread is not None:
|
||||
self._inspect_thread.quit()
|
||||
self._inspect_thread.wait(10000)
|
||||
self._inspect_thread = None
|
||||
self._inspect_worker = None
|
||||
self._inspect_window = None
|
||||
self._set_scan_buttons_enabled(True)
|
||||
self.inspect_angles_btn.setEnabled(True)
|
||||
|
||||
def _on_start_scan(self):
|
||||
try:
|
||||
plan, prefix, save_dir = self._build_scan_plan()
|
||||
@@ -1248,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(
|
||||
@@ -1263,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)
|
||||
@@ -1277,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(
|
||||
@@ -1318,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"
|
||||
@@ -1328,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:
|
||||
|
||||
+46
-3
@@ -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`. |
|
||||
|
||||
+11
-7
@@ -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")
|
||||
|
||||
@@ -0,0 +1,292 @@
|
||||
"""Pre-scan angle inspection, driven entirely by fake hardware.
|
||||
|
||||
The feature's defining constraint is that it reads nothing back from the
|
||||
scope — the operator looks at the instrument. These tests pin that, the scope
|
||||
state the app is responsible for putting the instrument into, and the motion
|
||||
sequence across angles.
|
||||
"""
|
||||
import random
|
||||
|
||||
import pytest
|
||||
|
||||
from core.angle_inspect import AngleInspector, InspectCallbacks
|
||||
from core.rotation import RotationAxis, RotationSettings
|
||||
from core.scan_engine import AXIS_X, AXIS_Y
|
||||
from core.scan_geometry import build_plan
|
||||
from core.scope_inspect import (
|
||||
BIAS_CHANNELS, BIAS_POSITION_DIV, BIAS_SCALE_V_DIV, BIAS_WINDOW_V,
|
||||
INSPECT_TRIG_LEVEL_V, inspect_channel_profiles,
|
||||
)
|
||||
from core.scope_sras import SRAS_CHANNELS
|
||||
from fakes import FakeScope, FakeStage, FakeT3R, Trace
|
||||
|
||||
SPF = 8
|
||||
|
||||
|
||||
def make_plan(num_angles=3):
|
||||
return build_plan(40.0, 30.0, 2.0, 1.0, num_angles, 0.25,
|
||||
laser_freq_hz=20000.0, velocity_mm_s=100.0)
|
||||
|
||||
|
||||
def build(num_angles=3, seed=1234, callbacks=None, rotator_open=True):
|
||||
trace = Trace()
|
||||
scope = FakeScope(trace, samples_per_frame=SPF)
|
||||
stage = FakeStage(trace, scope=scope)
|
||||
t3r = FakeT3R(trace, is_open=rotator_open)
|
||||
rotator = RotationAxis(t3r, RotationSettings())
|
||||
plan = make_plan(num_angles)
|
||||
insp = AngleInspector(stage, scope, rotator, plan,
|
||||
callbacks=callbacks or InspectCallbacks(),
|
||||
rng=random.Random(seed))
|
||||
return insp, trace, plan
|
||||
|
||||
|
||||
def writes(trace):
|
||||
return [c[1] for c in trace.of("write")]
|
||||
|
||||
|
||||
# ── The defining constraint ──────────────────────────────────────────────────
|
||||
|
||||
def test_inspection_never_reads_a_waveform_back():
|
||||
"""The operator reads the scope; the app must not pull data off it.
|
||||
|
||||
If this fails, someone has added a transfer path to a feature whose whole
|
||||
premise is that there isn't one.
|
||||
"""
|
||||
insp, trace, plan = build()
|
||||
insp.start()
|
||||
for i in range(plan.n_angles):
|
||||
insp.goto_angle(i)
|
||||
insp.new_point()
|
||||
insp.stop()
|
||||
|
||||
forbidden = {"transfer_fastframe", "transfer_fastframe_bulk",
|
||||
"transfer_curve", "set_data_source", "query_wfmoutpre"}
|
||||
assert forbidden.isdisjoint(set(trace.names()))
|
||||
assert "CURVe?" not in writes(trace)
|
||||
|
||||
|
||||
# ── Scope configuration ──────────────────────────────────────────────────────
|
||||
|
||||
def test_start_sets_an_edge_trigger_on_ch2_above_the_scan_level():
|
||||
insp, trace, _ = build()
|
||||
insp.start()
|
||||
|
||||
assert "TRIGger:A:TYPe EDGE" in writes(trace)
|
||||
assert trace.of("set_trigger_source")[-1][1] == 2
|
||||
assert trace.of("set_trigger_slope")[-1][1] == "RISE"
|
||||
ch, level = trace.of("set_trigger_level")[-1][1:3]
|
||||
assert (ch, level) == (2, INSPECT_TRIG_LEVEL_V)
|
||||
assert INSPECT_TRIG_LEVEL_V >= 2.0
|
||||
|
||||
|
||||
def test_start_disables_fastframe_averaging_and_the_logic_trigger():
|
||||
"""Everything the scan needs and inspection must not inherit."""
|
||||
insp, trace, _ = build()
|
||||
insp.start()
|
||||
|
||||
assert trace.of("set_fastframe_state")[-1][1] is False
|
||||
assert trace.of("set_acquire_mode")[-1][1] == "SAMPLE"
|
||||
w = writes(trace)
|
||||
assert not any("LOGIc" in cmd or "LOGICPattern" in cmd for cmd in w)
|
||||
|
||||
|
||||
def test_start_leaves_the_acquisition_free_running():
|
||||
"""The display has to keep updating while the operator looks at it."""
|
||||
insp, trace, _ = build()
|
||||
insp.start()
|
||||
|
||||
w = writes(trace)
|
||||
assert "ACQuire:STOPAfter RUNSTop" in w
|
||||
assert w.index("ACQuire:STOPAfter RUNSTop") < w.index("ACQuire:STATE RUN")
|
||||
assert "ACQuire:STATE STOP" not in w
|
||||
|
||||
|
||||
def test_bias_channels_are_directly_comparable():
|
||||
"""CH3/CH4 must share scale and position or the eye comparison is a lie."""
|
||||
profiles = inspect_channel_profiles()
|
||||
a, b = (profiles[ch] for ch in BIAS_CHANNELS)
|
||||
assert a.scale_v_div == b.scale_v_div
|
||||
assert a.position_div == b.position_div
|
||||
# Same front end as the scan records — only the display changes.
|
||||
for ch in BIAS_CHANNELS:
|
||||
assert profiles[ch].termination_ohm == SRAS_CHANNELS[ch].termination_ohm
|
||||
assert profiles[ch].coupling == SRAS_CHANNELS[ch].coupling
|
||||
assert profiles[ch].bandwidth_hz == SRAS_CHANNELS[ch].bandwidth_hz
|
||||
|
||||
|
||||
@pytest.mark.parametrize("n_divisions", [8, 10])
|
||||
def test_bias_window_shows_zero_to_700mv_with_headroom(n_divisions):
|
||||
"""0–700 mV must fit on screen, above ground, on either graticule size.
|
||||
|
||||
Ground sits BIAS_POSITION_DIV divisions below centre, so the visible
|
||||
window runs from (-N/2 - pos)*scale to (+N/2 - pos)*scale.
|
||||
"""
|
||||
half = n_divisions / 2
|
||||
bottom = (-half - BIAS_POSITION_DIV) * BIAS_SCALE_V_DIV
|
||||
top = (half - BIAS_POSITION_DIV) * BIAS_SCALE_V_DIV
|
||||
|
||||
assert bottom < 0.0, "no room below ground for undershoot"
|
||||
assert top > BIAS_WINDOW_V, "700 mV is clipped or sitting on the top edge"
|
||||
# The point of moving the trace down: most of the screen is above ground.
|
||||
assert abs(bottom) < top
|
||||
|
||||
|
||||
def test_ch1_keeps_the_acquisition_front_end():
|
||||
"""What you see at a point is what a scan would record there."""
|
||||
assert inspect_channel_profiles()[1] == SRAS_CHANNELS[1]
|
||||
|
||||
|
||||
# ── Stage and rotation ───────────────────────────────────────────────────────
|
||||
|
||||
def test_start_parks_on_the_first_angle():
|
||||
insp, _, plan = build()
|
||||
point = insp.start()
|
||||
|
||||
assert point.angle_idx == 0
|
||||
assert point.angle_deg == plan.per_angle[0].angle_deg
|
||||
assert insp.current_point == point
|
||||
|
||||
|
||||
def test_the_gate_is_off_for_the_whole_inspection():
|
||||
"""Nothing here is gated, and an armed output keeps driving the line."""
|
||||
insp, trace, _ = build()
|
||||
insp.start()
|
||||
insp.goto_angle(2)
|
||||
insp.new_point()
|
||||
|
||||
assert trace.count("set_trigger_gate_off") >= 1
|
||||
assert trace.count("set_trigger_trigout_maxv") == 0
|
||||
assert [c[2] for c in trace.of("arm_scan_gate") if c[2]] == []
|
||||
|
||||
|
||||
def test_points_land_on_the_scan_grid():
|
||||
"""A point the scan would never sample tells you nothing about the scan."""
|
||||
insp, _, plan = build()
|
||||
insp.start()
|
||||
|
||||
for i in range(plan.n_angles):
|
||||
pa = plan.per_angle[i]
|
||||
for _ in range(5):
|
||||
pt = insp.new_point() if insp.angle_idx == i else insp.goto_angle(i)
|
||||
assert pt.angle_idx == i
|
||||
assert pt.y_mm in pa.y_positions
|
||||
assert pa.x_start <= pt.x_mm <= pa.x_start + pa.x_delta
|
||||
|
||||
|
||||
def test_goto_angle_rotates_then_moves():
|
||||
insp, trace, plan = build()
|
||||
insp.start()
|
||||
trace.calls.clear()
|
||||
|
||||
insp.goto_angle(2)
|
||||
|
||||
# t3r_rotate carries the delta, so assert the resulting absolute angle.
|
||||
assert trace.count("t3r_rotate") == 1, "expected exactly one rotation"
|
||||
assert insp._rotator.current_deg == pytest.approx(plan.per_angle[2].angle_deg)
|
||||
moves = trace.of("move_axis_absolute")
|
||||
assert [m[1] for m in moves] == [AXIS_Y, AXIS_X], "Y then X, as the scan does"
|
||||
|
||||
|
||||
def test_new_point_re_rolls_without_rotating():
|
||||
"""Distinguishing a bad spot from a bad angle depends on not rotating."""
|
||||
insp, trace, _ = build()
|
||||
insp.start()
|
||||
insp.goto_angle(1)
|
||||
trace.calls.clear()
|
||||
|
||||
first = insp.current_point
|
||||
second = insp.new_point()
|
||||
|
||||
assert second.angle_idx == first.angle_idx == 1
|
||||
assert (second.x_mm, second.y_mm) != (first.x_mm, first.y_mm)
|
||||
assert trace.count("t3r_rotate") == 0, "new_point must not rotate"
|
||||
assert [m[1] for m in trace.of("move_axis_absolute")] == [AXIS_Y, AXIS_X]
|
||||
|
||||
|
||||
def test_next_and_prev_wrap_around():
|
||||
insp, _, plan = build(num_angles=3)
|
||||
insp.start()
|
||||
|
||||
assert insp.next_angle().angle_idx == 1
|
||||
assert insp.next_angle().angle_idx == 2
|
||||
assert insp.next_angle().angle_idx == 0, "should wrap forward"
|
||||
assert insp.prev_angle().angle_idx == plan.n_angles - 1, "should wrap back"
|
||||
|
||||
|
||||
def test_angle_labels_cover_every_angle():
|
||||
insp, _, plan = build(num_angles=9)
|
||||
labels = insp.angle_labels()
|
||||
assert len(labels) == 9
|
||||
assert labels[0].startswith("Angle 1/9")
|
||||
|
||||
|
||||
# ── Guards ───────────────────────────────────────────────────────────────────
|
||||
|
||||
def test_multi_angle_inspection_requires_the_rotator():
|
||||
insp, _, _ = build(num_angles=3, rotator_open=False)
|
||||
with pytest.raises(RuntimeError, match="T3R rotation stage"):
|
||||
insp.start()
|
||||
|
||||
|
||||
def test_single_angle_inspection_works_without_the_rotator():
|
||||
insp, _, _ = build(num_angles=1, rotator_open=False)
|
||||
point = insp.start()
|
||||
assert point.angle_idx == 0
|
||||
|
||||
|
||||
def test_navigation_before_start_is_rejected():
|
||||
insp, _, _ = build()
|
||||
with pytest.raises(RuntimeError, match="not been started"):
|
||||
insp.goto_angle(1)
|
||||
with pytest.raises(RuntimeError, match="not been started"):
|
||||
insp.new_point()
|
||||
|
||||
|
||||
def test_out_of_range_angle_is_rejected():
|
||||
insp, _, _ = build(num_angles=3)
|
||||
insp.start()
|
||||
with pytest.raises(IndexError):
|
||||
insp.goto_angle(3)
|
||||
|
||||
|
||||
def test_stop_halts_the_sweep_and_sends_the_rotator_home():
|
||||
insp, trace, _ = build()
|
||||
insp.start()
|
||||
insp.goto_angle(2)
|
||||
trace.calls.clear()
|
||||
|
||||
insp.stop()
|
||||
|
||||
assert "ACQuire:STATE STOP" in writes(trace)
|
||||
assert trace.count("t3r_rotate") == 1, "GR not sent home"
|
||||
assert insp._rotator.current_deg == pytest.approx(0.0)
|
||||
|
||||
|
||||
def test_stop_is_idempotent():
|
||||
insp, trace, _ = build()
|
||||
insp.start()
|
||||
insp.stop()
|
||||
trace.calls.clear()
|
||||
|
||||
insp.stop() # must not re-issue anything or raise
|
||||
|
||||
assert trace.calls == []
|
||||
|
||||
|
||||
def test_busy_callback_brackets_every_move():
|
||||
"""The window disables its controls on this, so it has to pair up."""
|
||||
events = []
|
||||
insp, _, _ = build(callbacks=InspectCallbacks(on_busy=events.append))
|
||||
insp.start()
|
||||
insp.goto_angle(1)
|
||||
insp.new_point()
|
||||
insp.stop()
|
||||
|
||||
assert events, "no busy events emitted"
|
||||
assert events[0] is True and events[-1] is False
|
||||
depth = 0
|
||||
for e in events:
|
||||
depth += 1 if e else -1
|
||||
assert depth in (0, 1), f"unbalanced busy events: {events}"
|
||||
assert depth == 0
|
||||
@@ -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
|
||||
@@ -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()
|
||||
|
||||
Reference in New Issue
Block a user