diff --git a/core/scan_engine.py b/core/scan_engine.py index 175438e..a268dfc 100644 --- a/core/scan_engine.py +++ b/core/scan_engine.py @@ -15,7 +15,7 @@ from dataclasses import dataclass, field from pathlib import Path from typing import Callable -from core import scope_sras +from core import scope_burst, scope_sras from core.rotation import RotationAxis from core.scan_geometry import ScanPlan, validate_plan from core.sras_format import SCAN_CHANNELS, create_scan_file @@ -95,7 +95,8 @@ class ScanEngine: def __init__(self, stage, scope, rotator: RotationAxis | None, plan: ScanPlan, out_path: Path, resume: ResumeState | None = None, - callbacks: ScanCallbacks | None = None): + callbacks: ScanCallbacks | None = None, + burst_mode: bool = False): self._stage = stage self._scope = scope self._rotator = rotator @@ -103,6 +104,11 @@ class ScanEngine: self._out_path = Path(out_path) self._resume = resume self._cb = callbacks if callbacks is not None else ScanCallbacks() + # Burst mode acquires as many whole rows per FastFrame acquisition as + # the scope's frame memory holds, instead of one row per acquisition. + self._burst_mode = burst_mode + self._max_frames = 0 + self._preflight_done = False self._abort = threading.Event() self._resume_event = threading.Event() @@ -207,6 +213,14 @@ class ScanEngine: self._scan_loop(scan_file, samples_per_frame, result) finally: scan_file.close() + # Leave the X trigger output inactive. Burst mode toggles it every + # row and could exit from either state; the per-row path used to + # leave TRIGOUT_MAXV armed for the rest of the session, which keeps + # driving the gate line on every later jog. + try: + self._stage.set_trigger_gate_off(AXIS_X) + except Exception: + logger.exception("Could not return the X trigger output to idle") # Return the GR axis home regardless of abort or error if rotator_ready and abs(self._rotator.current_deg) > 0.001: self._cb.on_status("Returning GR to home …") @@ -242,8 +256,14 @@ class ScanEngine: acceleration=SCAN_ACCEL_MM_S2) ctrl.set_velocity_params(AXIS_Y, max_velocity=SCAN_VELOCITY_MM_S, acceleration=SCAN_ACCEL_MM_S2) - # X trigger: logic-high output while the stage is at maximum velocity - ctrl.set_trigger_trigout_maxv(AXIS_X) + # X trigger: logic-high output while the stage is at maximum velocity. + # Burst mode arms it per acquiring pass instead — a burst spans several + # rows with the scope running throughout, so leaving it armed would let + # the flyback trigger frames between rows. + if self._burst_mode: + ctrl.arm_scan_gate(AXIS_X, False) + else: + ctrl.set_trigger_trigout_maxv(AXIS_X) def _prepare_scope(self) -> int: self._cb.on_status("Configuring oscilloscope …") @@ -290,6 +310,16 @@ class ScanEngine: ) scope_sras.configure_scan_trigger(self._scope) + + if self._burst_mode: + # Horizontal settings are fixed by now, so the capacity is stable + # for the whole scan; only rows-per-burst varies (n_frames is + # per-angle). + self._max_frames = scope_burst.max_frames(self._scope) + self._cb.on_status( + f"Burst mode: scope holds {self._max_frames} frames " + f"({samples_per_frame} samples/frame)" + ) return samples_per_frame def _open_output(self, samples_per_frame: int, result: ScanResult): @@ -309,7 +339,6 @@ class ScanEngine: plan = self._plan n_angles = plan.n_angles x_ramp_total = SCAN_RAMP_MM + SCAN_RAMP_BUFFER_MM - ctrl = self._stage scope = self._scope targets_by_ai = None @@ -337,57 +366,253 @@ class ScanEngine: f"Rotating GR to {pa.angle_deg:.1f}° (Δ{delta:+.1f}°) …") self._rotator.rotate_to(pa.angle_deg) - # Each angle's bounding box gives it its own points/row count, so - # the scope's FastFrame count must be re-armed per angle. - scope.set_fastframe_count(pa.n_frames) - - for ri, y_pos in enumerate(pa.y_positions): - self._pause_point() - - self._cb.on_row_started(ri + 1, pa.n_rows, ai + 1, n_angles) - self._cb.on_status( - f"Angle {ai+1}/{n_angles} Row {ri+1}/{pa.n_rows} " - f"(Y={y_pos:.3f} mm)" - ) - - # Position the stage one ramp-length + buffer before the data - # window so it is at full velocity before x_start. - ctrl.move_axis_absolute(AXIS_Y, y_pos, timeout=60.0) - ctrl.move_axis_absolute(AXIS_X, pa.x_start - x_ramp_total, timeout=30.0) - - scope_sras.arm_row(scope) - - # Data window + ramp + buffer run-off, so the stage does not - # begin decelerating before the last point. - x_end = pa.x_start + pa.x_delta + x_ramp_total - ctrl.move_axis_absolute(AXIS_X, x_end, timeout=120.0) - - scope_sras.finish_row(scope) - self._write_row(scan_file, samples_per_frame, ri) - - result.rows_written += 1 - self._cb.on_row_done(ri + 1, pa.n_rows, ai + 1, n_angles) + if self._burst_mode: + # Burst mode sizes the FastFrame count from the scope's whole + # capacity instead (see scope_burst.start_burst), so there is + # nothing to re-arm per angle here. + self._scan_rows_burst(scan_file, pa, ai, n_angles, + samples_per_frame, result, x_ramp_total) + else: + # Each angle's bounding box gives it its own points/row count, + # so the scope's FastFrame count must be re-armed per angle. + scope.set_fastframe_count(pa.n_frames) + self._scan_rows_serial(scan_file, pa, ai, n_angles, + samples_per_frame, result, x_ramp_total) result.angles_acquired.append(ai) - def _write_row(self, scan_file, samples_per_frame: int, row_idx: int): + # ── Per-row acquisition (one FastFrame acquisition per row) ─────────────── + + def _scan_rows_serial(self, scan_file, pa, ai: int, n_angles: int, + samples_per_frame: int, result: ScanResult, + x_ramp_total: float): + ctrl = self._stage + scope = self._scope + + for ri, y_pos in enumerate(pa.y_positions): + self._pause_point() + + self._cb.on_row_started(ri + 1, pa.n_rows, ai + 1, n_angles) + self._cb.on_status( + f"Angle {ai+1}/{n_angles} Row {ri+1}/{pa.n_rows} " + f"(Y={y_pos:.3f} mm)" + ) + + # Position the stage one ramp-length + buffer before the data + # window so it is at full velocity before x_start. + ctrl.move_axis_absolute(AXIS_Y, y_pos, timeout=60.0) + ctrl.move_axis_absolute(AXIS_X, pa.x_start - x_ramp_total, timeout=30.0) + + scope_sras.arm_row(scope) + + # Data window + ramp + buffer run-off, so the stage does not + # begin decelerating before the last point. + x_end = pa.x_start + pa.x_delta + x_ramp_total + ctrl.move_axis_absolute(AXIS_X, x_end, timeout=120.0) + + scope_sras.finish_row(scope) + self._write_row(scan_file, samples_per_frame, ri, pa.n_frames) + + result.rows_written += 1 + self._cb.on_row_done(ri + 1, pa.n_rows, ai + 1, n_angles) + + def _write_row(self, scan_file, samples_per_frame: int, row_idx: int, + n_frames: int): """Stream every channel from the scope into the file. CH3 is the max-vel gate signal — no useful waveform data — so zeroed frames are written to keep the file layout intact. """ scope = self._scope + ch_bytes = n_frames * samples_per_frame for ch in SCAN_CHANNELS: if ch == 3: self._cb.on_status("Writing zeroed CH3 frames …") - zero_frame = bytes(samples_per_frame) - for _ in range(scope_sras.frames_acquired(scope)): - scan_file.write(zero_frame) + scan_file.write(bytes(ch_bytes)) continue self._cb.on_status(f"Fetching CH{ch} data …") waveforms = scope_sras.transfer_channel(scope, ch) - if ch == 4 and waveforms: - self._cb.on_dc_bias(row_idx + 1, scope_sras.frame_means(waveforms)) - for w in waveforms: - scan_file.write(w) + if ch == SCAN_CHANNELS[0]: + self._warn_frame_delta(row_idx, len(waveforms), n_frames) + row = scope_burst.normalize_row( + b"".join(waveforms), 0, len(waveforms), n_frames, samples_per_frame) + if ch == 4: + self._cb.on_dc_bias(row_idx + 1, scope_burst.frame_means_block( + row, 0, n_frames, samples_per_frame)) + scan_file.write(row) + + # ── Burst acquisition (many whole rows per FastFrame acquisition) ───────── + + def _scan_rows_burst(self, scan_file, pa, ai: int, n_angles: int, + samples_per_frame: int, result: ScanResult, + x_ramp_total: float): + """Acquire the angle in bursts of as many whole rows as the scope holds. + + One ACQuire:STATE RUN spans the whole burst, so the gate is armed only + for each acquiring pass and dropped for the flyback — otherwise the + return move would reach max velocity and inject frames between rows. + """ + scope = self._scope + n_frames = pa.n_frames + x_lead_in = pa.x_start - x_ramp_total + x_end = pa.x_start + pa.x_delta + x_ramp_total + + if not self._preflight_done: + # Once per scan: the gate wiring can't change between angles, and + # the check costs two row-times. + self._gate_off_preflight(x_lead_in, x_end) + self._preflight_done = True + + row = 0 + while row < pa.n_rows: + self._pause_point() + n_burst = scope_burst.rows_per_burst( + self._max_frames, n_frames, samples_per_frame, pa.n_rows - row) + self._cb.on_status( + f"Angle {ai+1}/{n_angles} Rows {row+1}-{row+n_burst}/{pa.n_rows} " + f"in one acquisition ({n_burst * n_frames} frames) …" + ) + + burst_start = scan_file.tell() + cumulative = [] + baseline = scope_burst.start_burst(scope, self._max_frames) + try: + for r in range(n_burst): + self._check_abort() + self._cb.on_row_started(row + r + 1, pa.n_rows, + ai + 1, n_angles) + self._acquire_gated_row(pa.y_positions[row + r], + x_lead_in, x_end) + total = scope_burst.frames_acquired(scope) + if total >= self._max_frames: + raise RuntimeError( + f"FastFrame buffer full ({total}/{self._max_frames} " + f"frames) at row {row + r + 1} — later rows in this " + "burst would be misattributed. Raise " + "scope_burst.BURST_FRAME_HEADROOM and rerun." + ) + cumulative.append(total - baseline) + finally: + scope_burst.stop_burst(scope) + + counts = scope_burst.split_row_counts(cumulative) + self._write_burst(scan_file, burst_start, row, counts, + n_frames, samples_per_frame) + + for r in range(n_burst): + result.rows_written += 1 + self._cb.on_row_done(row + r + 1, pa.n_rows, ai + 1, n_angles) + row += n_burst + + def _acquire_gated_row(self, y_pos: float, x_lead_in: float, x_end: float): + """One row: step Y, fly back gated off, then acquire on the +X pass.""" + ctrl = self._stage + ctrl.move_axis_absolute(AXIS_Y, y_pos, timeout=60.0) + ctrl.move_axis_absolute(AXIS_X, x_lead_in, timeout=30.0) + ctrl.arm_scan_gate(AXIS_X, True) + ctrl.move_axis_absolute(AXIS_X, x_end, timeout=120.0) + ctrl.arm_scan_gate(AXIS_X, False) + time.sleep(scope_burst.BURST_ROW_SETTLE_S) + + def _gate_off_preflight(self, x_lead_in: float, x_end: float): + """Prove the gate really gates before trusting a multi-row burst. + + The value that makes the BBD trigger output idle low is not settled by + the protocol docs (see apt_constants.TRIGOUT_GATE_OFF), and getting it + wrong fills every burst with flyback frames that silently shift the + file. The scope already measures the gate on CH3, so this needs no + bench probe: one gated-off flyback must acquire nothing, and one gated + pass must acquire something — the second half is what stops a dark + laser from making the first half pass vacuously. + + Leaves the stage parked at x_end, where the burst loop expects it. + """ + ctrl, scope = self._stage, self._scope + self._cb.on_status("Burst preflight: checking the stage gate …") + + ctrl.arm_scan_gate(AXIS_X, False) + ctrl.move_axis_absolute(AXIS_X, x_end, timeout=120.0) + baseline = scope_burst.start_burst(scope, self._max_frames) + ctrl.move_axis_absolute(AXIS_X, x_lead_in, timeout=120.0) + scope_burst.stop_burst(scope) + leaked = scope_burst.frames_acquired(scope) - baseline + + ctrl.arm_scan_gate(AXIS_X, True) + baseline = scope_burst.start_burst(scope, self._max_frames) + ctrl.move_axis_absolute(AXIS_X, x_end, timeout=120.0) + ctrl.arm_scan_gate(AXIS_X, False) + scope_burst.stop_burst(scope) + gated = scope_burst.frames_acquired(scope) - baseline + + if gated <= 0: + raise RuntimeError( + "Burst preflight: no frames acquired with the gate armed. " + "Check that the Genesis laser is pulsing (CH2) and that the " + "BBD X trigger output reaches CH3 before scanning." + ) + if leaked: + raise RuntimeError( + f"Burst preflight: {leaked} frame(s) acquired during a flyback " + "that should have been gated off — the BBD trigger output is " + "not idling low. Set apt_constants.TRIGOUT_GATE_OFF to " + "TriggerBitsServo.TRIGOUT_HIGH and retry, or use per-row " + "acquisition." + ) + self._cb.on_status( + f"Burst preflight OK ({gated} frames gated on, 0 leaked).") + + def _write_burst(self, scan_file, burst_start: int, first_row: int, + counts: list[int], n_frames: int, samples_per_frame: int): + """Deinterleave one burst into the file's per-row, per-channel blocks. + + The wire is channel-major (every row of CH1, then every row of CH4); + the file is row-major with channels inner. Writing one channel at a + time to strided offsets keeps peak memory at a single channel's burst + instead of the whole thing. + """ + scope = self._scope + ch_bytes = n_frames * samples_per_frame + row_bytes = len(SCAN_CHANNELS) * ch_bytes + total_frames = sum(counts) + + for r, count in enumerate(counts): + self._warn_frame_delta(first_row + r, count, n_frames) + + for ch_idx, ch in enumerate(SCAN_CHANNELS): + if ch == 3: + self._cb.on_status("Writing zeroed CH3 frames …") + blob = None + else: + self._cb.on_status( + f"Fetching CH{ch} burst ({total_frames} frames) …") + blob = scope_burst.transfer_burst(scope, ch, total_frames, + samples_per_frame) + src = 0 + zeros = bytes(ch_bytes) if blob is None else None + for r, count in enumerate(counts): + scan_file.seek(burst_start + r * row_bytes + ch_idx * ch_bytes) + if blob is None: + scan_file.write(zeros) + else: + row = scope_burst.normalize_row( + blob, src, count, n_frames, samples_per_frame) + if ch == 4: + self._cb.on_dc_bias( + first_row + r + 1, + scope_burst.frame_means_block( + row, 0, n_frames, samples_per_frame)) + scan_file.write(row) + src += count * samples_per_frame + del blob + + scan_file.seek(burst_start + len(counts) * row_bytes) + + def _warn_frame_delta(self, row_idx: int, count: int, n_frames: int): + if count == n_frames: + return + verb = "zero-padded" if count < n_frames else "truncated" + msg = (f"Row {row_idx + 1}: {count} frames acquired, {n_frames} " + f"expected — {verb} to keep the file layout intact.") + logger.warning(msg) + self._cb.on_status(msg) diff --git a/core/scope_burst.py b/core/scope_burst.py new file mode 100644 index 0000000..dae8519 --- /dev/null +++ b/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) diff --git a/core/scope_sras.py b/core/scope_sras.py index 30ef616..0f7885d 100644 --- a/core/scope_sras.py +++ b/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() diff --git a/tests/fakes.py b/tests/fakes.py index ec1e7ba..ffddcbe 100644 --- a/tests/fakes.py +++ b/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): diff --git a/tests/test_scan_engine.py b/tests/test_scan_engine.py index 1ce547d..f338929 100644 --- a/tests/test_scan_engine.py +++ b/tests/test_scan_engine.py @@ -20,22 +20,24 @@ from fakes import FakeScope, FakeStage, FakeT3R, Trace SPF = 8 -def make_plan(num_angles=1): - # Small ROI well inside the stage limits: 1 row, few frames per angle. - return build_plan(40.0, 30.0, 0.02, 0.005, num_angles, 0.01, +def make_plan(num_angles=1, y_delta=0.005): + # Small ROI well inside the stage limits: few rows, few frames per angle. + return build_plan(40.0, 30.0, 0.02, y_delta, num_angles, 0.01, laser_freq_hz=20000.0, velocity_mm_s=100.0) -def build(tmp_path, num_angles=1, callbacks=None, resume=None, **kw): +def build(tmp_path, num_angles=1, callbacks=None, resume=None, plan=None, + burst_mode=False, max_frames=4096, out_name="out.sras", **kw): trace = Trace() - stage = FakeStage(trace) - scope = FakeScope(trace, samples_per_frame=SPF) + scope = FakeScope(trace, samples_per_frame=SPF, max_frames=max_frames) + stage = FakeStage(trace, scope=scope) t3r = FakeT3R(trace, **kw) rotator = RotationAxis(t3r, RotationSettings()) - plan = make_plan(num_angles) - engine = ScanEngine(stage, scope, rotator, plan, tmp_path / "out.sras", + plan = plan if plan is not None else make_plan(num_angles) + engine = ScanEngine(stage, scope, rotator, plan, tmp_path / out_name, resume=resume, - callbacks=callbacks or ScanCallbacks()) + callbacks=callbacks or ScanCallbacks(), + burst_mode=burst_mode) return engine, trace, plan @@ -192,8 +194,8 @@ def test_dc_bias_callback_reports_per_frame_means(tmp_path): def test_offstage_plan_rejected_before_touching_hardware(tmp_path): trace = Trace() - stage = FakeStage(trace) scope = FakeScope(trace, samples_per_frame=SPF) + stage = FakeStage(trace, scope=scope) # X range that runs off the 110 mm stage once ramps are added plan = build_plan(80.0, 30.0, 40.0, 5.0, 1, 0.25, laser_freq_hz=20000.0, velocity_mm_s=100.0) @@ -265,6 +267,137 @@ def test_resume_record_length_mismatch_rejected(tmp_path): engine2.run() +# ── Burst acquisition ──────────────────────────────────────────────────────── + +# 6 rows × 4 frames/row; max_frames=14 gives 14//4 = 3 rows per burst, so the +# angle needs two bursts and the second is not a whole burst wide. +BURST_PLAN = dict(y_delta=0.05) +BURST_MAX_FRAMES = 14 + + +def test_burst_and_serial_produce_identical_files(tmp_path): + """The whole point: burst mode must be a pure acquisition optimisation.""" + plan = make_plan(**BURST_PLAN) + assert plan.per_angle[0].n_rows == 6 and plan.per_angle[0].n_frames == 4 + + serial, _, _ = build(tmp_path, plan=plan, out_name="serial.sras") + serial.run() + burst, _, _ = build(tmp_path, plan=plan, out_name="burst.sras", + burst_mode=True, max_frames=BURST_MAX_FRAMES) + burst.run() + + assert (tmp_path / "burst.sras").read_bytes() == \ + (tmp_path / "serial.sras").read_bytes() + + +def test_burst_multi_angle_file_is_complete(tmp_path): + plan = make_plan(num_angles=3, **BURST_PLAN) + engine, trace, _ = build(tmp_path, plan=plan, burst_mode=True, + max_frames=BURST_MAX_FRAMES) + result = engine.run() + + assert result.rows_written == plan.total_rows + assert result.angles_acquired == [0, 1, 2] + sras = SrasFile(tmp_path / "out.sras") + assert [s.status for s in sras.angle_status()] == ["OK"] * 3 + + +def test_burst_gates_the_flyback_and_runs_once_per_burst(tmp_path): + plan = make_plan(**BURST_PLAN) + engine, trace, _ = build(tmp_path, plan=plan, burst_mode=True, + max_frames=BURST_MAX_FRAMES) + engine.run() + + # Two bursts (3 + 3 rows), two preflight acquisitions, one background. + runs = [c for c in trace.of("write") if c[1] == "ACQuire:STATE RUN"] + assert len(runs) == 5 + + # Every acquiring pass is bracketed by an arm/disarm, so the gate is low + # for each flyback. 6 rows + 1 preflight pass = 7 arms. + gate = [c[2] for c in trace.of("arm_scan_gate")] + assert gate.count(True) == 7 + # No two arms without a disarm between them — that is what would let a + # flyback into the acquisition. (A repeated disarm is just defensive.) + for a, b in zip(gate, gate[1:], strict=False): + assert not (a and b), f"acquiring pass with no disarm before it: {gate}" + assert gate[-1] is False, "scan left the gate armed" + + # One bulk transfer per data channel per burst, none per row. + assert [(c[1], c[2]) for c in trace.of("transfer_fastframe_bulk")] == [ + (1, 12), (4, 12), (1, 12), (4, 12)] + assert trace.count("transfer_fastframe") == 0 + + +def test_burst_preflight_rejects_a_leaky_gate(tmp_path): + plan = make_plan(**BURST_PLAN) + engine, trace, _ = build(tmp_path, plan=plan, burst_mode=True, + max_frames=BURST_MAX_FRAMES) + stage = engine._stage + + # A gate that ignores the disable request — the failure mode the preflight + # exists to catch (TRIGOUT_GATE_OFF set to the wrong mode value). + def stuck_gate(axis, armed, verify=True): + trace.record("arm_scan_gate", axis, bool(armed)) + stage.gate_armed = True + stage.arm_scan_gate = stuck_gate + + with pytest.raises(RuntimeError, match="not idling low"): + engine.run() + + +def test_burst_preflight_rejects_a_dark_laser(tmp_path): + """A gate that never fires would let a leak check pass vacuously.""" + plan = make_plan(**BURST_PLAN) + engine, trace, _ = build(tmp_path, plan=plan, burst_mode=True, + max_frames=BURST_MAX_FRAMES) + engine._stage.attach_scope(None) # no pulses ever reach the scope + + with pytest.raises(RuntimeError, match="no frames acquired"): + engine.run() + + +@pytest.mark.parametrize("burst_mode", [False, True]) +def test_short_row_is_padded_to_declared_frame_count(tmp_path, burst_mode): + """A clipped row must not shift every later row in the file. + + v6 declares n_frames per row up front and has no per-row length, so an + under-triggered row has to be squared up. In burst mode this also proves + the splitter advances by what actually arrived, not by n_frames. + """ + warnings = [] + plan = make_plan(**BURST_PLAN) + engine, trace, _ = build( + tmp_path, plan=plan, burst_mode=burst_mode, + max_frames=BURST_MAX_FRAMES, + callbacks=ScanCallbacks(on_status=warnings.append)) + # The preflight is covered by its own tests; skipping it keeps the + # acquiring-pass count below identical in both modes. + engine._preflight_done = True + + scope = engine._scope + real_acquire = scope.acquire_frames + passes = {"n": 0} + + def clipped(n): + if scope._running: + passes["n"] += 1 + if passes["n"] == 2: # second data row loses one frame + n -= 1 + real_acquire(n) + scope.acquire_frames = clipped + + result = engine.run() + + assert result.rows_written == 6 + assert any("Row 2: 3 frames acquired, 4 expected" in w for w in warnings) + assert any("zero-padded" in w for w in warnings) + sras = SrasFile(tmp_path / "out.sras") + assert [s.status for s in sras.angle_status()] == ["OK"] + # The padding lands at the end of the short row, not in the next one. + assert bytes(sras.load_row(0, 1, 0)[-1]) == bytes(SPF) + assert bytes(sras.load_row(0, 2, 0)[0]) != bytes(SPF) + + def test_engine_imports_without_qt(): """The engine must be usable from a non-Qt front end.""" import subprocess diff --git a/tests/test_scope_burst.py b/tests/test_scope_burst.py new file mode 100644 index 0000000..c119cf7 --- /dev/null +++ b/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]