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>

core/scan_engine.py

@@ -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,7 +256,13 @@ 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
+        # 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:
@@ -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,9 +366,28 @@ 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
+            if self._burst_mode:
-            # the scope's FastFrame count must be re-armed per angle.
+                # 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()
@@ -363,31 +411,208 @@ class ScanEngine:
             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)
+            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)
 
-            result.angles_acquired.append(ai)
+    def _write_row(self, scan_file, samples_per_frame: int, row_idx: int,
-
+                   n_frames: int):
-    def _write_row(self, scan_file, samples_per_frame: int, row_idx: 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)
+                scan_file.write(bytes(ch_bytes))
-                for _ in range(scope_sras.frames_acquired(scope)):
-                    scan_file.write(zero_frame)
                 continue
 
             self._cb.on_status(f"Fetching CH{ch} data …")
             waveforms = scope_sras.transfer_channel(scope, ch)
-            if ch == 4 and waveforms:
+            if ch == SCAN_CHANNELS[0]:
-                self._cb.on_dc_bias(row_idx + 1, scope_sras.frame_means(waveforms))
+                self._warn_frame_delta(row_idx, len(waveforms), n_frames)
-            for w in waveforms:
+            row = scope_burst.normalize_row(
-                scan_file.write(w)
+                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)

core/scope_burst.py

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

core/scope_sras.py

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

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 = []
+        return self._frames(self._source, self._acquired)
-        for i in range(self._n_frames):
+
-            frames.append(bytes((self.frame_seq + i + s) % 256
+    def transfer_fastframe_bulk(self, frame_count, samples_per_frame,
-                                for s in range(self.samples_per_frame)))
+                                bytes_per_sample=1):
-        self.frame_seq += 1
+        self._t.record("transfer_fastframe_bulk", self._source, frame_count)
-        return frames
+        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):
+def make_plan(num_angles=1, y_delta=0.005):
-    # Small ROI well inside the stage limits: 1 row, few frames per angle.
+    # Small ROI well inside the stage limits: few rows, few frames per angle.
-    return build_plan(40.0, 30.0, 0.02, 0.005, num_angles, 0.01,
+    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, max_frames=max_frames)
-    scope = FakeScope(trace, samples_per_frame=SPF)
+    stage = FakeStage(trace, scope=scope)
     t3r = FakeT3R(trace, **kw)
     rotator = RotationAxis(t3r, RotationSettings())
-    plan = make_plan(num_angles)
+    plan = plan if plan is not None else make_plan(num_angles)
-    engine = ScanEngine(stage, scope, rotator, plan, tmp_path / "out.sras",
+    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]