Thomas Ales
844fcd0297
A full multi-angle scan takes hours, and a rig whose angles disagree produces all of them before anyone finds out. This adds a test mode that acquires one row per angle — the row-wise middle of the ROI — and a viewer that puts every angle's SAW frequency on one graph. The default 80×50 mm ROI at 5 angles goes from 1461 rows to 5. Why the middle row answers an alignment question at all: build_plan centres every angle's rotated bounding box on the same nominal ROI centre, so each angle's middle row crosses that one point on the sample. All the angles measure the same material, so a spread in their frequencies belongs to the rig rather than to where each row happened to land. test_every_angles_middle_row_ crosses_the_roi_centre pins that premise, since the whole comparison rests on it and nothing else in the geometry code would notice it breaking. core/saw_check.py — both halves of the mode, kept together because neither is much use alone. middle_row_plan() reduces a ScanPlan to one row per angle (n_rows // 2, the upper of two centre rows when even); frequency_traces() and alignment_summary() turn the resulting file back into per-angle frequency traces and the scalars an operator is actually asking about — the spread of the per-angle medians, the worst drift along a row, the sparsest row. The verdict thresholds are labelled as rules of thumb, not physics: an anisotropic sample genuinely varies with angle, so a wide spread is a prompt to look at the curves rather than a verdict. Format v10: byte-identical to v6, one row per angle. The version byte earns its keep because the two are otherwise indistinguishable — a v6 scan aborted after its first row is not a check, and a reader guessing from the row count would read a failed scan as a deliberate measurement. create_scan_file() enforces the one-row rule at write time, since nothing downstream can recover from a v10 file that breaks it. ScanEngine gains file_version and is otherwise untouched: the acquisition, the abort/pause path and the background capture are the scan's, unchanged. sras_scan_manager.py now carries the source file's version through an export instead of stamping v6 on everything, which the wider reader would otherwise have made a lie. saw_check_viewer.py — frequency along the row, one curve per angle, over a common offset axis so the curves lie on the same piece of sample; a summary of each angle's median ±1σ against angle; and the per-angle numbers in a table. Analysis parameters (DC threshold, background, time gate) recompute on a worker thread; display ones (smoothing, axis, MHz↔m/s) only redraw. A full v6 scan opens too — the same middle row is pulled out of it — so a finished scan can be re-examined with the check's own read-out. In the app, a check finishes by handing the operator the file and an "Open Viewer" button rather than shutting the rig down the way a completed scan does. Burst mode is not offered: one row per angle means every burst would be a single row, so it buys nothing and still pays for the gate preflight. 137 tests passing, ruff clean. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
645 lines
28 KiB
Python
645 lines
28 KiB
Python
"""Headless SRAS scan engine.
|
||
|
||
Takes plain hardware drivers, a ScanPlan, and callbacks — no Qt, no
|
||
widgets. ``run()`` blocks, so the caller owns the thread; GUIs wrap this
|
||
with gui.scan_bridge.QtScanController, which adapts the callbacks to Qt
|
||
signals. A CLI or a simpler GUI can drive the same engine with nothing but
|
||
functions.
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import logging
|
||
import threading
|
||
import time
|
||
from dataclasses import dataclass, field
|
||
from pathlib import Path
|
||
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, VERSION, create_scan_file
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
SCAN_VELOCITY_MM_S = 100.0
|
||
SCAN_ACCEL_MM_S2 = 1500.0
|
||
LASER_FREQ_HZ = 20000.0 # laser pulse frequency during data acquisition
|
||
# Theoretical ramp distance: d = v² / (2a) = 100² / (2×1500) ≈ 3.33 mm
|
||
SCAN_RAMP_MM = SCAN_VELOCITY_MM_S**2 / (2.0 * SCAN_ACCEL_MM_S2)
|
||
# Extra buffer added to both ends of the ramp. The BBD202 controller begins
|
||
# decelerating slightly before the theoretical point to avoid overshoot,
|
||
# which drops TRIGOUT_MAXV early and clips the last few data points.
|
||
SCAN_RAMP_BUFFER_MM = 1.0
|
||
|
||
AXIS_X = 0x21
|
||
AXIS_Y = 0x22
|
||
|
||
|
||
class ScanAborted(Exception):
|
||
"""Raised inside the engine thread to unwind a scan cleanly."""
|
||
|
||
|
||
@dataclass
|
||
class ResumeTarget:
|
||
"""One angle selected for (re)acquisition in an existing file."""
|
||
angle_idx: int
|
||
data_offset: int
|
||
n_rows: int
|
||
angle_deg: float
|
||
|
||
|
||
@dataclass
|
||
class ResumeState:
|
||
path: Path
|
||
targets: list[ResumeTarget]
|
||
samples_per_frame: int
|
||
|
||
@property
|
||
def target_indices(self) -> set[int]:
|
||
return {t.angle_idx for t in self.targets}
|
||
|
||
|
||
@dataclass
|
||
class ScanCallbacks:
|
||
"""Progress reporting hooks. Every one is optional."""
|
||
on_status: Callable[[str], None] = lambda msg: None
|
||
on_started: Callable[[], None] = lambda: None
|
||
on_row_started: Callable[[int, int, int, int], None] = lambda r, nr, a, na: None
|
||
on_row_done: Callable[[int, int, int, int], None] = lambda r, nr, a, na: None
|
||
on_dc_bias: Callable[[int, list], None] = lambda row, means: None
|
||
on_paused_changed: Callable[[bool], None] = lambda paused: None
|
||
# Blocking operator prompt: must not return until acknowledged.
|
||
prompt: Callable[[str, str], None] = lambda title, msg: None
|
||
|
||
|
||
@dataclass
|
||
class ScanResult:
|
||
path: Path
|
||
rows_written: int = 0
|
||
aborted: bool = False
|
||
angles_acquired: list[int] = field(default_factory=list)
|
||
|
||
|
||
class ScanEngine:
|
||
"""Runs a full SRAS acquisition: stage, rotation, scope, and file output.
|
||
|
||
Constructed with the concrete drivers (not worker/queue wrappers), so
|
||
any front end can reuse it::
|
||
|
||
engine = ScanEngine(stage, scope, rotator, plan, out_path,
|
||
callbacks=ScanCallbacks(on_status=print))
|
||
result = engine.run() # blocking
|
||
"""
|
||
|
||
def __init__(self, stage, scope, rotator: RotationAxis | None,
|
||
plan: ScanPlan, out_path: Path,
|
||
resume: ResumeState | None = None,
|
||
callbacks: ScanCallbacks | None = None,
|
||
burst_mode: bool = False, strict_rows: bool = False,
|
||
file_version: int = VERSION):
|
||
self._stage = stage
|
||
self._scope = scope
|
||
self._rotator = rotator
|
||
self._plan = plan
|
||
self._out_path = Path(out_path)
|
||
self._resume = resume
|
||
self._cb = callbacks if callbacks is not None else ScanCallbacks()
|
||
# Burst mode acquires as many whole rows per FastFrame acquisition as
|
||
# the scope's frame memory holds, instead of one row per acquisition.
|
||
self._burst_mode = burst_mode
|
||
# 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
|
||
|
||
self._abort = threading.Event()
|
||
self._resume_event = threading.Event()
|
||
self._resume_event.set() # set = running, cleared = pause requested
|
||
|
||
# ── External control (thread-safe) ────────────────────────────────────────
|
||
|
||
def abort(self):
|
||
self._abort.set()
|
||
self._resume_event.set() # unblock a paused scan so it can exit
|
||
|
||
def pause(self):
|
||
"""Request a pause; takes effect at the next row boundary."""
|
||
self._resume_event.clear()
|
||
|
||
def resume(self):
|
||
self._resume_event.set()
|
||
|
||
@property
|
||
def aborted(self) -> bool:
|
||
return self._abort.is_set()
|
||
|
||
# ── Internals ─────────────────────────────────────────────────────────────
|
||
|
||
def _check_abort(self):
|
||
if self._abort.is_set():
|
||
raise ScanAborted("Scan aborted by user.")
|
||
|
||
def _pause_point(self):
|
||
"""Block here (between rows, hardware idle) while a pause is requested."""
|
||
if self._resume_event.is_set():
|
||
self._check_abort()
|
||
return
|
||
self._cb.on_status(
|
||
"Scan paused — lasers may be switched off. "
|
||
"Turn lasers back on before resuming."
|
||
)
|
||
self._cb.on_paused_changed(True)
|
||
while not self._resume_event.wait(0.2):
|
||
if self._abort.is_set():
|
||
break
|
||
self._cb.on_paused_changed(False)
|
||
self._check_abort()
|
||
self._cb.on_status("Scan resumed.")
|
||
|
||
def _prompt(self, title: str, message: str):
|
||
self._cb.prompt(title, message)
|
||
self._check_abort()
|
||
|
||
# ── Main sequence ─────────────────────────────────────────────────────────
|
||
|
||
def run(self) -> ScanResult:
|
||
"""Execute the scan. Blocking; returns a ScanResult.
|
||
|
||
Raises ScanGeometryError for an unrunnable plan, RuntimeError for
|
||
missing/mismatched hardware, or ScanAborted if the operator aborts.
|
||
"""
|
||
plan = self._plan
|
||
per_angle = plan.per_angle
|
||
n_angles = plan.n_angles
|
||
result = ScanResult(path=self._out_path)
|
||
|
||
validate_plan(plan, SCAN_RAMP_MM, SCAN_RAMP_BUFFER_MM)
|
||
|
||
geometry_summary = ", ".join(
|
||
f"{pa.angle_deg:.1f}°: {pa.n_rows} row(s) × {pa.n_frames} pts/row"
|
||
for pa in per_angle
|
||
)
|
||
self._cb.on_status(
|
||
f"Scan geometry: {n_angles} angle(s), {plan.total_rows} row(s) total "
|
||
f"(per-angle bounding box) | save → {self._out_path.parent}\n"
|
||
f"{geometry_summary}"
|
||
)
|
||
self._cb.on_started()
|
||
|
||
if self._stage is None:
|
||
raise RuntimeError("BBD202 not connected")
|
||
if self._scope is None:
|
||
raise RuntimeError("Oscilloscope not connected")
|
||
rotator_ready = self._rotator is not None and self._rotator.is_available
|
||
if n_angles > 1 and not rotator_ready:
|
||
raise RuntimeError(
|
||
f"NumAngles={n_angles} requires the T3R rotation stage (GR-axis), "
|
||
"but it is not connected. Connect T3R from the T3R panel before "
|
||
"starting a multi-angle scan, or set NumAngles to 1."
|
||
)
|
||
|
||
if rotator_ready:
|
||
s = self._rotator.settings
|
||
self._cb.on_status(
|
||
f"Configuring GR axis: {s.microsteps} µsteps, "
|
||
f"{s.run_current_ma}/{s.hold_current_ma} mA run/hold …"
|
||
)
|
||
self._rotator.configure()
|
||
time.sleep(0.2)
|
||
|
||
self._prepare_stage()
|
||
samples_per_frame = self._prepare_scope()
|
||
scan_file = self._open_output(samples_per_frame, result)
|
||
|
||
try:
|
||
self._scan_loop(scan_file, samples_per_frame, result)
|
||
finally:
|
||
scan_file.close()
|
||
# Leave the X trigger output inactive. Burst mode toggles it every
|
||
# row and could exit from either state; the per-row path used to
|
||
# leave TRIGOUT_MAXV armed for the rest of the session, which keeps
|
||
# driving the gate line on every later jog.
|
||
try:
|
||
self._stage.set_trigger_gate_off(AXIS_X)
|
||
except Exception:
|
||
logger.exception("Could not return the X trigger output to idle")
|
||
# Return the GR axis home regardless of abort or error
|
||
if 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")
|
||
|
||
if self._abort.is_set():
|
||
result.aborted = True
|
||
raise ScanAborted("Scan aborted by user.")
|
||
self._cb.on_status("Scan complete.")
|
||
return result
|
||
|
||
def _prepare_stage(self):
|
||
ctrl = self._stage
|
||
self._cb.on_status("Enabling stage axes …")
|
||
if not ctrl.am_enabled[0]:
|
||
ctrl.enable_axis(AXIS_X)
|
||
if not ctrl.am_enabled[1]:
|
||
ctrl.enable_axis(AXIS_Y)
|
||
time.sleep(0.2)
|
||
|
||
if not ctrl.am_homed[0] or not ctrl.am_homed[1]:
|
||
self._cb.on_status("Homing stage (may take up to 2 min) …")
|
||
if not ctrl.am_homed[0]:
|
||
ctrl.home_axis(AXIS_X, timeout=120.0)
|
||
if not ctrl.am_homed[1]:
|
||
ctrl.home_axis(AXIS_Y, timeout=120.0)
|
||
|
||
self._cb.on_status("Setting scan velocity …")
|
||
ctrl.set_velocity_params(AXIS_X, max_velocity=SCAN_VELOCITY_MM_S,
|
||
acceleration=SCAN_ACCEL_MM_S2)
|
||
ctrl.set_velocity_params(AXIS_Y, max_velocity=SCAN_VELOCITY_MM_S,
|
||
acceleration=SCAN_ACCEL_MM_S2)
|
||
# X trigger: logic-high output while the stage is at maximum velocity.
|
||
# Burst mode arms it per acquiring pass instead — a burst spans several
|
||
# rows with the scope running throughout, so leaving it armed would let
|
||
# the flyback trigger frames between rows.
|
||
if self._burst_mode:
|
||
ctrl.arm_scan_gate(AXIS_X, False)
|
||
else:
|
||
ctrl.set_trigger_trigout_maxv(AXIS_X)
|
||
|
||
def _prepare_scope(self) -> int:
|
||
self._cb.on_status("Configuring oscilloscope …")
|
||
samples_per_frame = scope_sras.configure_acquisition(self._scope)
|
||
|
||
if self._resume is None:
|
||
self._preambles = scope_sras.read_preambles(self._scope, SCAN_CHANNELS)
|
||
self._prompt(
|
||
"Background Capture",
|
||
"Please ensure the Helios laser is ON and the Genesis laser is OFF,\n"
|
||
"then click OK to capture the background waveform."
|
||
)
|
||
self._background = scope_sras.capture_background(
|
||
self._scope, should_abort=self._abort.is_set,
|
||
on_status=self._cb.on_status)
|
||
self._prompt(
|
||
"Begin Scanning",
|
||
"Background captured successfully.\n\n"
|
||
"Please ensure the Genesis laser is back ON,\n"
|
||
"then click OK to begin scanning."
|
||
)
|
||
else:
|
||
# Resuming: the file's existing background waveform and channel
|
||
# preambles are reused as-is (the format has no way to replace
|
||
# them without rewriting the whole file), so background capture is
|
||
# skipped. Sanity-check that this scope still produces the same
|
||
# record length the file was started with — a mismatch would
|
||
# silently corrupt the ragged per-row byte layout on append.
|
||
if samples_per_frame != self._resume.samples_per_frame:
|
||
raise RuntimeError(
|
||
f"Oscilloscope record length ({samples_per_frame} samples/frame) "
|
||
f"does not match the {self._resume.samples_per_frame} samples/frame "
|
||
f"this scan file was started with — cannot safely resume."
|
||
)
|
||
targets = ", ".join(str(t.angle_idx + 1) for t in self._resume.targets)
|
||
self._prompt(
|
||
"Resume Scan",
|
||
f"Resuming {self._resume.path.name} — will (re)acquire "
|
||
f"angle(s) {targets} of {self._plan.n_angles}.\n\n"
|
||
"Please re-home the GR axis to 0° before continuing — the scan "
|
||
"will rotate it directly from angle to angle before scanning resumes.\n\n"
|
||
"Please ensure the Genesis laser is ON,\n"
|
||
"then click OK to continue scanning."
|
||
)
|
||
|
||
scope_sras.configure_scan_trigger(self._scope)
|
||
|
||
if self._burst_mode:
|
||
# Horizontal settings are fixed by now, so the capacity is stable
|
||
# for the whole scan; only rows-per-burst varies (n_frames is
|
||
# per-angle).
|
||
self._max_frames = scope_burst.max_frames(self._scope)
|
||
self._cb.on_status(
|
||
f"Burst mode: scope holds {self._max_frames} frames "
|
||
f"({samples_per_frame} samples/frame)"
|
||
)
|
||
return samples_per_frame
|
||
|
||
def _open_output(self, samples_per_frame: int, result: ScanResult):
|
||
if self._resume is not None:
|
||
result.path = self._resume.path
|
||
self._cb.on_status(
|
||
f"Resuming {self._resume.path.name} — "
|
||
f"{len(self._resume.targets)} angle(s) to (re)acquire …"
|
||
)
|
||
return open(self._resume.path, "r+b")
|
||
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):
|
||
plan = self._plan
|
||
n_angles = plan.n_angles
|
||
x_ramp_total = SCAN_RAMP_MM + SCAN_RAMP_BUFFER_MM
|
||
scope = self._scope
|
||
|
||
targets_by_ai = None
|
||
if self._resume is not None:
|
||
targets_by_ai = {t.angle_idx: t for t in self._resume.targets}
|
||
|
||
# Both fresh and resumed scans assume the GR axis starts at home (0°)
|
||
# — the resume prompt instructs the operator to re-home it — so the
|
||
# first move always rotates directly from 0° to the starting angle.
|
||
for ai, pa in enumerate(plan.per_angle):
|
||
if targets_by_ai is not None and ai not in targets_by_ai:
|
||
continue # not selected for (re)acquisition
|
||
self._pause_point()
|
||
|
||
if targets_by_ai is not None:
|
||
# Interior angles may already have valid data on either side,
|
||
# so seek to this angle's fixed offset rather than relying on
|
||
# the file's current position.
|
||
scan_file.seek(targets_by_ai[ai].data_offset)
|
||
|
||
if self._rotator is not None and self._rotator.is_available:
|
||
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)
|
||
|
||
if self._burst_mode:
|
||
# Burst mode sizes the FastFrame count from the scope's whole
|
||
# capacity instead (see scope_burst.start_burst), so there is
|
||
# nothing to re-arm per angle here.
|
||
self._scan_rows_burst(scan_file, pa, ai, n_angles,
|
||
samples_per_frame, result, x_ramp_total)
|
||
else:
|
||
# Each angle's bounding box gives it its own points/row count,
|
||
# so the scope's FastFrame count must be re-armed per angle.
|
||
scope.set_fastframe_count(pa.n_frames)
|
||
self._scan_rows_serial(scan_file, pa, ai, n_angles,
|
||
samples_per_frame, result, x_ramp_total)
|
||
|
||
result.angles_acquired.append(ai)
|
||
|
||
# ── Per-row acquisition (one FastFrame acquisition per row) ───────────────
|
||
|
||
def _scan_rows_serial(self, scan_file, pa, ai: int, n_angles: int,
|
||
samples_per_frame: int, result: ScanResult,
|
||
x_ramp_total: float):
|
||
ctrl = self._stage
|
||
scope = self._scope
|
||
|
||
for ri, y_pos in enumerate(pa.y_positions):
|
||
self._pause_point()
|
||
|
||
self._cb.on_row_started(ri + 1, pa.n_rows, ai + 1, n_angles)
|
||
self._cb.on_status(
|
||
f"Angle {ai+1}/{n_angles} Row {ri+1}/{pa.n_rows} "
|
||
f"(Y={y_pos:.3f} mm)"
|
||
)
|
||
|
||
# Position the stage one ramp-length + buffer before the data
|
||
# window so it is at full velocity before x_start.
|
||
ctrl.move_axis_absolute(AXIS_Y, y_pos, timeout=60.0)
|
||
ctrl.move_axis_absolute(AXIS_X, pa.x_start - x_ramp_total, timeout=30.0)
|
||
|
||
scope_sras.arm_row(scope)
|
||
|
||
# Data window + ramp + buffer run-off, so the stage does not
|
||
# begin decelerating before the last point.
|
||
x_end = pa.x_start + pa.x_delta + x_ramp_total
|
||
ctrl.move_axis_absolute(AXIS_X, x_end, timeout=120.0)
|
||
|
||
scope_sras.finish_row(scope)
|
||
self._write_row(scan_file, samples_per_frame, ri, pa.n_frames)
|
||
|
||
result.rows_written += 1
|
||
self._cb.on_row_done(ri + 1, pa.n_rows, ai + 1, n_angles)
|
||
|
||
def _write_row(self, scan_file, samples_per_frame: int, row_idx: int,
|
||
n_frames: int):
|
||
"""Stream every channel from the scope into the file.
|
||
|
||
CH3 is the max-vel gate signal — no useful waveform data — so zeroed
|
||
frames are written to keep the file layout intact.
|
||
"""
|
||
scope = self._scope
|
||
ch_bytes = n_frames * samples_per_frame
|
||
for ch in SCAN_CHANNELS:
|
||
if ch == 3:
|
||
self._cb.on_status("Writing zeroed CH3 frames …")
|
||
scan_file.write(bytes(ch_bytes))
|
||
continue
|
||
|
||
self._cb.on_status(f"Fetching CH{ch} data …")
|
||
waveforms = scope_sras.transfer_channel(scope, ch)
|
||
if ch == SCAN_CHANNELS[0]:
|
||
self._check_frame_delta(row_idx, len(waveforms), n_frames)
|
||
row = scope_burst.normalize_row(
|
||
b"".join(waveforms), 0, len(waveforms), n_frames, samples_per_frame)
|
||
if ch == 4:
|
||
self._cb.on_dc_bias(row_idx + 1, scope_burst.frame_means_block(
|
||
row, 0, n_frames, samples_per_frame))
|
||
scan_file.write(row)
|
||
|
||
# ── Burst acquisition (many whole rows per FastFrame acquisition) ─────────
|
||
|
||
def _scan_rows_burst(self, scan_file, pa, ai: int, n_angles: int,
|
||
samples_per_frame: int, result: ScanResult,
|
||
x_ramp_total: float):
|
||
"""Acquire the angle in bursts of as many whole rows as the scope holds.
|
||
|
||
One ACQuire:STATE RUN spans the whole burst, so the gate is armed only
|
||
for each acquiring pass and dropped for the flyback — otherwise the
|
||
return move would reach max velocity and inject frames between rows.
|
||
"""
|
||
scope = self._scope
|
||
n_frames = pa.n_frames
|
||
x_lead_in = pa.x_start - x_ramp_total
|
||
x_end = pa.x_start + pa.x_delta + x_ramp_total
|
||
|
||
if not self._preflight_done:
|
||
# Once per scan: the gate wiring can't change between angles, and
|
||
# the check costs two row-times.
|
||
self._gate_off_preflight(x_lead_in, x_end)
|
||
self._preflight_done = True
|
||
|
||
row = 0
|
||
while row < pa.n_rows:
|
||
self._pause_point()
|
||
n_burst = scope_burst.rows_per_burst(
|
||
self._max_frames, n_frames, samples_per_frame, pa.n_rows - row)
|
||
self._cb.on_status(
|
||
f"Angle {ai+1}/{n_angles} Rows {row+1}-{row+n_burst}/{pa.n_rows} "
|
||
f"in one acquisition ({n_burst * n_frames} frames) …"
|
||
)
|
||
|
||
burst_start = scan_file.tell()
|
||
cumulative = []
|
||
baseline = scope_burst.start_burst(scope, self._max_frames)
|
||
try:
|
||
for r in range(n_burst):
|
||
self._check_abort()
|
||
self._cb.on_row_started(row + r + 1, pa.n_rows,
|
||
ai + 1, n_angles)
|
||
self._acquire_gated_row(pa.y_positions[row + r],
|
||
x_lead_in, x_end)
|
||
total = scope_burst.frames_acquired(scope)
|
||
if total >= self._max_frames:
|
||
raise RuntimeError(
|
||
f"FastFrame buffer full ({total}/{self._max_frames} "
|
||
f"frames) at row {row + r + 1} — later rows in this "
|
||
"burst would be misattributed. Raise "
|
||
"scope_burst.BURST_FRAME_HEADROOM and rerun."
|
||
)
|
||
cumulative.append(total - baseline)
|
||
finally:
|
||
scope_burst.stop_burst(scope)
|
||
|
||
counts = scope_burst.split_row_counts(cumulative)
|
||
self._write_burst(scan_file, burst_start, row, counts,
|
||
n_frames, samples_per_frame)
|
||
|
||
for r in range(n_burst):
|
||
result.rows_written += 1
|
||
self._cb.on_row_done(row + r + 1, pa.n_rows, ai + 1, n_angles)
|
||
row += n_burst
|
||
|
||
def _acquire_gated_row(self, y_pos: float, x_lead_in: float, x_end: float):
|
||
"""One row: step Y, fly back gated off, then acquire on the +X pass."""
|
||
ctrl = self._stage
|
||
ctrl.move_axis_absolute(AXIS_Y, y_pos, timeout=60.0)
|
||
ctrl.move_axis_absolute(AXIS_X, x_lead_in, timeout=30.0)
|
||
ctrl.arm_scan_gate(AXIS_X, True)
|
||
ctrl.move_axis_absolute(AXIS_X, x_end, timeout=120.0)
|
||
ctrl.arm_scan_gate(AXIS_X, False)
|
||
time.sleep(scope_burst.BURST_ROW_SETTLE_S)
|
||
|
||
def _gate_off_preflight(self, x_lead_in: float, x_end: float):
|
||
"""Prove the gate really gates before trusting a multi-row burst.
|
||
|
||
The value that makes the BBD trigger output idle low is not settled by
|
||
the protocol docs (see apt_constants.TRIGOUT_GATE_OFF), and getting it
|
||
wrong fills every burst with flyback frames that silently shift the
|
||
file. The scope already measures the gate on CH3, so this needs no
|
||
bench probe: one gated-off flyback must acquire nothing, and one gated
|
||
pass must acquire something — the second half is what stops a dark
|
||
laser from making the first half pass vacuously.
|
||
|
||
Leaves the stage parked at x_end, where the burst loop expects it.
|
||
"""
|
||
ctrl, scope = self._stage, self._scope
|
||
self._cb.on_status("Burst preflight: checking the stage gate …")
|
||
|
||
ctrl.arm_scan_gate(AXIS_X, False)
|
||
ctrl.move_axis_absolute(AXIS_X, x_end, timeout=120.0)
|
||
baseline = scope_burst.start_burst(scope, self._max_frames)
|
||
ctrl.move_axis_absolute(AXIS_X, x_lead_in, timeout=120.0)
|
||
scope_burst.stop_burst(scope)
|
||
leaked = scope_burst.frames_acquired(scope) - baseline
|
||
|
||
ctrl.arm_scan_gate(AXIS_X, True)
|
||
baseline = scope_burst.start_burst(scope, self._max_frames)
|
||
ctrl.move_axis_absolute(AXIS_X, x_end, timeout=120.0)
|
||
ctrl.arm_scan_gate(AXIS_X, False)
|
||
scope_burst.stop_burst(scope)
|
||
gated = scope_burst.frames_acquired(scope) - baseline
|
||
|
||
if gated <= 0:
|
||
raise RuntimeError(
|
||
"Burst preflight: no frames acquired with the gate armed. "
|
||
"Check that the Genesis laser is pulsing (CH2) and that the "
|
||
"BBD X trigger output reaches CH3 before scanning."
|
||
)
|
||
if leaked:
|
||
raise RuntimeError(
|
||
f"Burst preflight: {leaked} frame(s) acquired during a flyback "
|
||
"that should have been gated off — the BBD trigger output is "
|
||
"not idling low. Set apt_constants.TRIGOUT_GATE_OFF to "
|
||
"TriggerBitsServo.TRIGOUT_HIGH and retry, or use per-row "
|
||
"acquisition."
|
||
)
|
||
self._cb.on_status(
|
||
f"Burst preflight OK ({gated} frames gated on, 0 leaked).")
|
||
|
||
def _write_burst(self, scan_file, burst_start: int, first_row: int,
|
||
counts: list[int], n_frames: int, samples_per_frame: int):
|
||
"""Deinterleave one burst into the file's per-row, per-channel blocks.
|
||
|
||
The wire is channel-major (every row of CH1, then every row of CH4);
|
||
the file is row-major with channels inner. Writing one channel at a
|
||
time to strided offsets keeps peak memory at a single channel's burst
|
||
instead of the whole thing.
|
||
"""
|
||
scope = self._scope
|
||
ch_bytes = n_frames * samples_per_frame
|
||
row_bytes = len(SCAN_CHANNELS) * ch_bytes
|
||
total_frames = sum(counts)
|
||
|
||
for r, count in enumerate(counts):
|
||
self._check_frame_delta(first_row + r, count, n_frames)
|
||
|
||
for ch_idx, ch in enumerate(SCAN_CHANNELS):
|
||
if ch == 3:
|
||
self._cb.on_status("Writing zeroed CH3 frames …")
|
||
blob = None
|
||
else:
|
||
self._cb.on_status(
|
||
f"Fetching CH{ch} burst ({total_frames} frames) …")
|
||
blob = scope_burst.transfer_burst(scope, ch, total_frames,
|
||
samples_per_frame)
|
||
src = 0
|
||
zeros = bytes(ch_bytes) if blob is None else None
|
||
for r, count in enumerate(counts):
|
||
scan_file.seek(burst_start + r * row_bytes + ch_idx * ch_bytes)
|
||
if blob is None:
|
||
scan_file.write(zeros)
|
||
else:
|
||
row = scope_burst.normalize_row(
|
||
blob, src, count, n_frames, samples_per_frame)
|
||
if ch == 4:
|
||
self._cb.on_dc_bias(
|
||
first_row + r + 1,
|
||
scope_burst.frame_means_block(
|
||
row, 0, n_frames, samples_per_frame))
|
||
scan_file.write(row)
|
||
src += count * samples_per_frame
|
||
del blob
|
||
|
||
scan_file.seek(burst_start + len(counts) * row_bytes)
|
||
|
||
def _check_frame_delta(self, row_idx: int, count: int, n_frames: int):
|
||
"""Decide what to do with a row that did not acquire n_frames frames.
|
||
|
||
v6 declares n_frames per row in the header and has no per-row length
|
||
field, so a mismatched row cannot just be written as-is — that would
|
||
shift every later row in the file. The only two safe options are to
|
||
square it up or to stop, which is what strict_rows selects between.
|
||
|
||
Called before anything for the row is written (CH1 leads
|
||
SCAN_CHANNELS), so raising here leaves no partial row behind.
|
||
"""
|
||
if count == n_frames:
|
||
return
|
||
verb = "zero-padded" if count < n_frames else "truncated"
|
||
if self._strict_rows:
|
||
raise RuntimeError(
|
||
f"Row {row_idx + 1}: {count} frames acquired, {n_frames} "
|
||
f"expected. Strict row packing is on, so the scan stops here "
|
||
f"rather than writing a row that would be {verb}."
|
||
)
|
||
msg = (f"Row {row_idx + 1}: {count} frames acquired, {n_frames} "
|
||
f"expected — {verb} to keep the file layout intact.")
|
||
logger.warning(msg)
|
||
self._cb.on_status(msg)
|