Pre-scan angle inspection: park the rig per angle, read the response on the scope

A 9-angle scan takes hours, and an angle that responds poorly still produces
rows that look structurally fine in the file — the SAW packet is just not
there. This lets the operator walk the angles first, parking the rig at a
random point in each, and judge the response before committing to the run.

Nothing reads the scope. The operator inspects the instrument directly, so
there is no transfer path, no plotting, and no waveform crossing the module
boundary — test_inspection_never_reads_a_waveform_back pins that, since it is
the kind of premise a later change erodes without noticing.

core/scope_inspect.py — the scope state worth looking at, which is not the
scan's state:
- plain rising-edge trigger on CH2 at 2.0 V, not the scan's logic AND of the
  laser pulse and the stage gate, so a stationary stage still triggers
- FastFrame off, SAMPLE (no averaging) — a weak or intermittent response is
  exactly what is being looked for, and averaging would hide it
- free-running (STOPAfter RUNSTop + STATE RUN) so the trace keeps updating
  while the operator looks at it
- CH1 keeps the acquisition front-end verbatim, so what is on screen is what
  a scan would record
- CH3/CH4 become bias monitors sharing one scale and position, since the
  comparison is by eye and only works if a division means the same on each.
  100 mV/div with ground 3.5 divisions below centre puts 0–700 mV on screen
  with headroom on an 8- or 10-division graticule (the signal never goes
  negative, hence moving the trace down).

core/angle_inspect.py — AngleInspector, headless and Qt-free like ScanEngine.
Points are drawn from the angle's own bounding box: Y from its actual row
positions and X uniformly across its data window, so the point is somewhere
the scan would really sample rather than merely inside the box. New Point
re-rolls without rotating, which is what separates a bad spot on the sample
from a bad angle. The stage gate is held off throughout, and the rotator goes
home on stop.

gui/inspect_bridge.py — QtAngleInspector on the existing QueueWorker base.
Inspection is click-driven rather than one long run, so the worker blocks on
its queue between commands and an open window costs nothing. BBD position
polling is suppressed while inspecting, for the same reason the scan does it:
the shared TX queue.

sc3_aui_app.py — AngleInspectWindow (angle list, prev/next, New Point) driven
off the plan currently entered in the scan panel, so it inspects exactly the
scan about to be run. Navigation locks while the stage moves. The list syncs
via itemClicked rather than currentRowChanged, so echoing the worker's
position back does not re-trigger the move it is reporting.

README picks up the new modules, and scope_burst.py which the previous merge
left out of the structure listing. 114 tests passing, ruff clean.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

README.md

@@ -12,6 +12,8 @@ scanengine-3 is a unified platform for scanning acoustic microscopy and precisio
 - **Laser Systems**: Helios pulsed laser and Genesis CW laser control
 - **Data Acquisition**: Tektronix oscilloscope integration with fast-frame support
 - **Scan Planning**: Automated raster scan generation and execution
+- **Angle Inspection**: Park the rig at random points across a plan's angles
+  to check the SAW response on the scope before committing to a long scan
 - **Real-time Monitoring**: Live status updates and progress tracking
 
 ## Hardware Components
@@ -55,6 +57,9 @@ scanengine-3/
 │   ├── scan_geometry.py           # ScanPlan, rotated-bbox planning, limits
 │   ├── scan_resume.py             # Resume planning (frontier rule)
 │   ├── scope_sras.py              # Oscilloscope SCPI policy for SRAS
+│   ├── scope_burst.py             # Burst-mode FastFrame sizing + row splitting
+│   ├── scope_inspect.py           # Scope setup for pre-scan angle inspection
+│   ├── angle_inspect.py           # AngleInspector — park on a point per angle
 │   ├── rotation.py                # GR rotation axis settings + moves
 │   ├── sras_format.py             # v6 .sras writer/reader (memory-mapped)
 │   ├── sras_analysis.py           # Image reducers + SAW matched filter
@@ -72,6 +77,7 @@ scanengine-3/
 │
 ├── gui/                           # Shared PyQt6 layer
 │   ├── scan_bridge.py             # QtScanController over core.scan_engine
+│   ├── inspect_bridge.py          # QtAngleInspector over core.angle_inspect
 │   ├── qt_t3r.py                  # Qt adapter over the T3R driver
 │   ├── qt_workers.py              # QueueWorker / PollingQueueWorker bases
 │   └── widgets.py                 # ConnectionBar, LogConsole, PortSelector…

core/angle_inspect.py

@@ -0,0 +1,251 @@
+"""Pre-scan angle inspection: park the rig on a point and let the operator look.
+
+A multi-angle scan can take hours, and an angle that responds poorly produces
+rows that look fine in the file but carry no usable SAW packet.  This drives
+the rig through the same angles the scan will use, parking at a random point
+inside each angle's own bounding box so the response can be judged on the
+oscilloscope before committing to the run.
+
+Headless and Qt-free, like ScanEngine: gui/inspect_bridge.py wraps it.
+
+No waveform ever crosses this boundary.  The operator reads the scope screen
+directly; this module's job is only to put the hardware in the right place and
+the scope in a state worth looking at (see core.scope_inspect).
+"""
+from __future__ import annotations
+
+import logging
+import random
+from dataclasses import dataclass
+from typing import Callable
+
+from core import scope_inspect
+from core.rotation import RotationAxis
+from core.scan_engine import (
+    AXIS_X, AXIS_Y, SCAN_ACCEL_MM_S2, SCAN_VELOCITY_MM_S,
+)
+from core.scan_geometry import DEFAULT_STAGE_LIMITS, ScanPlan, StageLimits
+
+logger = logging.getLogger(__name__)
+
+# Positioning moves only — no data is taken while moving, so there is no
+# reason to cross the tray at full scan velocity.
+INSPECT_VELOCITY_MM_S = SCAN_VELOCITY_MM_S / 2.0
+
+
+@dataclass(frozen=True)
+class InspectionPoint:
+    """Where the rig is parked, and which angle it is parked for."""
+    angle_idx: int
+    angle_deg: float
+    x_mm: float
+    y_mm: float
+
+    def describe(self) -> str:
+        return (f"Angle {self.angle_idx + 1} ({self.angle_deg:.1f}°)  "
+                f"X={self.x_mm:.3f} mm  Y={self.y_mm:.3f} mm")
+
+
+@dataclass
+class InspectCallbacks:
+    """Progress reporting.  Defaults are no-ops so the core needs no front end."""
+    on_status: Callable[[str], None] = lambda msg: None
+    on_point: Callable[[InspectionPoint], None] = lambda pt: None
+    on_busy: Callable[[bool], None] = lambda busy: None
+
+
+@dataclass
+class _State:
+    angle_idx: int = 0
+    point: InspectionPoint | None = None
+    started: bool = False
+    rotator_ready: bool = False
+
+
+class AngleInspector:
+    """Drives stage + rotator to inspection points across a plan's angles."""
+
+    def __init__(self, stage, scope, rotator: RotationAxis | None,
+                 plan: ScanPlan,
+                 callbacks: InspectCallbacks | None = None,
+                 limits: StageLimits = DEFAULT_STAGE_LIMITS,
+                 rng: random.Random | None = None):
+        self._stage = stage
+        self._scope = scope
+        self._rotator = rotator
+        self._plan = plan
+        self._cb = callbacks if callbacks is not None else InspectCallbacks()
+        self._limits = limits
+        # Injectable so tests can pin the point selection.
+        self._rng = rng if rng is not None else random.Random()
+        self._st = _State()
+
+    # ── Introspection ─────────────────────────────────────────────────────────
+
+    @property
+    def n_angles(self) -> int:
+        return self._plan.n_angles
+
+    @property
+    def angle_idx(self) -> int:
+        return self._st.angle_idx
+
+    @property
+    def current_point(self) -> InspectionPoint | None:
+        return self._st.point
+
+    def angle_labels(self) -> list[str]:
+        return [f"Angle {i + 1}/{self.n_angles} — {pa.angle_deg:.2f}°"
+                for i, pa in enumerate(self._plan.per_angle)]
+
+    # ── Lifecycle ─────────────────────────────────────────────────────────────
+
+    def start(self) -> InspectionPoint:
+        """Configure the hardware and park on the first angle."""
+        if self._stage is None:
+            raise RuntimeError("BBD202 not connected")
+        if self._scope is None:
+            raise RuntimeError("Oscilloscope not connected")
+
+        self._st.rotator_ready = (self._rotator is not None
+                                  and self._rotator.is_available)
+        if self.n_angles > 1 and not self._st.rotator_ready:
+            raise RuntimeError(
+                f"Inspecting {self.n_angles} angles requires the T3R rotation "
+                "stage (GR-axis), but it is not connected. Connect T3R from "
+                "the T3R panel, or inspect a single-angle plan."
+            )
+
+        self._cb.on_busy(True)
+        try:
+            self._cb.on_status("Configuring stage for inspection …")
+            ctrl = self._stage
+            for axis in (AXIS_X, AXIS_Y):
+                ctrl.set_velocity_params(axis,
+                                         max_velocity=INSPECT_VELOCITY_MM_S,
+                                         acceleration=SCAN_ACCEL_MM_S2)
+            # Nothing here is gated, and an armed trigger output would keep
+            # driving the gate line on every positioning move.
+            ctrl.set_trigger_gate_off(AXIS_X)
+
+            if self._st.rotator_ready:
+                self._cb.on_status("Configuring GR axis …")
+                self._rotator.configure()
+
+            self._cb.on_status("Configuring oscilloscope for inspection …")
+            scope_inspect.configure_inspection(self._scope)
+
+            self._st.started = True
+            return self._goto(0, new_point=True)
+        finally:
+            self._cb.on_busy(False)
+
+    def stop(self) -> None:
+        """Stop the sweep and send the rotator home.  Safe to call twice."""
+        if not self._st.started:
+            return
+        self._st.started = False
+        self._cb.on_busy(True)
+        try:
+            try:
+                scope_inspect.stop_inspection(self._scope)
+            except Exception:
+                logger.exception("Could not stop the inspection acquisition")
+            if self._st.rotator_ready and abs(self._rotator.current_deg) > 0.001:
+                self._cb.on_status("Returning GR to home …")
+                try:
+                    self._rotator.return_to_zero()
+                except Exception:
+                    logger.exception("GR return-to-home failed")
+            self._cb.on_status("Inspection finished.")
+        finally:
+            self._cb.on_busy(False)
+
+    # ── Navigation ────────────────────────────────────────────────────────────
+
+    def goto_angle(self, angle_idx: int) -> InspectionPoint:
+        """Rotate to `angle_idx` and park on a fresh random point there."""
+        self._require_started()
+        self._cb.on_busy(True)
+        try:
+            return self._goto(angle_idx, new_point=True)
+        finally:
+            self._cb.on_busy(False)
+
+    def next_angle(self) -> InspectionPoint:
+        """Advance one angle, wrapping at the end."""
+        return self.goto_angle((self._st.angle_idx + 1) % self.n_angles)
+
+    def prev_angle(self) -> InspectionPoint:
+        return self.goto_angle((self._st.angle_idx - 1) % self.n_angles)
+
+    def new_point(self) -> InspectionPoint:
+        """Re-roll the point within the current angle, without rotating.
+
+        One point can be unrepresentative — a bad spot on the sample looks the
+        same as a bad angle.  Re-rolling a few times is how you tell them
+        apart, so this deliberately skips the rotation.
+        """
+        self._require_started()
+        self._cb.on_busy(True)
+        try:
+            return self._goto(self._st.angle_idx, new_point=True, rotate=False)
+        finally:
+            self._cb.on_busy(False)
+
+    # ── Internals ─────────────────────────────────────────────────────────────
+
+    def _require_started(self):
+        if not self._st.started:
+            raise RuntimeError("Inspection has not been started")
+
+    def _goto(self, angle_idx: int, new_point: bool,
+              rotate: bool = True) -> InspectionPoint:
+        if not 0 <= angle_idx < self.n_angles:
+            raise IndexError(
+                f"Angle {angle_idx} out of range (plan has {self.n_angles})")
+
+        pa = self._plan.per_angle[angle_idx]
+        self._st.angle_idx = angle_idx
+
+        if rotate and self._st.rotator_ready:
+            delta = pa.angle_deg - self._rotator.current_deg
+            if abs(delta) > 0.001:
+                self._cb.on_status(
+                    f"Rotating GR to {pa.angle_deg:.1f}° (Δ{delta:+.1f}°) …")
+                self._rotator.rotate_to(pa.angle_deg)
+
+        point = self._pick_point(angle_idx) if new_point else self._st.point
+
+        self._cb.on_status(f"Moving to {point.describe()} …")
+        # Y first, then X — the same order the scan uses to reach a row.
+        self._stage.move_axis_absolute(AXIS_Y, point.y_mm, timeout=60.0)
+        self._stage.move_axis_absolute(AXIS_X, point.x_mm, timeout=60.0)
+
+        self._st.point = point
+        self._cb.on_point(point)
+        self._cb.on_status(f"Parked at {point.describe()}")
+        return point
+
+    def _pick_point(self, angle_idx: int) -> InspectionPoint:
+        """A random point on this angle's scan grid.
+
+        Y is drawn from the angle's actual row positions and X uniformly from
+        its data window, so the point is somewhere the scan would really
+        sample — not merely inside the bounding box.
+        """
+        pa = self._plan.per_angle[angle_idx]
+        if not pa.y_positions:
+            raise ValueError(f"Angle {angle_idx + 1} has no rows to inspect")
+
+        y = self._rng.choice(pa.y_positions)
+        x = self._rng.uniform(pa.x_start, pa.x_start + pa.x_delta)
+
+        lim = self._limits
+        if not (lim.x_min <= x <= lim.x_max and lim.y_min <= y <= lim.y_max):
+            raise ValueError(
+                f"Inspection point X={x:.3f} Y={y:.3f} is outside the stage "
+                f"travel ({lim.x_min}–{lim.x_max} × {lim.y_min}–{lim.y_max} mm)"
+            )
+        return InspectionPoint(angle_idx=angle_idx, angle_deg=pa.angle_deg,
+                               x_mm=x, y_mm=y)

