From 95c273dce542021eab8107b476547c00f44beaf2 Mon Sep 17 00:00:00 2001 From: Thomas Ales Date: Thu, 30 Jul 2026 23:25:45 -0500 Subject: [PATCH] Add test tooling: synthetic .sras generator and equivalence harness tools/make_test_sras.py writes small v6 files with per-angle-varying geometry, distinct per-channel calibration, and deterministic waveform content (predictable FFT peak per pixel, predictable DC mean per pixel). tools/check_equivalence.py hashes DC/FFT/alignment outputs across channels, angles, bg-sub, pad factors, and thresholds. It imports from either the monolith or the split modules, so the same script captures both sides of a refactor. Hashes canonicalise to native float64 so a dtype or byte-order change that preserves values is not a false failure. Co-Authored-By: Claude Opus 5 --- tools/__init__.py | 0 tools/check_equivalence.py | 197 +++++++++++++++++++++++++++++++++++++ tools/make_test_sras.py | 142 ++++++++++++++++++++++++++ 3 files changed, 339 insertions(+) create mode 100644 tools/__init__.py create mode 100644 tools/check_equivalence.py create mode 100644 tools/make_test_sras.py diff --git a/tools/__init__.py b/tools/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tools/check_equivalence.py b/tools/check_equivalence.py new file mode 100644 index 0000000..fa11968 --- /dev/null +++ b/tools/check_equivalence.py @@ -0,0 +1,197 @@ +#!/usr/bin/env python3 +"""Golden-output equivalence harness for the sras-viewer refactor. + +Computes a battery of DC / FFT / alignment outputs and prints a stable hash +for each. Run it on the pre-refactor commit to capture a baseline, then again +after the refactor and diff the two reports — every line must match. + +Imports work against both the pre-refactor monolith (`sras_viewer`) and the +post-refactor split (`sras_format` + `sras_compute`), so the *same* script +produces both sides of the comparison. + +Hashes canonicalise to native little-endian float64 before hashing, so a +deliberate dtype/byte-order change that preserves values does not show up as +a false mismatch. + +Usage: + python tools/check_equivalence.py --out baseline.txt + python tools/check_equivalence.py --out after.txt --real /path/to/big.sras + diff baseline.txt after.txt +""" + +import argparse +import copy +import hashlib +import sys +from pathlib import Path + +import numpy as np + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +# --- Import shim: split modules if present, else the monolith -------------- +try: + from sras_format import SrasFile, CH1_IDX, CH3_IDX, CH4_IDX, adc_to_mv + import sras_compute as C + _LAYOUT = "split" +except ImportError: + import sras_viewer as _V + from sras_viewer import SrasFile, CH1_IDX, CH3_IDX, CH4_IDX, adc_to_mv + C = _V + _LAYOUT = "monolith" + +compute_dc_image = C.compute_dc_image +compute_rf_image = C.compute_rf_image +compute_alignment = C._compute_angle_alignment +apply_alignment = C.apply_alignment + +import tools.make_test_sras as gen # noqa: E402 + + +def h(arr) -> str: + """Stable hash of an array's *values*, independent of dtype/byte order.""" + a = np.ascontiguousarray(np.asarray(arr, dtype=np.float64)) + return hashlib.sha256(a.tobytes()).hexdigest()[:16] + + +def row_slice(sras: SrasFile, angle_idx: int, n_rows: int) -> SrasFile: + """A shallow view of *sras* restricted to the first *n_rows* rows of + *angle_idx*, so the huge real file can be exercised in seconds.""" + view = copy.copy(sras) + n = min(int(n_rows), int(sras.n_rows[angle_idx])) + view.n_rows = np.array(sras.n_rows, copy=True) + view.n_rows[angle_idx] = n + view.data = list(sras.data) + view.data[angle_idx] = sras.data[angle_idx][:n] + view._y_pos_per_angle = list(sras._y_pos_per_angle) + view._y_pos_per_angle[angle_idx] = sras._y_pos_per_angle[angle_idx][:n] + return view + + +def report(lines: list[str], label: str, value: str): + lines.append(f"{label:<58} {value}") + + +def check_file(path: Path, lines: list[str], tag: str, + angles: list[int], n_rows: int | None): + sras = SrasFile(str(path)) + report(lines, f"[{tag}] version/n_angles", + f"v{sras.version} n={sras.n_angles}") + report(lines, f"[{tag}] geometry", + f"rows={list(map(int, sras.n_rows))} frames={list(map(int, sras.n_frames))}") + report(lines, f"[{tag}] calibration", + " ".join(f"{m:.6g}/{o:.6g}/{z:.6g}" for m, o, z in + zip(sras.ch_ymult_mv, sras.ch_yoff_adc, sras.ch_yzero_mv))) + report(lines, f"[{tag}] angles_deg", h(sras.angles_deg)) + if sras.background is not None: + report(lines, f"[{tag}] background", h(sras.background)) + + for a in angles: + if a >= sras.n_angles: + continue + s = row_slice(sras, a, n_rows) if n_rows else sras + + report(lines, f"[{tag}] x_axis_mm(a={a})", h(s.x_axis_mm(a))) + report(lines, f"[{tag}] y_positions_mm(a={a})", h(s.y_positions_mm(a))) + + for ch, name in ((CH3_IDX, "CH3"), (CH4_IDX, "CH4")): + raw = compute_dc_image(s, a, ch) + report(lines, f"[{tag}] dc_adc(a={a},{name})", h(raw)) + mv = adc_to_mv(raw, s.ch_ymult_mv[ch], s.ch_yoff_adc[ch], + s.ch_yzero_mv[ch]) + report(lines, f"[{tag}] dc_mv(a={a},{name})", h(mv)) + + dc4 = adc_to_mv(compute_dc_image(s, a, CH4_IDX), + s.ch_ymult_mv[CH4_IDX], s.ch_yoff_adc[CH4_IDX], + s.ch_yzero_mv[CH4_IDX]) + thresholds = [-1e9, float(np.median(dc4))] + + for bg in (False, True): + if bg and s.background is None: + continue + for pad in (1, 2): + n_fft = s.samples_per_frame * pad if pad > 1 else None + for ti, thr in enumerate(thresholds): + img = compute_rf_image(s, a, dc_threshold_mv=thr, + apply_bg_sub=bg, n_fft=n_fft) + report(lines, + f"[{tag}] rf(a={a},bg={int(bg)},pad={pad},thr{ti})", + h(img)) + # dc4_mv passthrough must give the identical result + img2 = compute_rf_image(s, a, dc_threshold_mv=thr, + apply_bg_sub=bg, n_fft=n_fft, + dc4_mv=dc4) + same = "SAME" if h(img) == h(img2) else "DIFFER" + report(lines, + f"[{tag}] rf-dc4arg(a={a},bg={int(bg)},pad={pad},thr{ti})", + same) + + +def check_alignment(path: Path, lines: list[str], tag: str): + sras = SrasFile(str(path)) + if sras.n_angles < 2: + return + dc4 = adc_to_mv(compute_dc_image(sras, 0, CH4_IDX), + sras.ch_ymult_mv[CH4_IDX], sras.ch_yoff_adc[CH4_IDX], + sras.ch_yzero_mv[CH4_IDX]) + thr = float(np.median(dc4)) + res = compute_alignment(sras, 0, thr) + report(lines, f"[{tag}] align canvas_shape", str(res.canvas_shape)) + report(lines, f"[{tag}] align canvas_origin", + f"{res.canvas_origin_mm[0]:.9g},{res.canvas_origin_mm[1]:.9g}") + report(lines, f"[{tag}] align canvas_pitch", + f"{res.canvas_dx_mm:.9g},{res.canvas_dy_mm:.9g}") + for a in sorted(res.per_angle): + t = res.per_angle[a] + report(lines, f"[{tag}] align shift_mm(a={a})", + f"{t.shift_mm[0]:.9g},{t.shift_mm[1]:.9g}") + report(lines, f"[{tag}] align rot(a={a})", f"{t.rotation_deg:.9g}") + report(lines, f"[{tag}] align matrix(a={a})", h(t.matrix)) + report(lines, f"[{tag}] align offset(a={a})", h(t.offset)) + img = compute_dc_image(sras, a, CH4_IDX) + report(lines, f"[{tag}] align resampled(a={a})", + h(apply_alignment(res, a, img))) + + +def main(): + p = argparse.ArgumentParser(description=__doc__) + p.add_argument("--out", required=True, help="report file to write") + p.add_argument("--real", help="optional path to a real .sras file") + p.add_argument("--real-rows", type=int, default=2, + help="rows per angle to sample from the real file") + p.add_argument("--real-angles", type=int, default=2, + help="how many angles to sample from the real file") + p.add_argument("--scratch", default=".", + help="directory for generated synthetic files") + args = p.parse_args() + + lines = [f"# layout: {_LAYOUT}", f"# numpy: {np.__version__}"] + + scratch = Path(args.scratch) + synth = scratch / "equiv_synth.sras" + gen.write(synth, n_angles=4, seed=0, samples_per_frame=64) + check_file(synth, lines, "synth", angles=[0, 1, 2, 3], n_rows=None) + check_alignment(synth, lines, "synth") + + # A second synthetic with an odd sample count, to catch off-by-one in + # rfft bin handling and chunk-boundary arithmetic. + synth_odd = scratch / "equiv_synth_odd.sras" + gen.write(synth_odd, n_angles=2, seed=7, samples_per_frame=37) + check_file(synth_odd, lines, "odd", angles=[0, 1], n_rows=None) + + if args.real: + real = Path(args.real) + if real.exists(): + check_file(real, lines, "real", + angles=list(range(args.real_angles)), + n_rows=args.real_rows) + else: + lines.append(f"# real file not found: {real}") + + Path(args.out).write_text("\n".join(lines) + "\n") + print("\n".join(lines)) + print(f"\nWrote {args.out} ({len(lines)} lines)") + + +if __name__ == "__main__": + main() diff --git a/tools/make_test_sras.py b/tools/make_test_sras.py new file mode 100644 index 0000000..5ea0e00 --- /dev/null +++ b/tools/make_test_sras.py @@ -0,0 +1,142 @@ +#!/usr/bin/env python3 +"""Generate small synthetic .sras files for testing. + +Writes v6 files (per-angle geometry, ragged waveform blocks) matching +scan_format.md, with deterministic pseudo-random waveform content so a test +can compute expected DC/FFT images independently of the reader under test. + +Usage: + python tools/make_test_sras.py out.sras [--angles 3] [--seed 0] +""" + +import argparse +import struct +from pathlib import Path + +import numpy as np + +HDR_FMT_V6 = ">4sBHfffffffIdBB" +GEO_FMT_V6 = ">ffIH" + +# Per-angle (n_rows, n_frames) — deliberately different per angle so ragged +# geometry handling is actually exercised. +_GEOMETRY = [(5, 7), (4, 11), (6, 9), (3, 13), (7, 6)] + +_SAMPLE_RATE_HZ = 6.25e9 +_VELOCITY_MM_S = 20.0 +_LASER_FREQ_HZ = 1000.0 +_ROW_SPACING_MM = 0.05 + + +def _preamble(ymult_v: float, yoff_adc: float, yzero_v: float) -> bytes: + """A Tektronix WFMOutpre string in verbose (keyword) form — the reader + pulls YMULT/YOFF/YZERO out of it by name, so the keywords must be + present literally. YMULT/YZERO are in volts, as the scope reports them.""" + return ( + ":WFMOUTPRE:BYT_NR 1;BIT_NR 8;ENCDG BIN;BN_FMT RI;BYT_OR MSB;" + 'WFID "Ch1, DC coupling";NR_PT 2500;PT_FMT Y;' + "XINCR 1.6000E-10;XZERO 0.0E0;XUNIT \"s\";" + f"YMULT {ymult_v:.6E};YOFF {yoff_adc:.6E};YZERO {yzero_v:.6E};" + 'YUNIT "V"' + ).encode("utf-8") + + +def build(n_angles: int, seed: int, samples_per_frame: int) -> tuple[bytes, dict]: + rng = np.random.default_rng(seed) + geom = [_GEOMETRY[a % len(_GEOMETRY)] for a in range(n_angles)] + n_ch = 3 + bps = 1 + + angles_deg = np.linspace(0.0, 60.0, n_angles, dtype=np.float32) + # Distinct calibration per channel so a swapped-channel bug is visible. + cal = [ + (1.5625e-3, -87.04, 0.0), + (2.0000e-3, -60.00, 1.0e-3), + (2.5000e-3, -40.00, -2.0e-3), + ] + + out = bytearray() + out += struct.pack( + HDR_FMT_V6, b"SRAS", 6, n_angles, + 0.0, 0.0, 1.0, 1.0, _ROW_SPACING_MM, + _VELOCITY_MM_S, _LASER_FREQ_HZ, + samples_per_frame, _SAMPLE_RATE_HZ, bps, n_ch, + ) + out += angles_deg.astype(">f4").tobytes() + + x_starts = [] + for a, (n_rows, n_frames) in enumerate(geom): + x_start = -0.5 + 0.1 * a + x_starts.append(x_start) + out += struct.pack(GEO_FMT_V6, x_start, 1.0, n_frames, n_rows) + + y_positions = [] + for a, (n_rows, _) in enumerate(geom): + y = (0.2 * a + np.arange(n_rows) * _ROW_SPACING_MM).astype(np.float32) + y_positions.append(y) + out += y.astype(">f4").tobytes() + + for ymult_v, yoff, yzero_v in cal: + p = _preamble(ymult_v, yoff, yzero_v) + out += struct.pack(">H", len(p)) + p + + background = rng.integers(-8, 9, size=samples_per_frame, dtype=np.int8) + out += struct.pack(">I", samples_per_frame) + background.tobytes() + + # Waveform data. CH1 gets a sinusoid at a per-pixel frequency so the FFT + # peak is predictable; CH3/CH4 get per-pixel DC levels so the mean is too. + t = np.arange(samples_per_frame) + waveforms = [] + for a, (n_rows, n_frames) in enumerate(geom): + block = np.empty((n_rows, n_ch, n_frames, samples_per_frame), dtype=np.int8) + for r in range(n_rows): + for f in range(n_frames): + bin_idx = 3 + ((a + r + f) % 17) + phase = 2 * np.pi * bin_idx * t / samples_per_frame + block[r, 0, f] = np.clip( + np.round(60 * np.sin(phase)), -128, 127).astype(np.int8) + block[r, 1, f] = np.int8((a * 7 + r * 3 + f) % 100 - 50) + block[r, 2, f] = np.int8((a * 5 + r * 11 + f * 2) % 120 - 60) + waveforms.append(block) + out += block.tobytes() + + meta = { + "n_angles": n_angles, + "geometry": geom, + "angles_deg": angles_deg, + "x_starts": x_starts, + "y_positions": y_positions, + "cal": cal, + "background": background, + "waveforms": waveforms, + "samples_per_frame": samples_per_frame, + "sample_rate_hz": _SAMPLE_RATE_HZ, + } + return bytes(out), meta + + +def write(path: Path, n_angles: int = 3, seed: int = 0, + samples_per_frame: int = 64) -> dict: + payload, meta = build(n_angles, seed, samples_per_frame) + path.write_bytes(payload) + return meta + + +def main(): + p = argparse.ArgumentParser(description=__doc__) + p.add_argument("output") + p.add_argument("--angles", type=int, default=3) + p.add_argument("--seed", type=int, default=0) + p.add_argument("--spf", type=int, default=64, help="samples per frame") + args = p.parse_args() + + out = Path(args.output) + meta = write(out, args.angles, args.seed, args.spf) + print(f"Wrote {out} ({out.stat().st_size:,} bytes)") + print(f" angles : {meta['n_angles']}") + print(f" geometry : {meta['geometry']}") + print(f" spf : {meta['samples_per_frame']}") + + +if __name__ == "__main__": + main()