Burst acquisition: many whole rows per FastFrame acquisition
Per-row acquisition pays a full arm/stop/transfer round trip for every row, and the transfer is one IEEE-488.2 block read per frame (~16k frames a row). Burst mode runs one FastFrame acquisition across as many complete rows as the scope's frame memory holds and pulls each burst in a single CURVe? transaction, amortising the round trip over the whole burst. It is opt-in (ScanEngine(burst_mode=...), default False) and writes byte-identical files to the per-row path — test_burst_and_serial_produce_ identical_files runs the same plan both ways and compares the bytes, which is the property the whole feature rests on. core/scope_burst.py — the new policy module. Everything that computes rather than talks to hardware is a free function, so sizing and row-splitting are testable without a rig: rows_per_burst() (rounds down, since a partial row can't be written, and clamps to a transfer-buffer budget), split_row_counts(), normalize_row(), frame_means_block(). The hard part is that a burst carries no row markers — the scope returns one flat run of frames. Boundaries come from ACQuire:NUMFRAMESACQuired? sampled after each acquiring pass while the stage gate is already low, rebased on a baseline read back at RUN rather than assuming the counter resets. A counter that goes backwards means the acquisition restarted mid-burst and is now a hard error instead of silently misattributing every later row. core/scan_engine.py — the row loop splits into _scan_rows_serial and _scan_rows_burst. The wire is channel-major and the file is row-major with channels inner, so _write_burst deinterleaves by writing one channel at a time to strided offsets; peak memory stays at a single channel's burst instead of the whole thing. _gate_off_preflight is what makes this trustworthy on real hardware. The BBD value that idles the trigger output low is not settled by the protocol docs (see 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 the check 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. It runs once per scan and costs two row-times. Two fixes fall out of this work and apply to both paths: - Rows are now squared up to the declared n_frames (short rows zero-padded, long rows truncated, both warned). v6 commits to n_frames per row in the header and has no per-row length field, so an over- or under-triggered row used to shift every later row in the file. - The X trigger output is returned to idle in the run() finally block. The per-row path left TRIGOUT_MAXV armed for the rest of the session, so the gate line kept being driven on every later jog. core/scope_sras.py — pins DATa:ENCdg RIBinary and DATa:WIDth 1 during setup instead of inheriting front-panel state. The file header hardcodes bytes_per_sample=1; a scope left on 2 bytes would have corrupted every frame written. frames_acquired/frame_means move to scope_burst, where the offset- based variants serve both paths. tests/fakes.py — FakeStage and FakeScope are now wired together the way the rig is: a gated X move at scan velocity feeds frames into a running acquisition at the real 20 kHz / 100 mm/s rate, direction-agnostic. Both paths therefore derive frame counts from one model, which is what makes the byte-identity comparison meaningful, and a gate the engine forgets to drop shows up as extra frames instead of passing silently. Frame content is a function of (channel, index) alone, so the same frame sequence yields the same bytes however it is chopped into transfers. 87 tests passing, ruff clean. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
Thomas Ales
+268
-43
@@ -15,7 +15,7 @@ from dataclasses import dataclass, field
|
||||
from pathlib import Path
|
||||
from typing import Callable
|
||||
|
||||
from core import scope_sras
|
||||
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
|
||||
@@ -95,7 +95,8 @@ class ScanEngine:
|
||||
def __init__(self, stage, scope, rotator: RotationAxis | None,
|
||||
plan: ScanPlan, out_path: Path,
|
||||
resume: ResumeState | None = None,
|
||||
callbacks: ScanCallbacks | None = None):
|
||||
callbacks: ScanCallbacks | None = None,
|
||||
burst_mode: bool = False):
|
||||
self._stage = stage
|
||||
self._scope = scope
|
||||
self._rotator = rotator
|
||||
@@ -103,6 +104,11 @@ class ScanEngine:
|
||||
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
|
||||
self._max_frames = 0
|
||||
self._preflight_done = False
|
||||
|
||||
self._abort = threading.Event()
|
||||
self._resume_event = threading.Event()
|
||||
@@ -207,6 +213,14 @@ class ScanEngine:
|
||||
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 …")
|
||||
@@ -242,8 +256,14 @@ class ScanEngine:
|
||||
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
|
||||
ctrl.set_trigger_trigout_maxv(AXIS_X)
|
||||
# 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 …")
|
||||
@@ -290,6 +310,16 @@ class ScanEngine:
|
||||
)
|
||||
|
||||
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):
|
||||
@@ -309,7 +339,6 @@ class ScanEngine:
|
||||
plan = self._plan
|
||||
n_angles = plan.n_angles
|
||||
x_ramp_total = SCAN_RAMP_MM + SCAN_RAMP_BUFFER_MM
|
||||
ctrl = self._stage
|
||||
scope = self._scope
|
||||
|
||||
targets_by_ai = None
|
||||
@@ -337,57 +366,253 @@ class ScanEngine:
|
||||
f"Rotating GR to {pa.angle_deg:.1f}° (Δ{delta:+.1f}°) …")
|
||||
self._rotator.rotate_to(pa.angle_deg)
|
||||
|
||||
# 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)
|
||||
|
||||
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)
|
||||
|
||||
result.rows_written += 1
|
||||
self._cb.on_row_done(ri + 1, pa.n_rows, ai + 1, n_angles)
|
||||
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)
|
||||
|
||||
def _write_row(self, scan_file, samples_per_frame: int, row_idx: int):
|
||||
# ── 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 …")
|
||||
zero_frame = bytes(samples_per_frame)
|
||||
for _ in range(scope_sras.frames_acquired(scope)):
|
||||
scan_file.write(zero_frame)
|
||||
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 == 4 and waveforms:
|
||||
self._cb.on_dc_bias(row_idx + 1, scope_sras.frame_means(waveforms))
|
||||
for w in waveforms:
|
||||
scan_file.write(w)
|
||||
if ch == SCAN_CHANNELS[0]:
|
||||
self._warn_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._warn_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 _warn_frame_delta(self, row_idx: int, count: int, n_frames: int):
|
||||
if count == n_frames:
|
||||
return
|
||||
verb = "zero-padded" if count < n_frames else "truncated"
|
||||
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)
|
||||
|
||||
@@ -0,0 +1,149 @@
|
||||
"""Burst-mode FastFrame acquisition policy.
|
||||
|
||||
Per-row acquisition pays a full arm/stop/transfer round trip for every row,
|
||||
and the transfer alone is one IEEE-488.2 block read per frame (~16k frames a
|
||||
row). A burst instead runs one FastFrame acquisition across as many complete
|
||||
rows as the scope's frame memory holds, then pulls the whole thing in a single
|
||||
transaction — amortising the round trip over `rows_per_burst` rows.
|
||||
|
||||
The scope reports its capacity with ``HORizontal:FASTframe:MAXFRames?`` once
|
||||
the horizontal settings are fixed; ``rows_per_burst`` turns that into a row
|
||||
count. Everything here that computes rather than talks to hardware is a free
|
||||
function, so the row-splitting logic is testable without a rig.
|
||||
|
||||
The catch is that the burst contains no row markers: the scope hands back one
|
||||
flat run of frames. Boundaries come from polling ``ACQuire:NUMFRAMESACQuired?``
|
||||
after each row's acquiring pass, while the stage gate is already low — see
|
||||
``split_row_counts``.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import time
|
||||
|
||||
import numpy as np
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# Peak transfer buffer, per channel. The writer holds one channel at a time
|
||||
# (see ScanEngine._scan_rows_burst), so this is the real high-water mark.
|
||||
BURST_MEMORY_BUDGET_BYTES = 512 * 1024 * 1024
|
||||
|
||||
# Extra frames budgeted per row on top of n_frames. 0 gives the plain
|
||||
# floor(max_frames / n_frames) row count; raise it if the acquiring pass
|
||||
# routinely over-triggers (watch the pad/truncate warnings).
|
||||
BURST_FRAME_HEADROOM = 0
|
||||
|
||||
BURST_ARM_SETTLE_S = 0.05 # after ACQuire:STATE RUN, before the first move
|
||||
BURST_ROW_SETTLE_S = 0.05 # after the gate drops, before reading the counter
|
||||
|
||||
|
||||
# ── Pure helpers ─────────────────────────────────────────────────────────────
|
||||
|
||||
def rows_per_burst(max_frames: int, n_frames: int, samples_per_frame: int,
|
||||
rows_remaining: int,
|
||||
memory_budget: int = BURST_MEMORY_BUDGET_BYTES,
|
||||
headroom: int = BURST_FRAME_HEADROOM) -> int:
|
||||
"""How many complete rows fit in one acquisition.
|
||||
|
||||
Rounds down — a partial row is worthless, since a row must be transferred
|
||||
whole to be written. Clamped by the transfer buffer budget and by the rows
|
||||
actually left in the angle, and never below 1 (a single row always goes,
|
||||
even if it exceeds the budget, so the scan can still make progress).
|
||||
"""
|
||||
if n_frames < 1 or samples_per_frame < 1:
|
||||
raise ValueError(f"n_frames={n_frames} samples_per_frame={samples_per_frame}")
|
||||
|
||||
by_scope = max_frames // (n_frames + headroom)
|
||||
by_memory = memory_budget // (n_frames * samples_per_frame)
|
||||
return max(1, min(by_scope, by_memory, rows_remaining))
|
||||
|
||||
|
||||
def split_row_counts(cumulative: list[int]) -> list[int]:
|
||||
"""Per-row frame counts from the cumulative counter sampled after each row.
|
||||
|
||||
``cumulative`` is ``ACQuire:NUMFRAMESACQuired?`` read once per row, already
|
||||
rebased on the value at burst start.
|
||||
"""
|
||||
counts = []
|
||||
prev = 0
|
||||
for i, c in enumerate(cumulative):
|
||||
if c < prev:
|
||||
raise RuntimeError(
|
||||
f"FastFrame counter went backwards at row {i} ({prev} → {c}) — "
|
||||
"the acquisition was restarted mid-burst"
|
||||
)
|
||||
counts.append(c - prev)
|
||||
prev = c
|
||||
return counts
|
||||
|
||||
|
||||
def normalize_row(buf, offset: int, count: int, n_frames: int,
|
||||
samples_per_frame: int):
|
||||
"""Coerce one row's frames to exactly ``n_frames``.
|
||||
|
||||
The v6 format commits to n_frames per row in the header and has no per-row
|
||||
length field, so a row that over- or under-triggers must be squared up or
|
||||
every later row in the file shifts. Short rows are zero-padded, long rows
|
||||
lose their trailing frames. Returns something writable directly.
|
||||
"""
|
||||
want = n_frames * samples_per_frame
|
||||
end = min(offset + count * samples_per_frame, offset + want, len(buf))
|
||||
chunk = memoryview(buf)[offset:end]
|
||||
if len(chunk) == want:
|
||||
return chunk
|
||||
return bytes(chunk) + bytes(want - len(chunk))
|
||||
|
||||
|
||||
def frame_means_block(buf, offset: int, n_frames: int,
|
||||
samples_per_frame: int) -> list[float]:
|
||||
"""Per-frame DC mean over one row's slice of a burst buffer."""
|
||||
n = n_frames * samples_per_frame
|
||||
block = np.frombuffer(buf, dtype=np.int8, count=n, offset=offset)
|
||||
return block.reshape(n_frames, samples_per_frame).mean(
|
||||
axis=1, dtype=np.float32).tolist()
|
||||
|
||||
|
||||
# ── Instrument control ───────────────────────────────────────────────────────
|
||||
|
||||
def max_frames(scope) -> int:
|
||||
"""Frames the scope can hold under the current horizontal settings."""
|
||||
try:
|
||||
m = scope.get_fastframe_max_frames()
|
||||
except Exception as exc:
|
||||
raise RuntimeError(
|
||||
"Scope did not answer HORizontal:FASTframe:MAXFRames? — burst mode "
|
||||
"cannot size a burst without it. Use per-row acquisition on this "
|
||||
f"firmware. ({exc})"
|
||||
) from exc
|
||||
if m < 1:
|
||||
raise RuntimeError(f"Scope reports a FastFrame capacity of {m} frames")
|
||||
return m
|
||||
|
||||
|
||||
def start_burst(scope, frame_count: int) -> int:
|
||||
"""Arm one burst; returns the counter baseline to subtract from later reads.
|
||||
|
||||
Reading the baseline back beats assuming the counter resets to 0 on RUN —
|
||||
any residual is simply subtracted out instead of being misattributed to the
|
||||
first row.
|
||||
"""
|
||||
scope.set_fastframe_count(frame_count)
|
||||
scope.write("ACQuire:STATE RUN")
|
||||
time.sleep(BURST_ARM_SETTLE_S)
|
||||
return frames_acquired(scope)
|
||||
|
||||
|
||||
def stop_burst(scope) -> None:
|
||||
time.sleep(BURST_ROW_SETTLE_S)
|
||||
scope.write("ACQuire:STATE STOP")
|
||||
|
||||
|
||||
def frames_acquired(scope) -> int:
|
||||
return int(scope.query("ACQuire:NUMFRAMESACQuired?"))
|
||||
|
||||
|
||||
def transfer_burst(scope, ch: int, frame_count: int, samples_per_frame: int):
|
||||
"""Pull a whole burst for one channel in a single CURVe? transaction."""
|
||||
scope.set_data_source(ch)
|
||||
return scope.transfer_fastframe_bulk(frame_count, samples_per_frame)
|
||||
+5
-23
@@ -10,8 +10,6 @@ import logging
|
||||
import time
|
||||
from dataclasses import dataclass
|
||||
|
||||
import numpy as np
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
SAMPLE_RATE_HZ = 6.25e9 # 6.25 GS/s → 160 ps/sample
|
||||
@@ -66,6 +64,11 @@ def configure_acquisition(scope) -> int:
|
||||
scope.set_trigger_mode("NORMAL") # wait for trigger (don't auto-sweep)
|
||||
scope.set_acquire_mode("SAMPLE")
|
||||
scope.set_fastframe_state(False)
|
||||
# Pin the transfer format instead of inheriting front-panel state — the
|
||||
# file header hardcodes bytes_per_sample=1, and a scope left on 2 bytes
|
||||
# would corrupt every frame written.
|
||||
scope.set_data_encoding("RIBinary")
|
||||
scope.set_data_width(1)
|
||||
scope.set_sample_rate(SAMPLE_RATE_HZ)
|
||||
scope.write("HORizontal:POSition 30") # 10 % trigger offset
|
||||
time.sleep(0.3) # let the timebase settle before reading back
|
||||
@@ -140,28 +143,7 @@ def finish_row(scope) -> None:
|
||||
scope.write("ACQuire:STATE STOP")
|
||||
|
||||
|
||||
def frames_acquired(scope) -> int:
|
||||
return int(scope.query("ACQuire:NUMFRAMESACQuired?"))
|
||||
|
||||
|
||||
def transfer_channel(scope, ch: int) -> list[bytes]:
|
||||
"""Fetch one channel's FastFrame block as raw int8 frames."""
|
||||
scope.set_data_source(ch)
|
||||
return scope.transfer_fastframe(parse=False)
|
||||
|
||||
|
||||
def frame_means(waveforms: list[bytes]) -> list[float]:
|
||||
"""Per-frame DC mean of a raw int8 FastFrame block.
|
||||
|
||||
numpy over the joined buffer: the per-frame struct.unpack this replaces
|
||||
allocated a tuple of Python ints per frame (~16k frames per row).
|
||||
"""
|
||||
if not waveforms:
|
||||
return []
|
||||
n = len(waveforms[0])
|
||||
if n == 0 or any(len(w) != n for w in waveforms):
|
||||
# Ragged block (shouldn't happen) — fall back to per-frame means.
|
||||
return [float(np.frombuffer(w, dtype=np.int8).mean()) if len(w) else 0.0
|
||||
for w in waveforms]
|
||||
block = np.frombuffer(b"".join(waveforms), dtype=np.int8).reshape(len(waveforms), n)
|
||||
return block.mean(axis=1, dtype=np.float32).tolist()
|
||||
|
||||
+99
-16
@@ -3,9 +3,24 @@
|
||||
Each fake records an ordered call trace, so a test can assert the exact
|
||||
command sequence the engine issues — the property that matters when the
|
||||
real rig isn't available.
|
||||
|
||||
The stage and scope are wired together the way the rig is: an X move at scan
|
||||
velocity with the trigger gate armed feeds frames into a running acquisition,
|
||||
at the real 20 kHz / 100 mm/s rate. Per-row and burst acquisition therefore
|
||||
get their frame counts from the same model, which is what makes a
|
||||
byte-identity comparison between the two paths meaningful — and it means a
|
||||
gate the engine forgets to drop shows up as extra frames instead of passing
|
||||
silently.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
from core.scan_engine import (
|
||||
AXIS_X, LASER_FREQ_HZ, SCAN_RAMP_BUFFER_MM, SCAN_RAMP_MM,
|
||||
SCAN_VELOCITY_MM_S,
|
||||
)
|
||||
|
||||
RAMP_TOTAL_MM = SCAN_RAMP_MM + SCAN_RAMP_BUFFER_MM
|
||||
|
||||
|
||||
class Trace:
|
||||
"""Ordered record of hardware calls, shared by all fakes in one test."""
|
||||
@@ -29,29 +44,59 @@ class Trace:
|
||||
class FakeStage:
|
||||
"""Stands in for ThorlabsServoDriver."""
|
||||
|
||||
def __init__(self, trace: Trace, homed=(True, True), enabled=(True, True)):
|
||||
def __init__(self, trace: Trace, homed=(True, True), enabled=(True, True),
|
||||
scope=None):
|
||||
self._t = trace
|
||||
self.am_homed = list(homed)
|
||||
self.am_enabled = list(enabled)
|
||||
self.positions = [0.0, 0.0]
|
||||
self._scope = scope
|
||||
self.gate_armed = False
|
||||
|
||||
def attach_scope(self, scope):
|
||||
"""Route gated motion into `scope`, as the TRIGOUT pin does on the rig."""
|
||||
self._scope = scope
|
||||
|
||||
def enable_axis(self, axis):
|
||||
self._t.record("enable_axis", axis)
|
||||
self.am_enabled[0 if axis == 0x21 else 1] = True
|
||||
self.am_enabled[0 if axis == AXIS_X else 1] = True
|
||||
|
||||
def home_axis(self, axis, timeout=0.0):
|
||||
self._t.record("home_axis", axis)
|
||||
self.am_homed[0 if axis == 0x21 else 1] = True
|
||||
self.am_homed[0 if axis == AXIS_X else 1] = True
|
||||
|
||||
def set_velocity_params(self, axis, max_velocity=None, acceleration=None):
|
||||
self._t.record("set_velocity_params", axis, max_velocity, acceleration)
|
||||
|
||||
def set_trigger_trigout_maxv(self, axis):
|
||||
self._t.record("set_trigger_trigout_maxv", axis)
|
||||
if axis == AXIS_X:
|
||||
self.gate_armed = True
|
||||
|
||||
def set_trigger_gate_off(self, axis):
|
||||
self._t.record("set_trigger_gate_off", axis)
|
||||
if axis == AXIS_X:
|
||||
self.gate_armed = False
|
||||
|
||||
def arm_scan_gate(self, axis, armed, verify=True):
|
||||
self._t.record("arm_scan_gate", axis, bool(armed))
|
||||
if axis == AXIS_X:
|
||||
self.gate_armed = bool(armed)
|
||||
|
||||
def move_axis_absolute(self, axis, pos, timeout=0.0):
|
||||
idx = 0 if axis == AXIS_X else 1
|
||||
prev = self.positions[idx]
|
||||
self._t.record("move_axis_absolute", axis, round(pos, 6))
|
||||
self.positions[0 if axis == 0x21 else 1] = pos
|
||||
self.positions[idx] = pos
|
||||
|
||||
# The gate is high only at max velocity, i.e. over the move minus its
|
||||
# two ramps — direction-agnostic, so a flyback the engine failed to
|
||||
# gate off produces frames instead of quietly producing none.
|
||||
if axis == AXIS_X and self.gate_armed and self._scope is not None:
|
||||
at_speed_mm = abs(pos - prev) - 2 * RAMP_TOTAL_MM
|
||||
if at_speed_mm > 0:
|
||||
self._scope.acquire_frames(
|
||||
round(at_speed_mm * LASER_FREQ_HZ / SCAN_VELOCITY_MM_S))
|
||||
|
||||
|
||||
class FakeScope:
|
||||
@@ -61,24 +106,42 @@ class FakeScope:
|
||||
against an expected byte pattern.
|
||||
"""
|
||||
|
||||
def __init__(self, trace: Trace, samples_per_frame=8, n_frames=4):
|
||||
def __init__(self, trace: Trace, samples_per_frame=8, max_frames=4096):
|
||||
self._t = trace
|
||||
self.samples_per_frame = samples_per_frame
|
||||
self._n_frames = n_frames
|
||||
self.max_frames = max_frames
|
||||
self._acq_polls = 0
|
||||
self.frame_seq = 0
|
||||
self._running = False
|
||||
self._acquired = 0
|
||||
# Per-channel running frame index. Frame content is a function of
|
||||
# (channel, index) alone, so the same total frame sequence yields the
|
||||
# same bytes however it is chopped into transfers.
|
||||
self._next_frame: dict[int, int] = {}
|
||||
|
||||
# -- driven by FakeStage ------------------------------------------------
|
||||
def acquire_frames(self, n):
|
||||
if self._running:
|
||||
self._acquired += n
|
||||
|
||||
# -- writes / queries ---------------------------------------------------
|
||||
def write(self, cmd):
|
||||
self._t.record("write", cmd)
|
||||
if cmd == "ACQuire:STATE RUN":
|
||||
self._running = True
|
||||
self._acquired = 0
|
||||
elif cmd == "ACQuire:STATE STOP":
|
||||
self._running = False
|
||||
|
||||
def query(self, cmd):
|
||||
self._t.record("query", cmd)
|
||||
if cmd == "ACQuire:STATE?":
|
||||
self._acq_polls += 1
|
||||
# STOPAfter SEQuence self-stops when the sequence completes, so
|
||||
# reporting "stopped" and staying armed would be inconsistent.
|
||||
self._running = False
|
||||
return "0" # background average finished
|
||||
if cmd == "ACQuire:NUMFRAMESACQuired?":
|
||||
return str(self._n_frames)
|
||||
return str(self._acquired)
|
||||
return ""
|
||||
|
||||
# -- typed setters used by core.scope_sras ------------------------------
|
||||
@@ -102,7 +165,13 @@ class FakeScope:
|
||||
|
||||
def set_fastframe_count(self, n):
|
||||
self._t.record("set_fastframe_count", n)
|
||||
self._n_frames = n
|
||||
|
||||
def get_fastframe_state(self):
|
||||
return 1
|
||||
|
||||
def get_fastframe_max_frames(self):
|
||||
self._t.record("get_fastframe_max_frames")
|
||||
return self.max_frames
|
||||
|
||||
def set_sample_rate(self, sr):
|
||||
self._t.record("set_sample_rate", sr)
|
||||
@@ -114,6 +183,12 @@ class FakeScope:
|
||||
self._t.record("set_data_source", ch)
|
||||
self._source = ch
|
||||
|
||||
def set_data_encoding(self, encoding):
|
||||
self._t.record("set_data_encoding", encoding)
|
||||
|
||||
def set_data_width(self, width):
|
||||
self._t.record("set_data_width", width)
|
||||
|
||||
def query_wfmoutpre(self):
|
||||
return f"WFMOUTPRE:CH{self._source};YMULT 1.5625E-3;YOFF -87.04;YZERO 0.0"
|
||||
|
||||
@@ -121,14 +196,22 @@ class FakeScope:
|
||||
self._t.record("transfer_curve")
|
||||
return bytes(range(self.samples_per_frame))
|
||||
|
||||
def transfer_fastframe(self, parse=True):
|
||||
def _frames(self, ch, count):
|
||||
spf = self.samples_per_frame
|
||||
start = self._next_frame.get(ch, 0)
|
||||
self._next_frame[ch] = start + count
|
||||
return [bytes((ch * 31 + g + s) % 256 for s in range(spf))
|
||||
for g in range(start, start + count)]
|
||||
|
||||
def transfer_fastframe(self, parse=True, byte_count=1, signed=True,
|
||||
byte_order='MSB'):
|
||||
self._t.record("transfer_fastframe", self._source)
|
||||
frames = []
|
||||
for i in range(self._n_frames):
|
||||
frames.append(bytes((self.frame_seq + i + s) % 256
|
||||
for s in range(self.samples_per_frame)))
|
||||
self.frame_seq += 1
|
||||
return frames
|
||||
return self._frames(self._source, self._acquired)
|
||||
|
||||
def transfer_fastframe_bulk(self, frame_count, samples_per_frame,
|
||||
bytes_per_sample=1):
|
||||
self._t.record("transfer_fastframe_bulk", self._source, frame_count)
|
||||
return bytearray(b"".join(self._frames(self._source, frame_count)))
|
||||
|
||||
# channel config (only used by configure_channels)
|
||||
def set_channel_label_name(self, ch, name):
|
||||
|
||||
+143
-10
@@ -20,22 +20,24 @@ from fakes import FakeScope, FakeStage, FakeT3R, Trace
|
||||
SPF = 8
|
||||
|
||||
|
||||
def make_plan(num_angles=1):
|
||||
# Small ROI well inside the stage limits: 1 row, few frames per angle.
|
||||
return build_plan(40.0, 30.0, 0.02, 0.005, num_angles, 0.01,
|
||||
def make_plan(num_angles=1, y_delta=0.005):
|
||||
# Small ROI well inside the stage limits: few rows, few frames per angle.
|
||||
return build_plan(40.0, 30.0, 0.02, y_delta, num_angles, 0.01,
|
||||
laser_freq_hz=20000.0, velocity_mm_s=100.0)
|
||||
|
||||
|
||||
def build(tmp_path, num_angles=1, callbacks=None, resume=None, **kw):
|
||||
def build(tmp_path, num_angles=1, callbacks=None, resume=None, plan=None,
|
||||
burst_mode=False, max_frames=4096, out_name="out.sras", **kw):
|
||||
trace = Trace()
|
||||
stage = FakeStage(trace)
|
||||
scope = FakeScope(trace, samples_per_frame=SPF)
|
||||
scope = FakeScope(trace, samples_per_frame=SPF, max_frames=max_frames)
|
||||
stage = FakeStage(trace, scope=scope)
|
||||
t3r = FakeT3R(trace, **kw)
|
||||
rotator = RotationAxis(t3r, RotationSettings())
|
||||
plan = make_plan(num_angles)
|
||||
engine = ScanEngine(stage, scope, rotator, plan, tmp_path / "out.sras",
|
||||
plan = plan if plan is not None else make_plan(num_angles)
|
||||
engine = ScanEngine(stage, scope, rotator, plan, tmp_path / out_name,
|
||||
resume=resume,
|
||||
callbacks=callbacks or ScanCallbacks())
|
||||
callbacks=callbacks or ScanCallbacks(),
|
||||
burst_mode=burst_mode)
|
||||
return engine, trace, plan
|
||||
|
||||
|
||||
@@ -192,8 +194,8 @@ def test_dc_bias_callback_reports_per_frame_means(tmp_path):
|
||||
|
||||
def test_offstage_plan_rejected_before_touching_hardware(tmp_path):
|
||||
trace = Trace()
|
||||
stage = FakeStage(trace)
|
||||
scope = FakeScope(trace, samples_per_frame=SPF)
|
||||
stage = FakeStage(trace, scope=scope)
|
||||
# X range that runs off the 110 mm stage once ramps are added
|
||||
plan = build_plan(80.0, 30.0, 40.0, 5.0, 1, 0.25,
|
||||
laser_freq_hz=20000.0, velocity_mm_s=100.0)
|
||||
@@ -265,6 +267,137 @@ def test_resume_record_length_mismatch_rejected(tmp_path):
|
||||
engine2.run()
|
||||
|
||||
|
||||
# ── Burst acquisition ────────────────────────────────────────────────────────
|
||||
|
||||
# 6 rows × 4 frames/row; max_frames=14 gives 14//4 = 3 rows per burst, so the
|
||||
# angle needs two bursts and the second is not a whole burst wide.
|
||||
BURST_PLAN = dict(y_delta=0.05)
|
||||
BURST_MAX_FRAMES = 14
|
||||
|
||||
|
||||
def test_burst_and_serial_produce_identical_files(tmp_path):
|
||||
"""The whole point: burst mode must be a pure acquisition optimisation."""
|
||||
plan = make_plan(**BURST_PLAN)
|
||||
assert plan.per_angle[0].n_rows == 6 and plan.per_angle[0].n_frames == 4
|
||||
|
||||
serial, _, _ = build(tmp_path, plan=plan, out_name="serial.sras")
|
||||
serial.run()
|
||||
burst, _, _ = build(tmp_path, plan=plan, out_name="burst.sras",
|
||||
burst_mode=True, max_frames=BURST_MAX_FRAMES)
|
||||
burst.run()
|
||||
|
||||
assert (tmp_path / "burst.sras").read_bytes() == \
|
||||
(tmp_path / "serial.sras").read_bytes()
|
||||
|
||||
|
||||
def test_burst_multi_angle_file_is_complete(tmp_path):
|
||||
plan = make_plan(num_angles=3, **BURST_PLAN)
|
||||
engine, trace, _ = build(tmp_path, plan=plan, burst_mode=True,
|
||||
max_frames=BURST_MAX_FRAMES)
|
||||
result = engine.run()
|
||||
|
||||
assert result.rows_written == plan.total_rows
|
||||
assert result.angles_acquired == [0, 1, 2]
|
||||
sras = SrasFile(tmp_path / "out.sras")
|
||||
assert [s.status for s in sras.angle_status()] == ["OK"] * 3
|
||||
|
||||
|
||||
def test_burst_gates_the_flyback_and_runs_once_per_burst(tmp_path):
|
||||
plan = make_plan(**BURST_PLAN)
|
||||
engine, trace, _ = build(tmp_path, plan=plan, burst_mode=True,
|
||||
max_frames=BURST_MAX_FRAMES)
|
||||
engine.run()
|
||||
|
||||
# Two bursts (3 + 3 rows), two preflight acquisitions, one background.
|
||||
runs = [c for c in trace.of("write") if c[1] == "ACQuire:STATE RUN"]
|
||||
assert len(runs) == 5
|
||||
|
||||
# Every acquiring pass is bracketed by an arm/disarm, so the gate is low
|
||||
# for each flyback. 6 rows + 1 preflight pass = 7 arms.
|
||||
gate = [c[2] for c in trace.of("arm_scan_gate")]
|
||||
assert gate.count(True) == 7
|
||||
# No two arms without a disarm between them — that is what would let a
|
||||
# flyback into the acquisition. (A repeated disarm is just defensive.)
|
||||
for a, b in zip(gate, gate[1:], strict=False):
|
||||
assert not (a and b), f"acquiring pass with no disarm before it: {gate}"
|
||||
assert gate[-1] is False, "scan left the gate armed"
|
||||
|
||||
# One bulk transfer per data channel per burst, none per row.
|
||||
assert [(c[1], c[2]) for c in trace.of("transfer_fastframe_bulk")] == [
|
||||
(1, 12), (4, 12), (1, 12), (4, 12)]
|
||||
assert trace.count("transfer_fastframe") == 0
|
||||
|
||||
|
||||
def test_burst_preflight_rejects_a_leaky_gate(tmp_path):
|
||||
plan = make_plan(**BURST_PLAN)
|
||||
engine, trace, _ = build(tmp_path, plan=plan, burst_mode=True,
|
||||
max_frames=BURST_MAX_FRAMES)
|
||||
stage = engine._stage
|
||||
|
||||
# A gate that ignores the disable request — the failure mode the preflight
|
||||
# exists to catch (TRIGOUT_GATE_OFF set to the wrong mode value).
|
||||
def stuck_gate(axis, armed, verify=True):
|
||||
trace.record("arm_scan_gate", axis, bool(armed))
|
||||
stage.gate_armed = True
|
||||
stage.arm_scan_gate = stuck_gate
|
||||
|
||||
with pytest.raises(RuntimeError, match="not idling low"):
|
||||
engine.run()
|
||||
|
||||
|
||||
def test_burst_preflight_rejects_a_dark_laser(tmp_path):
|
||||
"""A gate that never fires would let a leak check pass vacuously."""
|
||||
plan = make_plan(**BURST_PLAN)
|
||||
engine, trace, _ = build(tmp_path, plan=plan, burst_mode=True,
|
||||
max_frames=BURST_MAX_FRAMES)
|
||||
engine._stage.attach_scope(None) # no pulses ever reach the scope
|
||||
|
||||
with pytest.raises(RuntimeError, match="no frames acquired"):
|
||||
engine.run()
|
||||
|
||||
|
||||
@pytest.mark.parametrize("burst_mode", [False, True])
|
||||
def test_short_row_is_padded_to_declared_frame_count(tmp_path, burst_mode):
|
||||
"""A clipped row must not shift every later row in the file.
|
||||
|
||||
v6 declares n_frames per row up front and has no per-row length, so an
|
||||
under-triggered row has to be squared up. In burst mode this also proves
|
||||
the splitter advances by what actually arrived, not by n_frames.
|
||||
"""
|
||||
warnings = []
|
||||
plan = make_plan(**BURST_PLAN)
|
||||
engine, trace, _ = build(
|
||||
tmp_path, plan=plan, burst_mode=burst_mode,
|
||||
max_frames=BURST_MAX_FRAMES,
|
||||
callbacks=ScanCallbacks(on_status=warnings.append))
|
||||
# The preflight is covered by its own tests; skipping it keeps the
|
||||
# acquiring-pass count below identical in both modes.
|
||||
engine._preflight_done = True
|
||||
|
||||
scope = engine._scope
|
||||
real_acquire = scope.acquire_frames
|
||||
passes = {"n": 0}
|
||||
|
||||
def clipped(n):
|
||||
if scope._running:
|
||||
passes["n"] += 1
|
||||
if passes["n"] == 2: # second data row loses one frame
|
||||
n -= 1
|
||||
real_acquire(n)
|
||||
scope.acquire_frames = clipped
|
||||
|
||||
result = engine.run()
|
||||
|
||||
assert result.rows_written == 6
|
||||
assert any("Row 2: 3 frames acquired, 4 expected" in w for w in warnings)
|
||||
assert any("zero-padded" in w for w in warnings)
|
||||
sras = SrasFile(tmp_path / "out.sras")
|
||||
assert [s.status for s in sras.angle_status()] == ["OK"]
|
||||
# The padding lands at the end of the short row, not in the next one.
|
||||
assert bytes(sras.load_row(0, 1, 0)[-1]) == bytes(SPF)
|
||||
assert bytes(sras.load_row(0, 2, 0)[0]) != bytes(SPF)
|
||||
|
||||
|
||||
def test_engine_imports_without_qt():
|
||||
"""The engine must be usable from a non-Qt front end."""
|
||||
import subprocess
|
||||
|
||||
@@ -0,0 +1,103 @@
|
||||
"""Burst sizing and row-splitting, exercised without any instrument."""
|
||||
import pytest
|
||||
|
||||
from core.scope_burst import (
|
||||
frame_means_block, normalize_row, rows_per_burst, split_row_counts,
|
||||
)
|
||||
|
||||
SPF = 8
|
||||
|
||||
|
||||
# ── rows_per_burst ───────────────────────────────────────────────────────────
|
||||
|
||||
def test_rows_per_burst_rounds_down():
|
||||
# 9.7 rows' worth of capacity is 9 rows: a partial row is unusable.
|
||||
assert rows_per_burst(97, 10, SPF, rows_remaining=100) == 9
|
||||
assert rows_per_burst(100, 10, SPF, rows_remaining=100) == 10
|
||||
|
||||
|
||||
def test_rows_per_burst_clamped_by_rows_remaining():
|
||||
assert rows_per_burst(1000, 10, SPF, rows_remaining=3) == 3
|
||||
|
||||
|
||||
def test_rows_per_burst_clamped_by_memory_budget():
|
||||
# Budget holds 4 rows of 10 frames × 8 samples; the scope would hold 100.
|
||||
assert rows_per_burst(1000, 10, SPF, rows_remaining=100,
|
||||
memory_budget=4 * 10 * SPF) == 4
|
||||
|
||||
|
||||
def test_rows_per_burst_headroom_reserves_slack_per_row():
|
||||
assert rows_per_burst(100, 10, SPF, rows_remaining=100, headroom=0) == 10
|
||||
assert rows_per_burst(100, 10, SPF, rows_remaining=100, headroom=2) == 8
|
||||
|
||||
|
||||
def test_rows_per_burst_never_returns_zero():
|
||||
"""A row too big for any budget still goes, or the scan cannot progress."""
|
||||
assert rows_per_burst(5, 10, SPF, rows_remaining=100) == 1
|
||||
assert rows_per_burst(1000, 10, SPF, rows_remaining=100,
|
||||
memory_budget=1) == 1
|
||||
|
||||
|
||||
def test_rows_per_burst_rejects_degenerate_geometry():
|
||||
with pytest.raises(ValueError):
|
||||
rows_per_burst(100, 0, SPF, rows_remaining=1)
|
||||
|
||||
|
||||
# ── split_row_counts ─────────────────────────────────────────────────────────
|
||||
|
||||
def test_split_row_counts_differences_the_cumulative_counter():
|
||||
assert split_row_counts([4, 8, 12]) == [4, 4, 4]
|
||||
assert split_row_counts([4, 7, 12]) == [4, 3, 5]
|
||||
assert split_row_counts([]) == []
|
||||
|
||||
|
||||
def test_split_row_counts_rejects_a_counter_that_went_backwards():
|
||||
# Only happens if the acquisition restarted mid-burst, which would
|
||||
# misattribute every later row.
|
||||
with pytest.raises(RuntimeError, match="backwards"):
|
||||
split_row_counts([8, 4])
|
||||
|
||||
|
||||
# ── normalize_row ────────────────────────────────────────────────────────────
|
||||
|
||||
def test_normalize_row_passes_an_exact_row_through():
|
||||
buf = bytes(range(4 * SPF))
|
||||
assert bytes(normalize_row(buf, 0, 4, 4, SPF)) == buf
|
||||
|
||||
|
||||
def test_normalize_row_pads_a_short_row():
|
||||
buf = bytes(range(3 * SPF))
|
||||
out = bytes(normalize_row(buf, 0, 3, 4, SPF))
|
||||
assert len(out) == 4 * SPF
|
||||
assert out[:3 * SPF] == buf
|
||||
assert out[3 * SPF:] == bytes(SPF)
|
||||
|
||||
|
||||
def test_normalize_row_truncates_a_long_row():
|
||||
buf = bytes(range(6 * SPF))
|
||||
out = bytes(normalize_row(buf, 0, 6, 4, SPF))
|
||||
assert out == buf[:4 * SPF]
|
||||
|
||||
|
||||
def test_normalize_row_reads_at_an_offset():
|
||||
buf = bytes(range(8 * SPF))
|
||||
out = bytes(normalize_row(buf, 2 * SPF, 4, 4, SPF))
|
||||
assert out == buf[2 * SPF:6 * SPF]
|
||||
|
||||
|
||||
def test_normalize_row_pads_a_buffer_that_ends_early():
|
||||
"""Defensive: a truncated transfer must not shorten the row on disk."""
|
||||
out = bytes(normalize_row(bytes(2 * SPF), 0, 4, 4, SPF))
|
||||
assert len(out) == 4 * SPF
|
||||
|
||||
|
||||
# ── frame_means_block ────────────────────────────────────────────────────────
|
||||
|
||||
def test_frame_means_block_is_per_frame():
|
||||
buf = bytes([1] * SPF + [3] * SPF)
|
||||
assert frame_means_block(buf, 0, 2, SPF) == [1.0, 3.0]
|
||||
|
||||
|
||||
def test_frame_means_block_reads_signed_samples_at_an_offset():
|
||||
buf = bytes([0] * SPF) + bytes([0xFF] * SPF) # 0xFF == -1 as int8
|
||||
assert frame_means_block(buf, SPF, 1, SPF) == [-1.0]
|
||||
Reference in New Issue
Block a user