core/config.py

@@ -23,6 +23,8 @@ class ScanDefaults:
     oscope_ip: str = "192.168.0.1"
     save_dir: str = str(DEFAULTS_PATH.parent / "scans")
     helios_port: str = "/dev/ttyUSB2"
+    burst_mode: bool = False
+    strict_rows: bool = False
 
     @classmethod
     def load(cls, path: Path = DEFAULTS_PATH) -> "ScanDefaults":

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, strict_rows: bool = False):
         self._stage = stage
         self._scope = scope
         self._rotator = rotator
@@ -103,6 +104,14 @@ 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
+        # Strict row packing stops the scan on a frame-count mismatch
+        # instead of squaring the row up (see _check_frame_delta).
+        self._strict_rows = strict_rows
+        self._max_frames = 0
+        self._preflight_done = False
 
         self._abort = threading.Event()
         self._resume_event = threading.Event()
@@ -207,6 +216,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 +259,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 +313,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 +342,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 +369,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
-            # the scope's FastFrame count must be re-armed per angle.
+            if self._burst_mode:
+                # Burst mode sizes the FastFrame count from the scope's whole
+                # capacity instead (see scope_burst.start_burst), so there is
+                # nothing to re-arm per angle here.
+                self._scan_rows_burst(scan_file, pa, ai, n_angles,
+                                      samples_per_frame, result, x_ramp_total)
+            else:
+                # Each angle's bounding box gives it its own points/row count,
+                # so the scope's FastFrame count must be re-armed per angle.
                 scope.set_fastframe_count(pa.n_frames)
+                self._scan_rows_serial(scan_file, pa, ai, n_angles,
+                                       samples_per_frame, result, x_ramp_total)
+
+            result.angles_acquired.append(ai)
+
+    # ── Per-row acquisition (one FastFrame acquisition per row) ───────────────
+
+    def _scan_rows_serial(self, scan_file, pa, ai: int, n_angles: int,
+                          samples_per_frame: int, result: ScanResult,
+                          x_ramp_total: float):
+        ctrl = self._stage
+        scope = self._scope
 
         for ri, y_pos in enumerate(pa.y_positions):
             self._pause_point()
