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>

tests/fakes.py

@@ -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):

tests/test_scan_engine.py

@@ -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

tests/test_scope_burst.py

@@ -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]