@@ -363,31 +414,224 @@ 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):
+    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._check_frame_delta(row_idx, len(waveforms), n_frames)
+            row = scope_burst.normalize_row(
+                b"".join(waveforms), 0, len(waveforms), n_frames, samples_per_frame)
+            if ch == 4:
+                self._cb.on_dc_bias(row_idx + 1, scope_burst.frame_means_block(
+                    row, 0, n_frames, samples_per_frame))
+            scan_file.write(row)
+
+    # ── Burst acquisition (many whole rows per FastFrame acquisition) ─────────
+
+    def _scan_rows_burst(self, scan_file, pa, ai: int, n_angles: int,
+                         samples_per_frame: int, result: ScanResult,
+                         x_ramp_total: float):
+        """Acquire the angle in bursts of as many whole rows as the scope holds.
+
+        One ACQuire:STATE RUN spans the whole burst, so the gate is armed only
+        for each acquiring pass and dropped for the flyback — otherwise the
+        return move would reach max velocity and inject frames between rows.
+        """
+        scope = self._scope
+        n_frames = pa.n_frames
+        x_lead_in = pa.x_start - x_ramp_total
+        x_end = pa.x_start + pa.x_delta + x_ramp_total
+
+        if not self._preflight_done:
+            # Once per scan: the gate wiring can't change between angles, and
+            # the check costs two row-times.
+            self._gate_off_preflight(x_lead_in, x_end)
+            self._preflight_done = True
+
+        row = 0
+        while row < pa.n_rows:
+            self._pause_point()
+            n_burst = scope_burst.rows_per_burst(
+                self._max_frames, n_frames, samples_per_frame, pa.n_rows - row)
+            self._cb.on_status(
+                f"Angle {ai+1}/{n_angles}  Rows {row+1}-{row+n_burst}/{pa.n_rows} "
+                f"in one acquisition ({n_burst * n_frames} frames) …"
+            )
+
+            burst_start = scan_file.tell()
+            cumulative = []
+            baseline = scope_burst.start_burst(scope, self._max_frames)
+            try:
+                for r in range(n_burst):
+                    self._check_abort()
+                    self._cb.on_row_started(row + r + 1, pa.n_rows,
+                                            ai + 1, n_angles)
+                    self._acquire_gated_row(pa.y_positions[row + r],
+                                            x_lead_in, x_end)
+                    total = scope_burst.frames_acquired(scope)
+                    if total >= self._max_frames:
+                        raise RuntimeError(
+                            f"FastFrame buffer full ({total}/{self._max_frames} "
+                            f"frames) at row {row + r + 1} — later rows in this "
+                            "burst would be misattributed.  Raise "
+                            "scope_burst.BURST_FRAME_HEADROOM and rerun."
+                        )
+                    cumulative.append(total - baseline)
+            finally:
+                scope_burst.stop_burst(scope)
+
+            counts = scope_burst.split_row_counts(cumulative)
+            self._write_burst(scan_file, burst_start, row, counts,
+                              n_frames, samples_per_frame)
+
+            for r in range(n_burst):
+                result.rows_written += 1
+                self._cb.on_row_done(row + r + 1, pa.n_rows, ai + 1, n_angles)
+            row += n_burst
+
+    def _acquire_gated_row(self, y_pos: float, x_lead_in: float, x_end: float):
+        """One row: step Y, fly back gated off, then acquire on the +X pass."""
+        ctrl = self._stage
+        ctrl.move_axis_absolute(AXIS_Y, y_pos, timeout=60.0)
+        ctrl.move_axis_absolute(AXIS_X, x_lead_in, timeout=30.0)
+        ctrl.arm_scan_gate(AXIS_X, True)
+        ctrl.move_axis_absolute(AXIS_X, x_end, timeout=120.0)
+        ctrl.arm_scan_gate(AXIS_X, False)
+        time.sleep(scope_burst.BURST_ROW_SETTLE_S)
+
+    def _gate_off_preflight(self, x_lead_in: float, x_end: float):
+        """Prove the gate really gates before trusting a multi-row burst.
+
+        The value that makes the BBD trigger output idle low is not settled by
+        the protocol docs (see apt_constants.TRIGOUT_GATE_OFF), and getting it
+        wrong fills every burst with flyback frames that silently shift the
+        file.  The scope already measures the gate on CH3, so this needs no
+        bench probe: one gated-off flyback must acquire nothing, and one gated
+        pass must acquire something — the second half is what stops a dark
+        laser from making the first half pass vacuously.
+
+        Leaves the stage parked at x_end, where the burst loop expects it.
+        """
+        ctrl, scope = self._stage, self._scope
+        self._cb.on_status("Burst preflight: checking the stage gate …")
+
+        ctrl.arm_scan_gate(AXIS_X, False)
+        ctrl.move_axis_absolute(AXIS_X, x_end, timeout=120.0)
+        baseline = scope_burst.start_burst(scope, self._max_frames)
+        ctrl.move_axis_absolute(AXIS_X, x_lead_in, timeout=120.0)
+        scope_burst.stop_burst(scope)
+        leaked = scope_burst.frames_acquired(scope) - baseline
+
+        ctrl.arm_scan_gate(AXIS_X, True)
+        baseline = scope_burst.start_burst(scope, self._max_frames)
+        ctrl.move_axis_absolute(AXIS_X, x_end, timeout=120.0)
+        ctrl.arm_scan_gate(AXIS_X, False)
+        scope_burst.stop_burst(scope)
+        gated = scope_burst.frames_acquired(scope) - baseline
+
+        if gated <= 0:
+            raise RuntimeError(
+                "Burst preflight: no frames acquired with the gate armed. "
+                "Check that the Genesis laser is pulsing (CH2) and that the "
+                "BBD X trigger output reaches CH3 before scanning."
+            )
+        if leaked:
+            raise RuntimeError(
+                f"Burst preflight: {leaked} frame(s) acquired during a flyback "
+                "that should have been gated off — the BBD trigger output is "
+                "not idling low.  Set apt_constants.TRIGOUT_GATE_OFF to "
+                "TriggerBitsServo.TRIGOUT_HIGH and retry, or use per-row "
+                "acquisition."
+            )
+        self._cb.on_status(
+            f"Burst preflight OK ({gated} frames gated on, 0 leaked).")
+
+    def _write_burst(self, scan_file, burst_start: int, first_row: int,
+                     counts: list[int], n_frames: int, samples_per_frame: int):
+        """Deinterleave one burst into the file's per-row, per-channel blocks.
+
+        The wire is channel-major (every row of CH1, then every row of CH4);
+        the file is row-major with channels inner.  Writing one channel at a
+        time to strided offsets keeps peak memory at a single channel's burst
+        instead of the whole thing.
+        """
+        scope = self._scope
+        ch_bytes = n_frames * samples_per_frame
+        row_bytes = len(SCAN_CHANNELS) * ch_bytes
+        total_frames = sum(counts)
+
+        for r, count in enumerate(counts):
+            self._check_frame_delta(first_row + r, count, n_frames)
+
+        for ch_idx, ch in enumerate(SCAN_CHANNELS):
+            if ch == 3:
+                self._cb.on_status("Writing zeroed CH3 frames …")
+                blob = None
+            else:
+                self._cb.on_status(
+                    f"Fetching CH{ch} burst ({total_frames} frames) …")
+                blob = scope_burst.transfer_burst(scope, ch, total_frames,
+                                                  samples_per_frame)
+            src = 0
+            zeros = bytes(ch_bytes) if blob is None else None
+            for r, count in enumerate(counts):
+                scan_file.seek(burst_start + r * row_bytes + ch_idx * ch_bytes)
+                if blob is None:
+                    scan_file.write(zeros)
+                else:
+                    row = scope_burst.normalize_row(
+                        blob, src, count, n_frames, samples_per_frame)
+                    if ch == 4:
+                        self._cb.on_dc_bias(
+                            first_row + r + 1,
+                            scope_burst.frame_means_block(
+                                row, 0, n_frames, samples_per_frame))
+                    scan_file.write(row)
+                src += count * samples_per_frame
+            del blob
+
+        scan_file.seek(burst_start + len(counts) * row_bytes)
+
+    def _check_frame_delta(self, row_idx: int, count: int, n_frames: int):
+        """Decide what to do with a row that did not acquire n_frames frames.
+
+        v6 declares n_frames per row in the header and has no per-row length
+        field, so a mismatched row cannot just be written as-is — that would
+        shift every later row in the file.  The only two safe options are to
+        square it up or to stop, which is what strict_rows selects between.
+
+        Called before anything for the row is written (CH1 leads
+        SCAN_CHANNELS), so raising here leaves no partial row behind.
+        """
+        if count == n_frames:
+            return
+        verb = "zero-padded" if count < n_frames else "truncated"
+        if self._strict_rows:
+            raise RuntimeError(
+                f"Row {row_idx + 1}: {count} frames acquired, {n_frames} "
+                f"expected. Strict row packing is on, so the scan stops here "
+                f"rather than writing a row that would be {verb}."
+            )
+        msg = (f"Row {row_idx + 1}: {count} frames acquired, {n_frames} "
+               f"expected — {verb} to keep the file layout intact.")
+        logger.warning(msg)
+        self._cb.on_status(msg)

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_inspect.py

@@ -0,0 +1,104 @@
+"""Oscilloscope configuration for pre-scan angle inspection.
+
+Inspection is read-on-the-instrument: nothing in this module transfers or
+plots waveform data.  The app puts the scope into a free-running, edge-
+triggered state and drives the stage to the point being inspected; the
+operator judges the SAW response and the bias levels on the scope screen.
+
+That split is deliberate.  A scan's acquisition trigger is the logic AND of
+the laser pulse and the stage's max-velocity gate, and its transfers are
+FastFrame blocks — neither is useful for looking at one point by eye.  Here
+the trigger is a plain edge on the laser pulse, FastFrame is off, and the
+acquisition free-runs, so the display updates continuously while the stage
+sits still.
+
+CH1 keeps the acquisition front-end so what is on screen is what a scan would
+record.  CH3 and CH4 are rescaled as DC bias monitors (see BIAS_* below).
+"""
+from __future__ import annotations
+
+import logging
+from dataclasses import replace
+
+from core.scope_sras import SAMPLE_RATE_HZ, SRAS_CHANNELS, configure_channels
+
+logger = logging.getLogger(__name__)
+
+# CH2 carries the laser pulse.  The scan triggers it at 0.5 V as one term of a
+# logic AND; inspection triggers well above that so a slow edge or a noisy
+# baseline cannot free-run the display.
+INSPECT_TRIG_LEVEL_V = 2.0
+
+# CH3/CH4 are the DC bias monitors during inspection.  The signal never goes
+# negative and spans roughly 0–700 mV, so both channels get the *same* scale
+# and position — the point of inspecting them is comparing the two by eye, and
+# that only works if a division means the same thing on each.
+#
+# Ground sits BIAS_POSITION_DIV divisions below centre, which puts the whole
+# 0–700 mV range above the centre line with a little room underneath for
+# undershoot.  With 100 mV/div and ground 3.5 divisions low, the visible window
+# runs from about -50 mV to +750 mV on an 8-division display and wider on a
+# 10-division one, so 0–700 mV sits comfortably inside either.
+BIAS_CHANNELS = (3, 4)
+BIAS_WINDOW_V = 0.700
+BIAS_SCALE_V_DIV = 0.100
+BIAS_POSITION_DIV = -3.5
+
+BIAS_LABELS = {3: "Bias - A", 4: "Bias - B"}
+
+
+def inspect_channel_profiles() -> dict:
+    """Channel front-end config for inspection.
+
+    CH1 and CH2 are the acquisition profiles verbatim.  CH3 and CH4 differ
+    only in label, scale and position — termination, coupling and bandwidth
+    stay as the scan sets them, so the bias reading is the same measurement
+    the scan records, just displayed usefully.
+    """
+    profiles = dict(SRAS_CHANNELS)
+    for ch in BIAS_CHANNELS:
+        profiles[ch] = replace(
+            SRAS_CHANNELS[ch],
+            label=BIAS_LABELS[ch],
+            scale_v_div=BIAS_SCALE_V_DIV,
+            position_div=BIAS_POSITION_DIV,
+        )
+    return profiles
+
+
+def configure_inspection(scope) -> None:
+    """Put the scope into free-running inspection mode.
+
+    Leaves the acquisition running, so the display stays live while the
+    operator moves between angles and points.
+    """
+    configure_channels(scope, inspect_channel_profiles())
+
+    # Plain edge trigger on the laser pulse — no logic pattern, so the stage
+    # gate plays no part and a stationary stage still triggers.
+    scope.write("TRIGger:A:TYPe EDGE")
+    scope.set_trigger_source(2)
+    scope.set_trigger_slope("RISE")
+    scope.set_trigger_level(2, INSPECT_TRIG_LEVEL_V)
+    scope.set_trigger_mode("NORMAL")
+
+    # No averaging: a weak or intermittent SAW response is exactly what the
+    # operator is looking for, and averaging would hide it.
+    scope.set_acquire_mode("SAMPLE")
+    scope.set_fastframe_state(False)
+
+    scope.set_sample_rate(SAMPLE_RATE_HZ)
+    scope.write("HORizontal:POSition 30")
+
+    # Free-run rather than single-sequence, so the trace keeps updating.
+    scope.write("ACQuire:STOPAfter RUNSTop")
+    scope.write("ACQuire:STATE RUN")
+
+
+def stop_inspection(scope) -> None:
+    """Halt the free-running acquisition.
+
+    The next scan reconfigures the scope from scratch, so this only needs to
+    stop the sweep — it does not try to restore the acquisition profile.
+    """
+    scope.write("ACQuire:STATE STOP")

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

gui/inspect_bridge.py

@@ -0,0 +1,113 @@
+"""Qt bridge over the headless AngleInspector.
+
+Inspection is command-driven rather than one long run: the operator clicks an
+angle, waits for the stage to park, looks at the scope, clicks again.  That is
+exactly the shape QueueWorker exists for — it blocks on the queue between
+commands instead of polling, so an inspection window left open costs nothing.
+
+Every stage move and rotation blocks for seconds, so all of it runs on this
+worker's thread; the window only ever enqueues and reacts to signals.
+"""
+from __future__ import annotations
+
+import traceback
+
+from PyQt6.QtCore import pyqtSignal
+
+from core.angle_inspect import AngleInspector, InspectCallbacks
+from gui.qt_workers import QueueWorker
+
+
+class QtAngleInspector(QueueWorker):
+    """Runs an AngleInspector on its own QThread and republishes its events."""
+
+    ready = pyqtSignal(object)            # InspectionPoint — start() succeeded
+    start_failed = pyqtSignal(str)
+    point_changed = pyqtSignal(object)    # InspectionPoint
+    status_msg = pyqtSignal(str)
+    busy_changed = pyqtSignal(bool)       # True while a move is in flight
+    stopped = pyqtSignal()
+
+    def __init__(self, stage, scope, rotator, plan, on_inspect_active=None):
+        super().__init__()
+        self._on_inspect_active = on_inspect_active
+
+        callbacks = InspectCallbacks(
+            on_status=self.status_msg.emit,
+            on_point=self.point_changed.emit,
+            on_busy=self.busy_changed.emit,
+        )
+        self._inspector = AngleInspector(stage, scope, rotator, plan,
+                                         callbacks=callbacks)
+        self._handlers = {
+            "start": self._do_start,
+            "goto": self._do_goto,
+            "new_point": self._do_new_point,
+            "stop": self._do_stop,
+        }
+
+    # ── Introspection (safe from the GUI thread: reads the plan, not the rig) ──
+
+    def angle_labels(self) -> list[str]:
+        return self._inspector.angle_labels()
+
+    @property
+    def n_angles(self) -> int:
+        return self._inspector.n_angles
+
+    # ── Command submission (GUI thread) ───────────────────────────────────────
+
+    def request_start(self):
+        self._enqueue("start")
+
+    def request_goto(self, angle_idx: int):
+        self._enqueue("goto", angle_idx=angle_idx)
+
+    def request_new_point(self):
+        self._enqueue("new_point")
+
+    def request_stop(self):
+        self._enqueue("stop")
+
+    # ── Handlers (worker thread) ──────────────────────────────────────────────
+
+    def _do_start(self):
+        if self._on_inspect_active is not None:
+            self._on_inspect_active(True)
+        try:
+            point = self._inspector.start()
+        except Exception as exc:
+            traceback.print_exc()
+            if self._on_inspect_active is not None:
+                self._on_inspect_active(False)
+            self.start_failed.emit(str(exc))
+            return
+        self.ready.emit(point)
+
+    def _do_goto(self, angle_idx: int):
+        self._inspector.goto_angle(angle_idx)
+
+    def _do_new_point(self):
+        self._inspector.new_point()
+
+    def _do_stop(self):
+        try:
+            self._inspector.stop()
+        finally:
+            if self._on_inspect_active is not None:
+                self._on_inspect_active(False)
+            self.stopped.emit()
+
+    def _on_stop(self):
+        """Worker loop exiting — make sure the rig is left in a safe state.
+
+        Covers the case where the window is closed without a clean stop
+        command reaching the queue.
+        """
+        try:
+            self._inspector.stop()
+        except Exception:
+            traceback.print_exc()
+        finally:
+            if self._on_inspect_active is not None:
+                self._on_inspect_active(False)

gui/scan_bridge.py

@@ -32,7 +32,8 @@ class QtScanController(QObject):
     paused_changed = pyqtSignal(bool)                 # True while paused at a row boundary
 
     def __init__(self, stage, scope, rotator, plan, out_path,
-                 resume=None, on_scan_active=None):
+                 resume=None, on_scan_active=None, burst_mode=False,
+                 strict_rows=False):
         super().__init__()
         self._prompt_event = threading.Event()
         self._on_scan_active = on_scan_active
@@ -47,7 +48,9 @@ class QtScanController(QObject):
             prompt=self._blocking_prompt,
         )
         self._engine = ScanEngine(stage, scope, rotator, plan, out_path,
-                                  resume=resume, callbacks=callbacks)
+                                  resume=resume, callbacks=callbacks,
+                                  burst_mode=burst_mode,
+                                  strict_rows=strict_rows)
 
     # ── Engine control (called from the GUI thread) ───────────────────────────
 

hardware/pybbd202/apt_constants.py

@@ -45,5 +45,20 @@ class TriggerBitsServo(IntFlag):
     TRIGOUT_MAXV = TRIGOUT_HIGH | TRIGOUT_MAXVELOCITY
 
 
+# Gate off: no trigger-out function selected, so the pin idles inactive.
+#
+# Treat this as unverified until it has been checked on the rig.  §7.6 of
+# docs/hardware/BBD203_Communications_Protocol.md documents `mode` as an
+# enumeration capping at 0x11, which flatly contradicts the bitmask this
+# driver actually sends (TRIGOUT_MAXV = 0x90, known working), so the doc
+# cannot settle what makes the pin idle low.  Under the bitmask reading 0x00
+# clears everything and the pin sits low.  If instead TRIGOUT_HIGH is an
+# active-high *polarity* bit, clearing it means active-low and the pin idles
+# HIGH — which in burst mode floods the acquisition with flyback frames.
+# ScanEngine's gate-off preflight catches that; if it trips, change this to
+# TriggerBitsServo.TRIGOUT_HIGH.
+TRIGOUT_GATE_OFF = TriggerBitsServo(0)
+
+
 
 

hardware/pybbd202/bbd20x.py

@@ -7,7 +7,7 @@
 import time
 from threading import Thread, Event
 from queue import Queue, Empty
-from .apt_constants import StatusBits, TriggerBitsServo
+from .apt_constants import StatusBits, TriggerBitsServo, TRIGOUT_GATE_OFF
 from .apt_messages import APTProtocol
 from .serial_comms import SerialSnooper
 
@@ -465,3 +465,42 @@ class ThorlabsServoDriver():
     def set_trigger_trigout_maxv(self, axis):
         '''Set trigger output high + pulse at max velocity (TRIGOUT_MAXV).'''
         self.set_trigger(axis, TriggerBitsServo.TRIGOUT_MAXV)
+
+    def set_trigger_gate_off(self, axis):
+        '''Drive the trigger output inactive, so no pulses reach the gate.'''
+        self.set_trigger(axis, TRIGOUT_GATE_OFF)
+
+    def arm_scan_gate(self, axis, armed, verify=True):
+        '''
+            arm_scan_gate(axis, armed): Arms or drops the max-velocity trigger
+            output the oscilloscope AND-gate uses.
+
+            Burst acquisition runs one scope acquisition across many rows, so
+            the gate must be armed only for the acquiring pass and dropped for
+            the flyback — otherwise the return move hits max velocity and
+            injects frames between rows.
+        '''
+        mode = TriggerBitsServo.TRIGOUT_MAXV if armed else TRIGOUT_GATE_OFF
+        if verify:
+            self.set_trigger_verified(axis, mode)
+        else:
+            self.set_trigger(axis, mode)
+
+    def set_trigger_verified(self, axis, mode, timeout=5.0, retries=2):
+        '''
+            set_trigger_verified(axis, mode): Sets the trigger mode and reads
+            it back to confirm it landed.
+
+            set_trigger is fire-and-forget over the shared TX queue.  Burst
+            acquisition toggles the gate between every row, and a dropped
+            change there silently fills the acquisition with flyback frames —
+            so confirm rather than assume.
+        '''
+        for _ in range(retries + 1):
+            self.set_trigger(axis, mode)
+            if int(self.get_trigger(axis, timeout=timeout)) == int(mode):
+                return
+        raise RuntimeError(
+            f"Axis 0x{axis:02X} did not accept trigger mode 0x{int(mode):02X} "
+            f"after {retries + 1} attempts"
+        )

hardware/tektronix_base.py

@@ -183,6 +183,10 @@ class TektronixOscilloscopeBase:
 
         self.write(f"HORizontal:FASTframe:COUNt {count}")
 
+    def get_fastframe_max_frames(self):
+        """Query how many FastFrame frames the current horizontal settings allow."""
+        return int(self.query("HORizontal:FASTframe:MAXFRames?"))
+
     def get_record_length(self):
         """Query the current horizontal record length."""
         response = self.query("HORizontal:MODe:RECOrdlength?")
@@ -361,6 +365,21 @@ class TektronixOscilloscopeBase:
 
         self.write(f"DATa:SOUrce {source}")
 
+    def set_data_encoding(self, encoding):
+        """Set the curve transfer encoding (e.g. RIBinary = signed int, MSB first)."""
+        valid = ('ASCII', 'RIBinary', 'RPBinary', 'FPBinary',
+                 'SRIbinary', 'SRPbinary', 'SFPbinary')
+        if encoding.upper() not in [v.upper() for v in valid]:
+            raise ValueError(f"Invalid data encoding: {encoding}. "
+                             f"Valid options: {', '.join(valid)}")
+        self.write(f"DATa:ENCdg {encoding}")
+
+    def set_data_width(self, width):
+        """Set bytes per sample for curve transfers."""
+        if width not in (1, 2):
+            raise ValueError(f"Invalid data width: {width}. Must be 1 or 2")
+        self.write(f"DATa:WIDth {width}")
+
     def query_wfmoutpre(self):
         """Query all waveform output preamble parameters."""
         return self.query("WFMOutpre?")
@@ -455,6 +474,43 @@ class TektronixOscilloscopeBase:
 
         return waveforms
 
+    def transfer_fastframe_bulk(self, frame_count, samples_per_frame,
+                                bytes_per_sample=1):
+        """Transfer a whole FastFrame burst as one contiguous buffer.
+
+        Unlike transfer_fastframe this does not care how the scope frames the
+        response — it accumulates blocks until it has the expected byte count,
+        so one large IEEE block and one block per frame both work.  Returns a
+        bytearray of frame_count * samples_per_frame * bytes_per_sample bytes.
+        """
+        if not self.get_fastframe_state():
+            raise RuntimeError("FastFrame is not enabled. Enable it with set_fastframe_state(True)")
+
+        expected = frame_count * samples_per_frame * bytes_per_sample
+        if expected <= 0:
+            raise RuntimeError(
+                f"Nothing to transfer: {frame_count} frames × "
+                f"{samples_per_frame} samples × {bytes_per_sample} bytes"
+            )
+
+        self.write("CURVe?")
+
+        buf = bytearray()
+        while len(buf) < expected:
+            block = self.read_raw(expected_bytes=expected - len(buf))
+            if not block:
+                raise RuntimeError(
+                    f"Scope returned an empty block {len(buf)}/{expected} bytes "
+                    "into the burst transfer"
+                )
+            buf += block
+
+        if len(buf) != expected:
+            raise RuntimeError(
+                f"Burst transfer overran: got {len(buf)} bytes, expected {expected}"
+            )
+        return buf
+
     def parse_curve_data(self, curve_bytes, byte_count=1, signed=True, byte_order='MSB'):
         """Parse raw curve data into integer array."""
         if byte_count not in (1, 2):
@@ -526,7 +582,19 @@ class TektronixOscilloscopeBase:
 
         return response.decode('ascii').strip()
 
-    def read_raw(self):
+    def _recv_exact(self, count):
+        """Read exactly `count` bytes; recv() is free to return fewer."""
+        chunks = []
+        remaining = count
+        while remaining > 0:
+            chunk = self.socket.recv(min(remaining, 65536))
+            if not chunk:
+                raise RuntimeError("Connection closed while reading data")
+            chunks.append(chunk)
+            remaining -= len(chunk)
+        return b''.join(chunks)
+
+    def read_raw(self, expected_bytes=None):
         """
         Read raw binary data from the instrument.
 
@@ -535,6 +603,10 @@ class TektronixOscilloscopeBase:
         where N is a digit indicating how many digits follow,
         and those digits specify the length of the data block.
 
+        `#0` announces an indeterminate-length block, normally delimited by EOI
+        — which a raw socket never sees.  Pass expected_bytes to say how many
+        bytes to take in that case.
+
         Returns:
             bytes: Raw binary data (without IEEE 488.2 header)
 
@@ -564,27 +636,24 @@ class TektronixOscilloscopeBase:
 
         num_digits = int(length_of_length)
 
-        # Read the data length
-        length_bytes = self.socket.recv(num_digits)
-        if len(length_bytes) != num_digits:
-            raise RuntimeError("Failed to read data length")
+        if num_digits == 0:
+            # Indeterminate length: no byte count follows, and the EOI that
+            # would delimit it does not exist on a raw socket.
+            if expected_bytes is None:
+                raise RuntimeError(
+                    "Scope returned an indeterminate-length block (#0); "
+                    "read_raw needs expected_bytes to size it over a socket"
+                )
+            data_length = expected_bytes
+        else:
+            data_length = int(self._recv_exact(num_digits))
 
-        data_length = int(length_bytes)
+        data = self._recv_exact(data_length)
 
-        # Read the actual binary data
-        chunks = []
-        remaining = data_length
-        while remaining > 0:
-            chunk = self.socket.recv(min(remaining, 65536))
-            if not chunk:
-                raise RuntimeError("Connection closed while reading data")
-            chunks.append(chunk)
-            remaining -= len(chunk)
+        # Read the trailing newline / block separator
+        self._recv_exact(1)
 
-        # Read the trailing newline
-        self.socket.recv(1)
-
-        return b''.join(chunks)
+        return data
 
     @property
     def is_connected(self):

sc3-aui-main.ui

@@ -997,6 +997,36 @@
         </property>
        </widget>
       </item>
+      <item>
+       <widget class="QCheckBox" name="burst_mode_check">
+        <property name="toolTip">
+         <string>Acquire as many whole rows per FastFrame acquisition as the scope can hold, and transfer each burst in one CURVe? transaction. The stage trigger output is gated off for the flyback between rows.</string>
+        </property>
+        <property name="text">
+         <string>Burst acquisition (multi-row FastFrame)</string>
+        </property>
+       </widget>
+      </item>
+      <item>
+       <widget class="QCheckBox" name="strict_rows_check">
+        <property name="toolTip">
+         <string>Stop the scan if a row does not acquire the expected number of frames, instead of zero-padding a short row or truncating a long one. Use for data runs where a silently squared-up row would be worse than a failed scan.</string>
+        </property>
+        <property name="text">
+         <string>Strict row packing (abort on frame-count mismatch)</string>
+        </property>
+       </widget>
+      </item>
+      <item>
+       <widget class="QPushButton" name="inspect_angles_btn">
+        <property name="toolTip">
+         <string>Rotate through the angles of the scan currently entered, parking at a random point in each so the SAW response can be checked on the oscilloscope before committing to the run.</string>
+        </property>
+        <property name="text">
+         <string>Inspect Angles…</string>
+        </property>
+       </widget>
+      </item>
       <item>
        <widget class="QPushButton" name="start_scan_btn">
         <property name="text">

sc3_aui_app.py

@@ -38,6 +38,7 @@ from core.scope_sras import SAMPLE_RATE_HZ, configure_channels
 from core.sras_format import SCAN_CHANNELS, SrasFile, plan_from_header
 from gui.qt_t3r import QtT3RAdapter
 from gui.qt_workers import PollingQueueWorker, QueueWorker
+from gui.inspect_bridge import QtAngleInspector
 from gui.scan_bridge import QtScanController
 from hardware.helios_laser import HeliosLaser
 from hardware.pybbd202 import AXIS_X, AXIS_Y, ThorlabsServoDriver
@@ -844,6 +845,104 @@ class ScanProgressWindow(QWidget):
 
 # ── Main window ───────────────────────────────────────────────────────────────
 
+class AngleInspectWindow(QWidget):
+    """Click through a plan's angles, parking the rig at a point in each.
+
+    Deliberately shows no waveform.  The operator reads the SAW response and
+    the bias levels off the oscilloscope itself — this window only says where
+    the rig is and lets them move it somewhere else.
+    """
+
+    goto_requested = pyqtSignal(int)
+    new_point_requested = pyqtSignal()
+    stop_requested = pyqtSignal()
+
+    def __init__(self, angle_labels: list[str], parent: QWidget | None = None):
+        super().__init__(parent, Qt.WindowType.Window)
+        self.setWindowTitle("Inspect Angles")
+        self.resize(420, 460)
+
+        layout = QVBoxLayout(self)
+        layout.addWidget(QLabel(
+            "Select an angle to rotate to it and park at a random point in "
+            "its scan area.\nRead the SAW response on the oscilloscope."
+        ))
+
+        self.angle_list = QListWidget(self)
+        for i, label in enumerate(angle_labels):
+            item = QListWidgetItem(label)
+            item.setData(Qt.ItemDataRole.UserRole, i)
+            self.angle_list.addItem(item)
+        self.angle_list.setCurrentRow(0)
+        # currentRowChanged would also fire when the code syncs the highlight
+        # back after a move, re-triggering the move it is reporting.
+        self.angle_list.itemClicked.connect(self._on_item_clicked)
+        layout.addWidget(self.angle_list)
+
+        nav_row = QHBoxLayout()
+        self.prev_btn = QPushButton("◀ Previous")
+        self.next_btn = QPushButton("Next ▶")
+        self.new_point_btn = QPushButton("New Point")
+        self.new_point_btn.setToolTip(
+            "Pick another random point at this angle without rotating — a bad "
+            "spot on the sample looks the same as a bad angle until you move."
+        )
+        self.prev_btn.clicked.connect(self._on_prev)
+        self.next_btn.clicked.connect(self._on_next)
+        self.new_point_btn.clicked.connect(self.new_point_requested.emit)
+        nav_row.addWidget(self.prev_btn)
+        nav_row.addWidget(self.next_btn)
+        nav_row.addWidget(self.new_point_btn)
+        layout.addLayout(nav_row)
+
+        self.point_label = QLabel("—")
+        self.point_label.setStyleSheet("font-weight: bold;")
+        layout.addWidget(self.point_label)
+
+        self.status_label = QLabel("Starting …")
+        self.status_label.setWordWrap(True)
+        layout.addWidget(self.status_label)
+
+        self.close_btn = QPushButton("Close")
+        self.close_btn.clicked.connect(self.close)
+        layout.addWidget(self.close_btn)
+
+        self._n_angles = len(angle_labels)
+        self._angle_idx = 0
+        self.set_busy(True)
+
+    # ── Worker → window ───────────────────────────────────────────────────────
+
+    def set_busy(self, busy: bool):
+        """Lock navigation while the stage is moving; the rig is not re-entrant."""
+        for w in (self.angle_list, self.prev_btn, self.next_btn,
+                  self.new_point_btn):
+            w.setEnabled(not busy)
+
+    def on_status(self, msg: str):
+        self.status_label.setText(msg)
+
+    def on_point(self, point):
+        self._angle_idx = point.angle_idx
+        self.angle_list.setCurrentRow(point.angle_idx)
+        self.point_label.setText(point.describe())
+
+    # ── Window → worker ───────────────────────────────────────────────────────
+
+    def _on_item_clicked(self, item):
+        self.goto_requested.emit(item.data(Qt.ItemDataRole.UserRole))
+
+    def _on_prev(self):
+        self.goto_requested.emit((self._angle_idx - 1) % self._n_angles)
+
+    def _on_next(self):
+        self.goto_requested.emit((self._angle_idx + 1) % self._n_angles)
+
+    def closeEvent(self, event):
+        self.stop_requested.emit()
+        super().closeEvent(event)
+
+
 class MainWindow(QMainWindow):
     def __init__(self):
         super().__init__()
@@ -878,6 +977,9 @@ class MainWindow(QMainWindow):
         self._scan_progress = ScanProgressWindow()
 
         self._scan_worker: QtScanController | None = None
+        self._inspect_thread: QThread | None = None
+        self._inspect_worker: QtAngleInspector | None = None
+        self._inspect_window: AngleInspectWindow | None = None
         self._scan_thread:  QThread | None = None
 
         self._bbd_active_jog: tuple[str, int] | None = None   # (axis, direction)
@@ -917,6 +1019,8 @@ class MainWindow(QMainWindow):
         self.row_spacing_edit.setText("0.250")
         self.scan_prefix_edit.setText("scan")
         self.scan_save_dir_edit.setText(DEFAULTS.save_dir)
+        self.burst_mode_check.setChecked(DEFAULTS.burst_mode)
+        self.strict_rows_check.setChecked(DEFAULTS.strict_rows)
 
         # Hide the old T3R manual controls; the connect toggle becomes the panel button.
         for w in (
@@ -977,6 +1081,8 @@ class MainWindow(QMainWindow):
         self.t3r_comport_edit.editingFinished.connect(self._persist_defaults)
         self.bbd202_comport_edit.editingFinished.connect(self._persist_defaults)
         self.oscope_ip_edit.editingFinished.connect(self._persist_defaults)
+        self.burst_mode_check.toggled.connect(self._persist_defaults)
+        self.strict_rows_check.toggled.connect(self._persist_defaults)
 
         # Camera
         self.show_camera_toggle.toggled.connect(self._on_camera_toggle)
@@ -995,6 +1101,7 @@ class MainWindow(QMainWindow):
 
         # Scan
         self.start_scan_btn.clicked.connect(self._on_start_scan)
+        self.inspect_angles_btn.clicked.connect(self._on_inspect_angles)
         self.save_dir_browse_btn.clicked.connect(self._on_browse_save_dir)
         self._scan_progress.abort_requested.connect(self._on_abort_scan)
         self._scan_progress.pause_toggled.connect(self._on_pause_scan)
@@ -1134,6 +1241,8 @@ class MainWindow(QMainWindow):
         DEFAULTS.bbd_port = self.bbd202_comport_edit.text().strip()
         DEFAULTS.oscope_ip = self.oscope_ip_edit.text().strip()
         DEFAULTS.save_dir = self.scan_save_dir_edit.text().strip()
+        DEFAULTS.burst_mode = self.burst_mode_check.isChecked()
+        DEFAULTS.strict_rows = self.strict_rows_check.isChecked()
         DEFAULTS.save()
 
     # ── Camera toggle ─────────────────────────────────────────────────────────
@@ -1166,6 +1275,79 @@ class MainWindow(QMainWindow):
             self.scan_save_dir_edit.setText(d)
             self._persist_defaults()
 
+    def _on_inspect_angles(self):
+        """Open the pre-scan angle inspector for the plan currently entered."""
+        if self._scan_thread is not None and self._scan_thread.isRunning():
+            QMessageBox.warning(
+                self, "Scan In Progress",
+                "A scan is running — abort it before inspecting angles."
+            )
+            return
+        if self._inspect_thread is not None and self._inspect_thread.isRunning():
+            self._inspect_window.raise_()
+            self._inspect_window.activateWindow()
+            return
+
+        try:
+            plan, _, _ = self._build_scan_plan()
+        except ValueError as e:
+            QMessageBox.warning(self, "Invalid Scan Parameters", str(e))
+            return
+
+        rotator = RotationAxis(self._t3r_driver.driver, DEFAULT_ROTATION)
+        self._inspect_thread = QThread(self)
+        self._inspect_worker = QtAngleInspector(
+            stage=self._bbd_worker.controller,
+            scope=self._oscope_worker.scope,
+            rotator=rotator,
+            plan=plan,
+            # Same reason the scan does it: the inspector drives the stage from
+            # its own thread, and the position poll shares the BBD TX queue.
+            on_inspect_active=lambda active: setattr(
+                self._bbd_worker, "scanning_active", active),
+        )
+        self._inspect_worker.moveToThread(self._inspect_thread)
+
+        window = AngleInspectWindow(self._inspect_worker.angle_labels(), self)
+        self._inspect_window = window
+
+        window.goto_requested.connect(self._inspect_worker.request_goto)
+        window.new_point_requested.connect(self._inspect_worker.request_new_point)
+        window.stop_requested.connect(self._on_inspect_window_closed)
+
+        self._inspect_worker.status_msg.connect(window.on_status)
+        self._inspect_worker.point_changed.connect(window.on_point)
+        self._inspect_worker.busy_changed.connect(window.set_busy)
+        self._inspect_worker.start_failed.connect(self._on_inspect_failed)
+        self._inspect_worker.error_occurred.connect(
+            lambda m: QMessageBox.warning(self, "Inspection Error", m))
+
+        self._inspect_thread.started.connect(self._inspect_worker.run)
+        self.start_scan_btn.setEnabled(False)
+        self.inspect_angles_btn.setEnabled(False)
+        window.show()
+        self._inspect_thread.start()
+        self._inspect_worker.request_start()
+
+    def _on_inspect_failed(self, message: str):
+        QMessageBox.warning(self, "Cannot Inspect Angles", message)
+        if self._inspect_window is not None:
+            self._inspect_window.close()
+
+    def _on_inspect_window_closed(self):
+        """Tear the worker down and hand the hardware back to the scan panel."""
+        if self._inspect_worker is not None:
+            self._inspect_worker.request_stop()
+            self._inspect_worker.stop_worker()
+        if self._inspect_thread is not None:
+            self._inspect_thread.quit()
+            self._inspect_thread.wait(10000)
+        self._inspect_thread = None
+        self._inspect_worker = None
+        self._inspect_window = None
+        self.start_scan_btn.setEnabled(True)
+        self.inspect_angles_btn.setEnabled(True)
+
     def _on_start_scan(self):
         try:
             plan, prefix, save_dir = self._build_scan_plan()
@@ -1257,6 +1439,8 @@ class MainWindow(QMainWindow):
             # a worker concern, not the engine's.
             on_scan_active=lambda active: setattr(
                 self._bbd_worker, "scanning_active", active),
+            burst_mode=self.burst_mode_check.isChecked(),
+            strict_rows=self.strict_rows_check.isChecked(),
         )
         self._scan_worker.moveToThread(self._scan_thread)
         self._scan_thread.started.connect(self._scan_worker.run)

scan_format.md

@@ -183,19 +183,63 @@ using that angle's `x_start` from the Per-Angle Geometry Table (not
 
 ---
 
-## Acquisition Settings (fixed by sc3_aui_app.py)
+## Acquisition Settings (fixed by core/scope_sras.py)
 
 | Parameter            | Value                                    |
-|-----------------------|------------------------------|
-| Oscilloscope trigger | CH2, rising edge, 1.24 V     |
-| Trigger offset       | 0 % (trigger at left edge)   |
+|-----------------------|------------------------------------------|
+| Setup trigger        | CH2, rising edge, 0.500 V (`TRIG_LEVEL_V`) |
+| Scan trigger         | Logic AND, CH2 HIGH ∧ CH3 HIGH, 0.500 V  |
+| Horizontal position  | 30 (`HORizontal:POSition`)               |
 | Sample rate          | 6.25 GS/s  (160 ps/sample)               |
+| Transfer format      | `DATa:ENCdg RIBinary`, `DATa:WIDth 1`    |
 | Channels recorded    | CH1, CH3, CH4                            |
 | Stage X velocity     | 100 mm/s                                 |
 | Stage X acceleration | 1500 mm/s²                               |
-| Stage X trigger out  | Logic-high at max velocity   |
+| Stage X trigger out  | Logic-high at max velocity (`TRIGOUT_MAXV`) |
 | Acquisition mode     | FastFrame, Normal trigger                |
 
+None of these are stored in the file, so they do not affect byte layout — but
+they do set where the acoustic packet lands inside each frame.  Read them from
+`core/scope_sras.py`; earlier revisions of this table drifted from the code.
+
+---
+
+## Acquisition Paths
+
+Two acquisition strategies write **byte-identical** files; the choice is a
+runtime flag (`ScanEngine(burst_mode=…)`, exposed as a checkbox in the app) and
+is not recorded in the file.
+
+| | Per-row (default) | Burst |
+|---|---|---|
+| FastFrame acquisitions | one per row | one per `floor(max_frames / n_frames)` rows |
+| Curve transfers | one per channel per row | one per channel per burst |
+| Stage X trigger out | armed for the whole scan | armed per acquiring pass, dropped for the flyback |
+
+Burst mode runs a single acquisition across several rows, so the return move
+must not trigger: the trigger output is dropped before each flyback and
+re-armed for each acquiring pass.  Row boundaries inside the burst come from
+`ACQuire:NUMFRAMESACQuired?` sampled after each pass — the burst itself carries
+no row markers.  See `core/scope_burst.py`.
+
+### Row packing
+
+The format has no per-row length field, so a row that over- or under-triggers
+cannot be written as it arrived — that would shift every later row.  Two
+policies are selectable (`ScanEngine(strict_rows=…)`, a checkbox in the app),
+and the choice is not recorded in the file:
+
+| | Pad (default) | Strict |
+|---|---|---|
+| Short row | zero-padded to `n_frames`, warned | scan stops |
+| Long row | trailing frames dropped, warned | scan stops |
+
+Pad keeps a scan running through an occasional mis-trigger, at the cost that
+the affected row is indistinguishable from a good one afterwards — nothing in
+the file records that it was padded.  Strict is for data runs where that
+ambiguity is worse than a failed scan: it aborts before writing the row, so
+the file always ends on a whole-row boundary.
+
 ---
 
 ## Version History

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_angle_inspect.py

@@ -0,0 +1,292 @@
+"""Pre-scan angle inspection, driven entirely by fake hardware.
+
+The feature's defining constraint is that it reads nothing back from the
+scope — the operator looks at the instrument. These tests pin that, the scope
+state the app is responsible for putting the instrument into, and the motion
+sequence across angles.
+"""
+import random
+
+import pytest
+
+from core.angle_inspect import AngleInspector, InspectCallbacks
+from core.rotation import RotationAxis, RotationSettings
+from core.scan_engine import AXIS_X, AXIS_Y
+from core.scan_geometry import build_plan
+from core.scope_inspect import (
+    BIAS_CHANNELS, BIAS_POSITION_DIV, BIAS_SCALE_V_DIV, BIAS_WINDOW_V,
+    INSPECT_TRIG_LEVEL_V, inspect_channel_profiles,
+)
+from core.scope_sras import SRAS_CHANNELS
+from fakes import FakeScope, FakeStage, FakeT3R, Trace
+
+SPF = 8
+
+
+def make_plan(num_angles=3):
+    return build_plan(40.0, 30.0, 2.0, 1.0, num_angles, 0.25,
+                      laser_freq_hz=20000.0, velocity_mm_s=100.0)
+
+
+def build(num_angles=3, seed=1234, callbacks=None, rotator_open=True):
+    trace = Trace()
+    scope = FakeScope(trace, samples_per_frame=SPF)
+    stage = FakeStage(trace, scope=scope)
+    t3r = FakeT3R(trace, is_open=rotator_open)
+    rotator = RotationAxis(t3r, RotationSettings())
+    plan = make_plan(num_angles)
+    insp = AngleInspector(stage, scope, rotator, plan,
+                          callbacks=callbacks or InspectCallbacks(),
+                          rng=random.Random(seed))
+    return insp, trace, plan
+
+
+def writes(trace):
+    return [c[1] for c in trace.of("write")]
+
+
+# ── The defining constraint ──────────────────────────────────────────────────
+
+def test_inspection_never_reads_a_waveform_back():
+    """The operator reads the scope; the app must not pull data off it.
+
+    If this fails, someone has added a transfer path to a feature whose whole
+    premise is that there isn't one.
+    """
+    insp, trace, plan = build()
+    insp.start()
+    for i in range(plan.n_angles):
+        insp.goto_angle(i)
+        insp.new_point()
+    insp.stop()
+
+    forbidden = {"transfer_fastframe", "transfer_fastframe_bulk",
+                 "transfer_curve", "set_data_source", "query_wfmoutpre"}
+    assert forbidden.isdisjoint(set(trace.names()))
+    assert "CURVe?" not in writes(trace)
+
+
+# ── Scope configuration ──────────────────────────────────────────────────────
+
+def test_start_sets_an_edge_trigger_on_ch2_above_the_scan_level():
+    insp, trace, _ = build()
+    insp.start()
+
+    assert "TRIGger:A:TYPe EDGE" in writes(trace)
+    assert trace.of("set_trigger_source")[-1][1] == 2
+    assert trace.of("set_trigger_slope")[-1][1] == "RISE"
+    ch, level = trace.of("set_trigger_level")[-1][1:3]
+    assert (ch, level) == (2, INSPECT_TRIG_LEVEL_V)
+    assert INSPECT_TRIG_LEVEL_V >= 2.0
+
+
+def test_start_disables_fastframe_averaging_and_the_logic_trigger():
+    """Everything the scan needs and inspection must not inherit."""
+    insp, trace, _ = build()
+    insp.start()
+
+    assert trace.of("set_fastframe_state")[-1][1] is False
+    assert trace.of("set_acquire_mode")[-1][1] == "SAMPLE"
+    w = writes(trace)
+    assert not any("LOGIc" in cmd or "LOGICPattern" in cmd for cmd in w)
+
+
+def test_start_leaves_the_acquisition_free_running():
+    """The display has to keep updating while the operator looks at it."""
+    insp, trace, _ = build()
+    insp.start()
+
+    w = writes(trace)
+    assert "ACQuire:STOPAfter RUNSTop" in w
+    assert w.index("ACQuire:STOPAfter RUNSTop") < w.index("ACQuire:STATE RUN")
+    assert "ACQuire:STATE STOP" not in w
+
+
+def test_bias_channels_are_directly_comparable():
+    """CH3/CH4 must share scale and position or the eye comparison is a lie."""
+    profiles = inspect_channel_profiles()
+    a, b = (profiles[ch] for ch in BIAS_CHANNELS)
+    assert a.scale_v_div == b.scale_v_div
+    assert a.position_div == b.position_div
+    # Same front end as the scan records — only the display changes.
+    for ch in BIAS_CHANNELS:
+        assert profiles[ch].termination_ohm == SRAS_CHANNELS[ch].termination_ohm
+        assert profiles[ch].coupling == SRAS_CHANNELS[ch].coupling
+        assert profiles[ch].bandwidth_hz == SRAS_CHANNELS[ch].bandwidth_hz
+
+
+@pytest.mark.parametrize("n_divisions", [8, 10])
+def test_bias_window_shows_zero_to_700mv_with_headroom(n_divisions):
+    """0–700 mV must fit on screen, above ground, on either graticule size.
+
+    Ground sits BIAS_POSITION_DIV divisions below centre, so the visible
+    window runs from (-N/2 - pos)*scale to (+N/2 - pos)*scale.
+    """
+    half = n_divisions / 2
+    bottom = (-half - BIAS_POSITION_DIV) * BIAS_SCALE_V_DIV
+    top = (half - BIAS_POSITION_DIV) * BIAS_SCALE_V_DIV
+
+    assert bottom < 0.0, "no room below ground for undershoot"
+    assert top > BIAS_WINDOW_V, "700 mV is clipped or sitting on the top edge"
+    # The point of moving the trace down: most of the screen is above ground.
+    assert abs(bottom) < top
+
+
+def test_ch1_keeps_the_acquisition_front_end():
+    """What you see at a point is what a scan would record there."""
+    assert inspect_channel_profiles()[1] == SRAS_CHANNELS[1]
+
+
+# ── Stage and rotation ───────────────────────────────────────────────────────
+
+def test_start_parks_on_the_first_angle():
+    insp, _, plan = build()
+    point = insp.start()
+
+    assert point.angle_idx == 0
+    assert point.angle_deg == plan.per_angle[0].angle_deg
+    assert insp.current_point == point
+
+
+def test_the_gate_is_off_for_the_whole_inspection():
+    """Nothing here is gated, and an armed output keeps driving the line."""
+    insp, trace, _ = build()
+    insp.start()
+    insp.goto_angle(2)
+    insp.new_point()
+
+    assert trace.count("set_trigger_gate_off") >= 1
+    assert trace.count("set_trigger_trigout_maxv") == 0
+    assert [c[2] for c in trace.of("arm_scan_gate") if c[2]] == []
+
+
+def test_points_land_on_the_scan_grid():
+    """A point the scan would never sample tells you nothing about the scan."""
+    insp, _, plan = build()
+    insp.start()
+
+    for i in range(plan.n_angles):
+        pa = plan.per_angle[i]
+        for _ in range(5):
+            pt = insp.new_point() if insp.angle_idx == i else insp.goto_angle(i)
+            assert pt.angle_idx == i
+            assert pt.y_mm in pa.y_positions
+            assert pa.x_start <= pt.x_mm <= pa.x_start + pa.x_delta
+
+
+def test_goto_angle_rotates_then_moves():
+    insp, trace, plan = build()
+    insp.start()
+    trace.calls.clear()
+
+    insp.goto_angle(2)
+
+    # t3r_rotate carries the delta, so assert the resulting absolute angle.
+    assert trace.count("t3r_rotate") == 1, "expected exactly one rotation"
+    assert insp._rotator.current_deg == pytest.approx(plan.per_angle[2].angle_deg)
+    moves = trace.of("move_axis_absolute")
+    assert [m[1] for m in moves] == [AXIS_Y, AXIS_X], "Y then X, as the scan does"
+
+
+def test_new_point_re_rolls_without_rotating():
+    """Distinguishing a bad spot from a bad angle depends on not rotating."""
+    insp, trace, _ = build()
+    insp.start()
+    insp.goto_angle(1)
+    trace.calls.clear()
+
+    first = insp.current_point
+    second = insp.new_point()
+
+    assert second.angle_idx == first.angle_idx == 1
+    assert (second.x_mm, second.y_mm) != (first.x_mm, first.y_mm)
+    assert trace.count("t3r_rotate") == 0, "new_point must not rotate"
+    assert [m[1] for m in trace.of("move_axis_absolute")] == [AXIS_Y, AXIS_X]
+
+
+def test_next_and_prev_wrap_around():
+    insp, _, plan = build(num_angles=3)
+    insp.start()
+
+    assert insp.next_angle().angle_idx == 1
+    assert insp.next_angle().angle_idx == 2
+    assert insp.next_angle().angle_idx == 0, "should wrap forward"
+    assert insp.prev_angle().angle_idx == plan.n_angles - 1, "should wrap back"
+
+
+def test_angle_labels_cover_every_angle():
+    insp, _, plan = build(num_angles=9)
+    labels = insp.angle_labels()
+    assert len(labels) == 9
+    assert labels[0].startswith("Angle 1/9")
+
+
+# ── Guards ───────────────────────────────────────────────────────────────────
+
+def test_multi_angle_inspection_requires_the_rotator():
+    insp, _, _ = build(num_angles=3, rotator_open=False)
+    with pytest.raises(RuntimeError, match="T3R rotation stage"):
+        insp.start()
+
+
+def test_single_angle_inspection_works_without_the_rotator():
+    insp, _, _ = build(num_angles=1, rotator_open=False)
+    point = insp.start()
+    assert point.angle_idx == 0
+
+
+def test_navigation_before_start_is_rejected():
+    insp, _, _ = build()
+    with pytest.raises(RuntimeError, match="not been started"):
+        insp.goto_angle(1)
+    with pytest.raises(RuntimeError, match="not been started"):
+        insp.new_point()
+
+
+def test_out_of_range_angle_is_rejected():
+    insp, _, _ = build(num_angles=3)
+    insp.start()
+    with pytest.raises(IndexError):
+        insp.goto_angle(3)
+
+
+def test_stop_halts_the_sweep_and_sends_the_rotator_home():
+    insp, trace, _ = build()
+    insp.start()
+    insp.goto_angle(2)
+    trace.calls.clear()
+
+    insp.stop()
+
+    assert "ACQuire:STATE STOP" in writes(trace)
+    assert trace.count("t3r_rotate") == 1, "GR not sent home"
+    assert insp._rotator.current_deg == pytest.approx(0.0)
+
+
+def test_stop_is_idempotent():
+    insp, trace, _ = build()
+    insp.start()
+    insp.stop()
+    trace.calls.clear()
+
+    insp.stop()      # must not re-issue anything or raise
+
+    assert trace.calls == []
+
+
+def test_busy_callback_brackets_every_move():
+    """The window disables its controls on this, so it has to pair up."""
+    events = []
+    insp, _, _ = build(callbacks=InspectCallbacks(on_busy=events.append))
+    insp.start()
+    insp.goto_angle(1)
+    insp.new_point()
+    insp.stop()
+
+    assert events, "no busy events emitted"
+    assert events[0] is True and events[-1] is False
+    depth = 0
+    for e in events:
+        depth += 1 if e else -1
+        assert depth in (0, 1), f"unbalanced busy events: {events}"
+    assert depth == 0

tests/test_scan_engine.py

@@ -20,22 +20,25 @@ 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",
+          strict_rows=False, **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, strict_rows=strict_rows)
     return engine, trace, plan
 
 
@@ -192,8 +195,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 +268,203 @@ 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()
+
+
+def _clip_one_row(engine, which_pass=2, lost=1):
+    """Make one acquiring pass come up `lost` frames short.
+
+    `which_pass` counts acquiring passes from 1, so the default clips the
+    second data row (row 2) — far enough in that a mishandled short row shows
+    up as a shift in the rows after it.
+    """
+    scope = engine._scope
+    real_acquire = scope.acquire_frames
+    passes = {"n": 0}
+
+    def clipped(n):
+        if scope._running:
+            passes["n"] += 1
+            if passes["n"] == which_pass:
+                n -= lost
+        real_acquire(n)
+    scope.acquire_frames = clipped
+
+
+@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
+
+    _clip_one_row(engine)
+
+    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)
+
+
+# ── Strict row packing ───────────────────────────────────────────────────────
+
+@pytest.mark.parametrize("burst_mode", [False, True])
+def test_strict_row_packing_aborts_on_a_short_row(tmp_path, burst_mode):
+    """Strict mode fails the scan instead of silently squaring a row up.
+
+    The default padding keeps the file readable but makes a mis-triggered row
+    indistinguishable from a good one after the fact, since v6 records no
+    per-row frame count. Strict mode trades the salvaged rows for knowing.
+    """
+    plan = make_plan(**BURST_PLAN)
+    engine, _, _ = build(tmp_path, plan=plan, burst_mode=burst_mode,
+                         max_frames=BURST_MAX_FRAMES, strict_rows=True)
+    engine._preflight_done = True
+    _clip_one_row(engine)
+
+    with pytest.raises(RuntimeError, match="Row 2: 3 frames acquired, 4 expected"):
+        engine.run()
+
+
+@pytest.mark.parametrize("burst_mode", [False, True])
+def test_strict_row_packing_does_not_disturb_a_clean_scan(tmp_path, burst_mode):
+    """Strict mode is inert when every row acquires what it declared."""
+    plan = make_plan(**BURST_PLAN)
+    engine, _, _ = build(tmp_path, plan=plan, burst_mode=burst_mode,
+                         max_frames=BURST_MAX_FRAMES, strict_rows=True)
+    engine._preflight_done = True
+
+    result = engine.run()
+
+    assert result.rows_written == plan.total_rows
+    sras = SrasFile(tmp_path / "out.sras")
+    assert [s.status for s in sras.angle_status()] == ["OK"]
+
+
+def test_strict_row_packing_writes_nothing_for_the_failed_row(tmp_path):
+    """The abort must not leave a half-written row behind.
+
+    CH1 leads SCAN_CHANNELS, so the frame count is known before any of the
+    row's channels are written — the file should end on a whole-row boundary.
+    """
+    plan = make_plan(**BURST_PLAN)
+    engine, _, _ = build(tmp_path, plan=plan, strict_rows=True)
+    engine._preflight_done = True
+    _clip_one_row(engine)
+
+    with pytest.raises(RuntimeError, match="Strict row packing"):
+        engine.run()
+
+    # Row 1 was written in full; row 2 aborted before writing anything, so
+    # the file ends exactly on a row boundary.
+    sras = SrasFile(tmp_path / "out.sras")
+    written = (tmp_path / "out.sras").stat().st_size - sras.data_start_offset
+    assert written == sras.row_bytes(0)
+
+
 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]