tka/sras-viewer
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Convert test scripts to pytest; extend the equivalence harness

Browse Source 
- pyproject.toml replaces sras_viewer_requirements.txt (same pins) and
  adds a dev extra with pytest.
- tools/test_refactor.py, test_alignment.py, test_gui.py become
  tests/test_compute.py, tests/test_alignment.py, tests/test_gui.py with
  assertions preserved verbatim. test_gui.py stays one ordered
  integration sequence over a shared module-scoped window.
- check_equivalence.py: drop the dead pre-refactor monolith shim (and the
  _compute_angle_alignment alias it consumed), extend the pad sweep to
  (1, 2, 4, 8, 40), add legacy-v4 and big-endian int16 legs (new bps=2
  option in make_test_sras) so the padded FFT path and the >i2 memmap
  path are in the baseline before the FFT rewrite.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Thomas Ales
2026-08-06 09:45:35 -05:00
parent dc513e3fd0
commit 007089dd48
12 changed files with 1065 additions and 1104 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tools/check_equivalence.py
+24 -27
View File
@@ -1,13 +1,9 @@
#!/usr/bin/env python3
"""Golden-output equivalence harness for the sras-viewer refactor.
"""Golden-output equivalence harness for compute-path refactors.
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.
for each. Run it before a refactor to capture a baseline, then again after
and diff the two reports — every line must match.
Hashes canonicalise to native little-endian float64 before hashing, so a
deliberate dtype/byte-order change that preserves values does not show up as
@@ -29,23 +25,11 @@ 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
from sras_format import SrasFile, CH3_IDX, CH4_IDX, adc_to_mv # noqa: E402
from sras_compute import ( # noqa: E402
apply_alignment, compute_angle_alignment, compute_dc_image, compute_rf_image,
)
import tools.make_test_sras as gen # noqa: E402
def h(arr) -> str:
@@ -109,7 +93,7 @@ def check_file(path: Path, lines: list[str], tag: str,
for bg in (False, True):
if bg and s.background is None:
continue
for pad in (1, 2):
for pad in (1, 2, 4, 8, 40):
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,
@@ -135,7 +119,7 @@ def check_alignment(path: Path, lines: list[str], tag: str):
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)
res = compute_angle_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}")
@@ -165,7 +149,7 @@ def main():
help="directory for generated synthetic files")
args = p.parse_args()
lines = [f"# layout: {_LAYOUT}", f"# numpy: {np.__version__}"]
lines = [f"# numpy: {np.__version__}"]
scratch = Path(args.scratch)
synth = scratch / "equiv_synth.sras"
@@ -179,6 +163,19 @@ def main():
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)
# A legacy v4 file exercises the uniform-geometry legacy layout through
# the same DC/FFT battery.
synth_v4 = scratch / "equiv_synth_v4.sras"
gen.write_legacy(synth_v4, version=4, n_angles=2, n_rows=6,
n_frames=14, samples_per_frame=48, seed=5)
check_file(synth_v4, lines, "v4", angles=[0, 1], n_rows=None)
# A big-endian int16 v6 file (real acquisitions are >i2; the other
# synthetics are int8).
synth_i16 = scratch / "equiv_synth_i16.sras"
gen.write(synth_i16, n_angles=2, seed=9, samples_per_frame=64, bps=2)
check_file(synth_i16, lines, "int16", angles=[0, 1], n_rows=None)
if args.real:
real = Path(args.real)
if real.exists():
tools/make_test_sras.py
+8 -5
View File
@@ -42,12 +42,12 @@ def _preamble(ymult_v: float, yoff_adc: float, yzero_v: float) -> bytes:
def build(n_angles: int, seed: int, samples_per_frame: int,
geometry: list[tuple[int, int]] | None = None) -> tuple[bytes, dict]:
geometry: list[tuple[int, int]] | None = None,
bps: int = 1) -> tuple[bytes, dict]:
rng = np.random.default_rng(seed)
src_geom = geometry or _GEOMETRY
geom = [src_geom[a % len(src_geom)] 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.
@@ -100,7 +100,9 @@ def build(n_angles: int, seed: int, samples_per_frame: int,
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()
# bps=2 stores the same values big-endian int16, exercising the
# reader's >i2 memmap path.
out += (block.astype(">i2") if bps == 2 else block).tobytes()
meta = {
"n_angles": n_angles,
@@ -119,8 +121,9 @@ def build(n_angles: int, seed: int, samples_per_frame: int,
def write(path: Path, n_angles: int = 3, seed: int = 0,
samples_per_frame: int = 64,
geometry: list[tuple[int, int]] | None = None) -> dict:
payload, meta = build(n_angles, seed, samples_per_frame, geometry)
geometry: list[tuple[int, int]] | None = None,
bps: int = 1) -> dict:
payload, meta = build(n_angles, seed, samples_per_frame, geometry, bps=bps)
path.write_bytes(payload)
return meta
tools/test_alignment.py
-223
View File
@@ -1,223 +0,0 @@
#!/usr/bin/env python3
"""Angle-alignment tests: does registration actually stack the scans?
Builds a synthetic scan in which one sample is imaged at several *known*
rotations and offsets (tools/make_test_sras.write_rotating) and checks that the
alignment path recovers them, that the shared canvas is angle 0's own pixel
grid extended, and that nothing in the result depends on any other angle's
stage coordinates.
No Qt — this exercises sras_compute directly. See tools/test_gui.py for the
dialog and Aligned-View plumbing.
Usage: python tools/test_alignment.py
"""
import sys
import tempfile
from pathlib import Path
import numpy as np
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
import sras_compute as compute # noqa: E402
from sras_format import CH4_IDX, SrasFile, adc_to_mv # noqa: E402
import tools.make_test_sras as gen # noqa: E402
# Registration is limited by how far a feature moves per degree: with this
# sample's ~1 mm radius and a ~16 µm registration pitch, a quarter degree is
# already sub-pixel, so it is the floor of what any metric can resolve here.
_ROT_TOL_DEG = 0.5
_SHIFT_TOL_MM = 0.02
_STACK_IOU_MIN = 0.90
_THRESHOLD_MV = 80.0
_failures: list[str] = []
def check(name: str, ok: bool, detail: str = ""):
print(f" {'PASS' if ok else 'FAIL'} {name}" + (f" — {detail}" if detail else ""))
if not ok:
_failures.append(name)
def dc4_images(sras: SrasFile) -> dict[int, np.ndarray]:
return {a: adc_to_mv(compute.compute_dc_image(sras, a, CH4_IDX), *sras.cal(CH4_IDX))
for a in range(sras.n_angles)}
def mm_transform(sras: SrasFile, result, angle_idx: int) -> np.ndarray:
"""Recover the pure mm-space rotation from a canvas->raw affine.
matrix == D @ R^T @ A_out, where A_out and D only carry the canvas and
per-angle pixel pitches; undoing both must leave something orthonormal, or
the transform is smuggling in a scale or a shear.
"""
dx_a, dy_a = compute._pixel_pitch_mm(sras, angle_idx)
A_out = np.array([[0.0, result.canvas_dx_mm], [result.canvas_dy_mm, 0.0]])
D = np.array([[0.0, 1.0 / dy_a], [1.0 / dx_a, 0.0]])
return np.linalg.inv(D) @ result.per_angle[angle_idx].matrix @ np.linalg.inv(A_out)
def main() -> int:
tmpdir = Path(tempfile.mkdtemp(prefix="sras_align_"))
path = tmpdir / "rotating.sras"
meta = gen.write_rotating(path, n_angles=5)
sras = SrasFile(str(path))
truth = meta["truth"]
print(f"\nrotating-sample scan: {sras.n_angles} angles, "
f"shapes {[sras.image_shape(a) for a in range(sras.n_angles)]}")
print("\nper-angle rigid registration (rotation + translation, no scale)")
dc4 = dc4_images(sras)
fits = {a: compute.register_angle_to_reference(
sras, a, 0, dc4, dc_threshold_mv=_THRESHOLD_MV)
for a in range(sras.n_angles)}
for a, fit in fits.items():
t_rot, t_shift = truth[a]
rot_err = abs(fit.rotation_deg - t_rot)
shift_err = float(np.hypot(fit.shift_mm[0] - t_shift[0],
fit.shift_mm[1] - t_shift[1]))
check(f"angle {a} rotation within {_ROT_TOL_DEG}° of truth",
rot_err <= _ROT_TOL_DEG,
f"got {fit.rotation_deg:.3f}°, truth {t_rot:.3f}° (err {rot_err:.3f}°)")
check(f"angle {a} translation within {_SHIFT_TOL_MM} mm of truth",
shift_err <= _SHIFT_TOL_MM, f"err {shift_err:.4f} mm")
check("reference angle registers as exact identity",
fits[0] == compute.RigidFit(0.0, (0.0, 0.0), 1.0, "reference"))
# The stage's rotational sense relative to this module's math-positive
# convention is not knowable from the file, and the old code hardcoded a
# guess. Flipping every reported angle must therefore change nothing: the
# search scores both signs and the images decide.
flipped = SrasFile(str(path))
flipped.angles_deg = -flipped.angles_deg
flipped_fits = {a: compute.register_angle_to_reference(
flipped, a, 0, dc4, dc_threshold_mv=_THRESHOLD_MV)
for a in range(1, flipped.n_angles)}
check("negating every reported stage angle changes no fit",
all(flipped_fits[a] == fits[a] for a in flipped_fits),
str({a: (flipped_fits[a].rotation_deg, fits[a].rotation_deg)
for a in flipped_fits if flipped_fits[a] != fits[a]}))
print("\nper-angle stage coordinates are not consulted")
# Move every non-reference angle's scan window somewhere else entirely.
# Only angle 0's coordinates may matter, so every fit must be untouched.
moved = SrasFile(str(path))
for a in range(1, moved.n_angles):
moved.x_start_mm[a] += 13.5 * a
moved._y_pos_per_angle[a] = moved._y_pos_per_angle[a] - 9.25 * a
moved_dc4 = dc4_images(moved)
moved_fits = {a: compute.register_angle_to_reference(
moved, a, 0, moved_dc4, dc_threshold_mv=_THRESHOLD_MV)
for a in range(1, moved.n_angles)}
check("relocating every other angle's scan window changes no fit",
all(moved_fits[a] == fits[a] for a in moved_fits),
str({a: (round(moved_fits[a].rotation_deg, 4), fits[a].rotation_deg)
for a in moved_fits if moved_fits[a] != fits[a]}))
print("\nshared canvas is angle 0's own pixel grid, extended")
result = compute.compute_angle_alignment(sras, 0, _THRESHOLD_MV)
t0 = result.per_angle[0]
check("angle 0's transform has no rotation, scale or shear",
np.allclose(t0.matrix, np.eye(2)), str(t0.matrix))
check("angle 0 lands on whole canvas pixels (no resampling of the reference)",
np.allclose(t0.offset, np.round(t0.offset)), str(t0.offset))
check("canvas pitch is angle 0's own pitch",
(result.canvas_dx_mm, result.canvas_dy_mm)
== compute._pixel_pitch_mm(sras, 0))
n_rows, n_cols = result.canvas_shape
x_axis = result.canvas_origin_mm[0] + np.arange(n_cols) * result.canvas_dx_mm
y_axis = result.canvas_origin_mm[1] + np.arange(n_rows) * result.canvas_dy_mm
row0, col0 = int(round(-t0.offset[0])), int(round(-t0.offset[1]))
a0_rows, a0_cols = sras.image_shape(0)
check("canvas X axis reproduces angle 0's own X coordinates",
np.allclose(x_axis[col0:col0 + a0_cols], sras.x_axis_mm(0)))
check("canvas Y axis reproduces angle 0's own Y coordinates",
np.allclose(y_axis[row0:row0 + a0_rows], sras.y_positions_mm(0)))
check("canvas covers every angle's footprint",
n_rows >= max(int(sras.n_rows[a]) for a in range(sras.n_angles))
and n_cols >= max(int(sras.n_frames[a]) for a in range(sras.n_angles)),
str(result.canvas_shape))
print("\nno scaling anywhere in the per-angle transforms")
for a in range(sras.n_angles):
R = mm_transform(sras, result, a)
check(f"angle {a}'s mm-space transform is a pure rotation",
np.allclose(R @ R.T, np.eye(2), atol=1e-9)
and abs(abs(np.linalg.det(R)) - 1.0) < 1e-9,
f"det={np.linalg.det(R):.6f}")
print("\nall angles stack into one shape")
aligned = {a: compute.apply_alignment(result, a, dc4[a])
for a in range(sras.n_angles)}
base = aligned[0] >= _THRESHOLD_MV
for a in range(1, sras.n_angles):
other = aligned[a] >= _THRESHOLD_MV
iou = float((base & other).sum()) / max(1, int((base | other).sum()))
check(f"angle {a}'s aligned sample overlaps angle 0's (IoU >= {_STACK_IOU_MIN})",
iou >= _STACK_IOU_MIN, f"IoU {iou:.4f}")
print("\ndownsampled preview lands where the full-resolution image does")
# ManualAlignmentDialog reprojects block-mean-downsampled masks, so the
# affine has to account for the factor. When it did not, every preview
# layer came out magnified by that factor and offset — the overlay showed a
# blown-up crop of each mask, which is not something you can align by eye.
pitch = (result.canvas_dx_mm, result.canvas_dy_mm)
a = sras.n_angles - 1
p = result.per_angle[a]
full_mask = (dc4[a] >= _THRESHOLD_MV).astype(np.float32)
full = compute.reproject_mask(
sras, a, 0, full_mask, p.rotation_deg, p.shift_mm, pitch,
result.canvas_origin_mm, result.canvas_shape)
fy, fx = 4, 16
small = compute.reproject_mask(
sras, a, 0, compute._block_mean_2d(full_mask, fy, fx),
p.rotation_deg, p.shift_mm, (pitch[0] * fx, pitch[1] * fy),
result.canvas_origin_mm,
(result.canvas_shape[0] // fy, result.canvas_shape[1] // fx),
src_downsample=(fy, fx))
# Compare in mm, via each layer's own center of mass.
def com_mm(layer, px, py):
rows, cols = np.nonzero(layer > 0.5)
return np.array([cols.mean() * px, rows.mean() * py])
d = com_mm(small, pitch[0] * fx, pitch[1] * fy) - com_mm(full, *pitch)
check("a downsampled preview layer lands within a preview pixel of the "
"full-resolution one",
abs(d[0]) <= abs(pitch[0] * fx) and abs(d[1]) <= abs(pitch[1] * fy),
f"offset {d[0]:+.4f}, {d[1]:+.4f} mm")
print("\nmanual path reproduces the same geometry")
params = {a: compute.ManualAngleParams(t.rotation_deg, t.shift_mm)
for a, t in result.per_angle.items()}
manual = compute.build_manual_alignment(sras, 0, _THRESHOLD_MV, params)
check("build_manual_alignment matches compute_angle_alignment for the same params",
manual.canvas_shape == result.canvas_shape
and np.allclose(manual.canvas_origin_mm, result.canvas_origin_mm)
and all(np.allclose(manual.per_angle[a].matrix, result.per_angle[a].matrix)
and np.allclose(manual.per_angle[a].offset, result.per_angle[a].offset)
for a in range(sras.n_angles)))
print("\nsidecar round-trip")
compute.save_manual_alignment(sras, 0, _THRESHOLD_MV, params)
loaded = compute.load_manual_alignment(sras)
check("sidecar reloads every angle's params",
loaded is not None
and all(np.isclose(loaded.per_angle[a].rotation_deg, params[a].rotation_deg)
and np.allclose(loaded.per_angle[a].shift_mm, params[a].shift_mm)
for a in range(sras.n_angles)))
check("sidecar deletes cleanly", compute.delete_manual_alignment(sras))
print()
if _failures:
print(f"{len(_failures)} FAILURE(S): " + ", ".join(_failures))
return 1
print("All alignment checks passed.")
return 0
if __name__ == "__main__":
sys.exit(main())
tools/test_gui.py
-476
View File
@@ -1,476 +0,0 @@
#!/usr/bin/env python3
"""Headless GUI test: drives SrasViewerWindow through the real Qt widgets,
signals and worker threads under the offscreen platform plugin.
Covers the interactions a manual smoke test would: load, switch angles and
channels, background DC precompute, lazy FFT compute, threshold and bg-sub
changes, angle alignment, manual angle alignment, aligned view, ROI
draw/move, and CSV export.
Usage: QT_QPA_PLATFORM=offscreen python tools/test_gui.py [file.sras]
"""
import json
import os
import sys
import tempfile
from pathlib import Path
from unittest.mock import patch
os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")
import numpy as np # noqa: E402
from PyQt6.QtCore import QEventLoop, Qt, QTimer # noqa: E402
from PyQt6.QtTest import QTest # noqa: E402
from PyQt6.QtWidgets import QApplication, QMessageBox # noqa: E402
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
import sras_compute as compute # noqa: E402
from sras_format import CH1_IDX, CH3_IDX, CH4_IDX, SrasFile # noqa: E402
from sras_viewer import RoiQuad, SrasViewerWindow, VELOCITY_MODE_IDX # noqa: E402
import tools.make_test_sras as gen # noqa: E402
_failures: list[str] = []
def check(name: str, ok: bool, detail: str = ""):
print(f" {'PASS' if ok else 'FAIL'} {name}" + (f" — {detail}" if detail else ""))
if not ok:
_failures.append(name)
def pump(ms: int = 250):
"""Run the event loop for a while so queued signals and worker threads
make progress."""
loop = QEventLoop()
QTimer.singleShot(ms, loop.quit)
loop.exec()
def wait_until(pred, timeout_ms: int = 20000, step: int = 100) -> bool:
waited = 0
while waited < timeout_ms:
if pred():
return True
pump(step)
waited += step
return pred()
def main():
app = QApplication(sys.argv)
errors: list[str] = []
tmpdir = Path(tempfile.mkdtemp(prefix="sras_gui_"))
path = Path(sys.argv[1]) if len(sys.argv) > 1 else tmpdir / "gui.sras"
if len(sys.argv) <= 1:
gen.write(path, n_angles=4, seed=11, samples_per_frame=256)
print(f"\nloading {path.name}")
win = SrasViewerWindow()
win.show()
# Capture anything the app reports as an error via the status bar.
win.statusBar().messageChanged.connect(
lambda m: errors.append(m) if m and "error" in m.lower() else None)
win._load_file(str(path))
check("file loaded", wait_until(lambda: win._sras is not None))
s = win._sras
check("parsed as v6", s.version == 6, f"v{s.version}")
check("defaults to CH4", win.combo_channel.currentIndex() == CH4_IDX)
check("image displayed", win._current_image is not None)
check("angle spinbox ranges over all angles",
win.spin_angle.maximum() == s.n_angles - 1)
check("scan info populated",
win._info["Angles"].text() == f"Angles: {s.n_angles}",
win._info["Angles"].text())
print("\nbackground DC precompute (all angles)")
ok = wait_until(lambda: all((a, CH4_IDX) in win._dc_cache
and (a, CH3_IDX) in win._dc_cache
for a in range(s.n_angles)))
check("every angle cached for CH3 and CH4", ok,
f"{len(win._dc_cache)} entries")
check("status label reports completion",
"ready for all angles" in win.lbl_dc_precompute.text(),
win.lbl_dc_precompute.text())
print("\nangle switching (DC, should be served from cache)")
for a in range(s.n_angles):
win.spin_angle.setValue(a)
win._on_view_changed()
pump(60)
expected = win._sras.image_shape(a)
check(f"angle {a} shows its own geometry {expected}",
win._current_image.shape == expected,
str(win._current_image.shape))
check("no compute job needed for cached DC angles",
not win._job_running("compute"))
print("\nchannel switching")
win.spin_angle.setValue(0)
win._on_view_changed()
pump(60)
win.combo_channel.setCurrentIndex(CH3_IDX)
check("CH3 displayed", wait_until(lambda: win._current_ch == CH3_IDX))
win.combo_channel.setCurrentIndex(CH1_IDX)
check("CH1 (FFT) computed", wait_until(
lambda: win._current_ch == CH1_IDX and not win._job_running("compute")))
check("FFT result cached", len(win._fft_cache) > 0, f"{len(win._fft_cache)} keys")
rf_img = win._current_image
check("FFT image is non-degenerate", len(np.unique(rf_img)) > 1,
f"{len(np.unique(rf_img))} unique values")
print("\nvelocity mode (pure post-multiply, no recompute)")
n_fft_before = len(win._fft_cache)
win.combo_channel.setCurrentIndex(VELOCITY_MODE_IDX)
check("velocity displayed", wait_until(
lambda: win._current_ch == VELOCITY_MODE_IDX and not win._job_running("compute")))
grating = win.spin_grating_um.value()
check("velocity == freq x grating",
np.allclose(win._current_image, rf_img * grating, atol=1e-3))
check("velocity reused the cached FFT", len(win._fft_cache) == n_fft_before,
f"{n_fft_before} -> {len(win._fft_cache)}")
check("grating spinbox visible in velocity mode", win.grp_velocity.isVisible())
print("\nthreshold change (genuine cache-key change)")
win.combo_channel.setCurrentIndex(CH1_IDX)
wait_until(lambda: not win._job_running("compute"))
dc4 = win._dc_cache[(0, CH4_IDX)]
win.spin_threshold_mv.setValue(float(np.median(dc4)))
win._on_threshold_changed()
check("recomputed at new threshold", wait_until(
lambda: not win._job_running("compute") and len(win._fft_cache) > n_fft_before))
check("masking zeroed some pixels",
int((win._current_image == 0).sum()) > 0,
f"{int((win._current_image == 0).sum())} of {win._current_image.size}")
print("\nbackground subtraction toggle")
n_before = len(win._fft_cache)
win.chk_bg_sub.setChecked(False)
check("recomputed without bg-sub", wait_until(
lambda: not win._job_running("compute") and len(win._fft_cache) > n_before))
win.chk_bg_sub.setChecked(True)
pump(200)
check("returning to bg-sub was a cache hit (no recompute)",
not win._job_running("compute"))
print("\nROI")
x = s.x_axis_mm(0)
y = s.y_positions_mm(0)
roi = RoiQuad.from_bbox(float(x[1]), float(y[1]),
float(x[-2]), float(y[-2]))
win.image_canvas.set_roi(roi)
pump(120)
check("ROI registered", win.image_canvas.get_roi() is not None)
check("pixel count reported",
"pixels inside" in win.lbl_roi_npix.text()
and win.lbl_roi_npix.text() != "pixels inside: —",
win.lbl_roi_npix.text())
npix = int(win.lbl_roi_npix.text().split(":")[1])
check("ROI pixel count is plausible",
0 < npix <= win._current_image.size, f"{npix}")
check("Export ROI enabled", win.btn_export_roi.isEnabled())
csv_path = tmpdir / "roi.csv"
with patch("sras_viewer.QFileDialog.getSaveFileName",
return_value=(str(csv_path), "")):
win._on_export_roi_csv()
check("ROI CSV written", csv_path.exists())
if csv_path.exists():
body = [l for l in csv_path.read_text().splitlines() if not l.startswith("#")]
check("ROI CSV has header + one line per pixel",
len(body) == npix + 1, f"{len(body)} lines for {npix} pixels")
img_csv = tmpdir / "img.csv"
with patch("sras_viewer.QFileDialog.getSaveFileName",
return_value=(str(img_csv), "")):
win._on_export_csv()
check("image CSV written", img_csv.exists())
if img_csv.exists():
arr = np.loadtxt(img_csv, delimiter=",")
check("image CSV round-trips the displayed image",
arr.shape == win._current_image.shape
and np.allclose(arr, win._current_image, rtol=1e-5, atol=1e-4))
print("\nROI survives angle and channel switches")
win.spin_angle.setValue(1)
win._on_view_changed()
wait_until(lambda: not win._job_running("compute"))
check("ROI still present after angle switch",
win.image_canvas.get_roi() is not None)
win.combo_channel.setCurrentIndex(CH4_IDX)
wait_until(lambda: win._current_ch == CH4_IDX)
check("ROI still present after channel switch",
win.image_canvas.get_roi() is not None)
print("\nangle alignment (Fusion)")
win.spin_angle.setValue(0)
win._on_view_changed()
wait_until(lambda: not win._job_running("compute"))
check("alignment action enabled", win._alignment_act.isEnabled())
win._on_angle_alignment()
check("alignment completed", wait_until(
lambda: win._alignment_result is not None and not win._job_running("align"),
timeout_ms=60000))
if win._alignment_result is not None:
r = win._alignment_result
check("transform for every angle", len(r.per_angle) == s.n_angles)
check("canvas is at least as large as any single angle",
all(r.canvas_shape[0] >= int(s.n_rows[a])
and r.canvas_shape[1] >= int(s.n_frames[a])
for a in range(s.n_angles)), str(r.canvas_shape))
check("reference angle has zero shift",
r.per_angle[r.ref_angle_idx].shift_mm == (0.0, 0.0))
check("Aligned View auto-enabled and checked",
win.chk_aligned_view.isEnabled() and win.chk_aligned_view.isChecked())
pump(200)
check("displayed image is on the alignment canvas",
win.image_canvas._img_shape == r.canvas_shape,
f"{win.image_canvas._img_shape} vs {r.canvas_shape}")
win.chk_aligned_view.setChecked(False)
pump(200)
check("unchecking returns to the raw per-angle grid",
win.image_canvas._img_shape == s.image_shape(0),
str(win.image_canvas._img_shape))
print("\nmanual alignment (Fusion)")
check("manual alignment action enabled", win._manual_align_act.isEnabled())
# --- Local mm is anchored on each angle's array center, not its stage --
# position: that is what makes a scan's placement independent of where its
# window happened to sit. (Registration accuracy itself is covered by
# tools/test_alignment.py, which has a synthetic sample to register.)
n_rows, n_frames = s.image_shape(0)
check("array center is the geometric center of the pixel grid",
np.allclose(compute._center_idx(s, 0),
[(n_rows - 1) / 2, (n_frames - 1) / 2]))
dx0, dy0 = compute._pixel_pitch_mm(s, 0)
check("local half-extent is derived from shape and pitch alone",
np.allclose(compute._local_half_extent_mm(s, 0),
[(n_frames - 1) / 2 * abs(dx0), (n_rows - 1) / 2 * abs(dy0)]))
identity = {a: compute.ManualAngleParams() for a in range(s.n_angles)}
origin_a, shape_a = compute.canvas_for_params(s, 0, (dx0, dy0), identity)
moved = SrasFile(str(path))
for a in range(1, moved.n_angles):
moved.x_start_mm[a] += 7.5
moved._y_pos_per_angle[a] = moved._y_pos_per_angle[a] + 3.25
origin_b, shape_b = compute.canvas_for_params(moved, 0, (dx0, dy0), identity)
check("moving every non-reference angle's scan window leaves the canvas "
"unchanged (only angle 0's coordinates are used)",
shape_a == shape_b and np.allclose(origin_a, origin_b),
f"{origin_a} {shape_a} vs {origin_b} {shape_b}")
# --- Both signs of the stage's reported angle are searched --------------
cands = compute._rotation_candidates(30.0, 6.0, 2.0)
check("rotation candidates bracket both signs of the stage angle",
min(cands) < -29.0 and max(cands) > 29.0, f"{min(cands)}..{max(cands)}")
# --- Whole-pixel translation must not wrap content around the edge ------
arr = np.zeros((6, 6), dtype=np.float32)
arr[0, 0] = 1.0
check("_shift_into zero-fills rather than wrapping",
compute._shift_into(arr, -1, -1).sum() == 0.0)
check("_shift_into moves content by exactly the requested offset",
compute._shift_into(arr, 2, 3)[2, 3] == 1.0)
# --- Open: must NOT seed from the still-live automatic AlignmentResult --
# Manual mode exists to fix up whatever the automatic registration got
# wrong, so it must start from identity (every angle centered on the
# reference, no rotation) regardless of whatever the automatic run last
# computed. Only a previously *saved manual* alignment (sidecar) should
# ever seed this dialog.
win._on_manual_alignment()
check("dialog opened", win._manual_align_dialog is not None)
dlg = win._manual_align_dialog
check("mask prep needed no background worker (already DC-cached)",
not win._job_running("manual_align_masks"))
check("no manual sidecar yet -> dialog starts at identity, not the "
"automatic result",
all(dlg._angle_params[a] == compute.ManualAngleParams()
for a in range(s.n_angles)))
# --- Reference angle is locked -------------------------------------------
dlg.combo_active_angle.setCurrentIndex(dlg._ref_angle_idx)
pump(30)
before_ref = dlg._angle_params[dlg._ref_angle_idx]
dlg._on_nudge_translate(1, 0, False)
dlg._on_nudge_rotate(1, False)
check("reference angle group disabled", not dlg.grp_manual_adjust.isEnabled())
check("reference angle untouched by nudge attempts",
dlg._angle_params[dlg._ref_angle_idx] == before_ref)
# --- Nudging a real angle (fine + coarse, translate + rotate) -----------
active = 1 if s.n_angles > 1 else 0
dlg.combo_active_angle.setCurrentIndex(active)
pump(30)
before = dlg._angle_params[active].shift_mm
dlg._on_nudge_translate(1, 0, False) # fine +X
fine_step = dlg.spin_step_translate_mm.value()
check("fine translate nudge moved shift_x by exactly one fine step",
abs(dlg._angle_params[active].shift_mm[0] - (before[0] + fine_step)) < 1e-9)
before = dlg._angle_params[active].shift_mm
dlg._on_nudge_translate(0, -1, True) # coarse -Y
coarse_step = fine_step * dlg.spin_step_multiplier.value()
check("coarse translate nudge uses the multiplier",
abs(dlg._angle_params[active].shift_mm[1] - (before[1] - coarse_step)) < 1e-9)
before_rot = dlg._angle_params[active].rotation_deg
dlg._on_nudge_rotate(1, False)
check("rotate nudge changed rotation_deg",
dlg._angle_params[active].rotation_deg != before_rot)
check("preview canvas rebuilt for every angle after a rotation nudge",
len(dlg._preview_layers) == s.n_angles)
# --- Real key-event wiring (proves keyPressEvent -> signal -> slot) -----
before = dlg._angle_params[active].shift_mm
QTest.keyClick(dlg.canvas, Qt.Key.Key_Right)
check("a real Right-arrow key event nudged shift_x",
dlg._angle_params[active].shift_mm[0] > before[0])
# --- Auto De-rotate: seeds rotation from the stage angle, no translation -
shift_before_derotate = dlg._angle_params[active].shift_mm
dlg._on_auto_derotate()
nominal = compute._nominal_delta_deg(s, active, dlg._ref_angle_idx)
check("auto de-rotate seeded rotation from the stage's reported angle",
abs(dlg._angle_params[active].rotation_deg - nominal) < 1e-6)
check("auto de-rotate left translation untouched",
dlg._angle_params[active].shift_mm == shift_before_derotate)
check("reference angle stays identity after auto de-rotate",
dlg._angle_params[dlg._ref_angle_idx].rotation_deg == 0.0)
# Clicking again offers the other sign, since which one lines the scans up
# is not knowable from the file.
dlg._on_auto_derotate()
check("auto de-rotate offers the opposite sign on a second click",
abs(dlg._angle_params[active].rotation_deg + nominal) < 1e-6)
# --- Auto Cross-Correlate: searches rotation *and* translation ----------
check("cross-correlate action enabled once masks are ready",
dlg.btn_auto_correlate.isEnabled())
for label_idx, (label, _sources) in enumerate(dlg._CORRELATE_SOURCES):
dlg.combo_correlate_source.setCurrentIndex(label_idx)
dlg._on_auto_correlate()
check(f"auto cross-correlate completed ({label})", wait_until(
lambda: not win._job_running("manual_align_correlate"), timeout_ms=60000))
check(f"every non-reference angle got a fit ({label})",
all(a in dlg._fit_notes for a in range(s.n_angles)
if a != dlg._ref_angle_idx))
check("auto cross-correlate reference angle stays identity",
dlg._angle_params[dlg._ref_angle_idx] == compute.ManualAngleParams())
check("auto cross-correlate re-enabled controls when done",
dlg.grp_correlate.isEnabled() and dlg.btn_save.isEnabled())
check("preview canvas rebuilt after cross-correlate",
len(dlg._preview_layers) == s.n_angles)
check("fit quality is reported per angle", bool(dlg._fit_report()),
dlg._fit_report())
# --- Save -----------------------------------------------------------------
dlg._on_save()
sidecar = compute.sidecar_path(s.path)
check("sidecar file written", sidecar.exists())
raw = json.loads(sidecar.read_text()) if sidecar.exists() else {}
check("sidecar schema_version is current",
raw.get("schema_version") == compute._SIDECAR_SCHEMA_VERSION)
check("sidecar per_angle round-trips the dialog's resolved params",
all(raw.get("per_angle", {}).get(str(a), {}).get("rotation_deg")
== dlg._angle_params[a].rotation_deg for a in range(s.n_angles)))
check("main window's alignment_result replaced by the manual build",
win._alignment_result is not None
and win._alignment_result.per_angle[active].rotation_deg
== dlg._angle_params[active].rotation_deg)
check("Aligned View auto-enabled after Save",
win.chk_aligned_view.isEnabled() and win.chk_aligned_view.isChecked())
# --- An old-schema sidecar (pre-pivot/sign fix) is treated as absent ------
stale = dict(raw)
stale["schema_version"] = compute._SIDECAR_SCHEMA_VERSION - 1
sidecar.write_text(json.dumps(stale))
check("a sidecar with an old schema_version is not loaded",
compute.load_manual_alignment(s) is None)
sidecar.write_text(json.dumps(raw)) # restore for the rest of this section
# --- Clear (with confirmation) --------------------------------------------
with patch("sras_viewer.QMessageBox.question",
return_value=QMessageBox.StandardButton.Yes):
dlg._on_clear()
check("sidecar file deleted", not sidecar.exists())
check("dialog params reset to identity",
all(dlg._angle_params[a] == compute.ManualAngleParams()
for a in range(s.n_angles)))
check("main window alignment_result cleared", win._alignment_result is None)
check("Aligned View disabled after Clear",
not win.chk_aligned_view.isEnabled() and not win.chk_aligned_view.isChecked())
dlg.close()
pump(150)
check("dialog reference released on close", win._manual_align_dialog is None)
# --- Sidecar auto-restore on next load ------------------------------------
win._on_manual_alignment()
dlg = win._manual_align_dialog
dlg.combo_active_angle.setCurrentIndex(active)
pump(30)
dlg._on_auto_derotate()
dlg._on_nudge_translate(1, 1, True)
saved_rotation = dlg._angle_params[active].rotation_deg
saved_shift = dlg._angle_params[active].shift_mm
dlg._on_save()
dlg.close()
pump(150)
old_sras_id = id(win._sras)
win._load_file(str(path)) # reload the same file fresh
check("file reloaded", wait_until(
lambda: win._sras is not None and id(win._sras) != old_sras_id))
s = win._sras
check("manual dialog force-closed by a reload", win._manual_align_dialog is None)
check("reload restores the saved manual alignment automatically",
win._alignment_result is not None)
if win._alignment_result is not None:
check("restored rotation matches what was saved",
abs(win._alignment_result.per_angle[active].rotation_deg
- saved_rotation) < 1e-9)
check("restored shift matches what was saved",
win._alignment_result.per_angle[active].shift_mm == saved_shift)
check("Aligned View auto-checked after restoring a saved alignment",
win.chk_aligned_view.isChecked())
print("\npixel inspector")
win.chk_aligned_view.setChecked(False)
pump(100)
win._on_pixel_clicked(0, 0)
pump(150)
check("waveform hint hidden after a click", win.lbl_wave_hint.isHidden())
win.combo_channel.setCurrentIndex(CH1_IDX)
wait_until(lambda: not win._job_running("compute"))
win._on_pixel_clicked(1, 1)
pump(150)
check("RF waveform panel rendered",
len(win.wave_canvas.ax_wave.lines) > 0,
f"{len(win.wave_canvas.ax_wave.lines)} lines")
print("\nshutdown")
win.close()
pump(400)
check("all background jobs released", len(win._jobs) == 0,
f"{list(win._jobs)}")
print()
unexpected = [e for e in errors if e]
if unexpected:
print(f"status-bar errors seen: {unexpected}")
_failures.append("status-bar errors")
if _failures:
print(f"{len(_failures)} FAILURE(S): " + ", ".join(_failures))
return 1
print("All GUI checks passed.")
return 0
if __name__ == "__main__":
sys.exit(main())
tools/test_refactor.py
-358
View File
@@ -1,358 +0,0 @@
#!/usr/bin/env python3
"""Behavioural tests for the sras-viewer refactor.
Covers what the golden-hash harness can't: the v6->v7 cache round-trip
(including block carry-forward), parallel-vs-serial identity, the no-mask
fast path, and the ROI bounding-box mask optimisation.
Usage: python tools/test_refactor.py [--scratch DIR]
"""
import argparse
import shutil
import sys
import tempfile
from pathlib import Path
import numpy as np
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
import sras_compute as compute # noqa: E402
from sras_compute import ( # noqa: E402
cache_file, compute_dc_image, compute_rf_image, dc_image_mv,
)
from sras_format import CH3_IDX, CH4_IDX, SrasFile, adc_to_mv # noqa: E402
import tools.make_test_sras as gen # noqa: E402
_failures: list[str] = []
def check(name: str, ok: bool, detail: str = ""):
print(f" {'PASS' if ok else 'FAIL'} {name}" + (f" — {detail}" if detail else ""))
if not ok:
_failures.append(name)
def test_cache_roundtrip(scratch: Path):
"""v6 -> v7 for DC, then FFT, asserting the first block survives the
second write (the carry-forward path in write_v7_cache)."""
print("\ncache round-trip (v6 -> v7, both blocks)")
path = scratch / "roundtrip.sras"
gen.write(path, n_angles=3, seed=1, samples_per_frame=64)
src = SrasFile(str(path))
check("source is v6", src.version == 6, f"got v{src.version}")
expect_dc3 = [dc_image_mv(src, a, CH3_IDX) for a in range(src.n_angles)]
expect_dc4 = [dc_image_mv(src, a, CH4_IDX) for a in range(src.n_angles)]
expect_fft = [compute_rf_image(src, a, dc_threshold_mv=None, apply_bg_sub=True)
for a in range(src.n_angles)]
err = cache_file(str(path), "dc", True)
check("dc cache_file succeeded", err == "", err)
after_dc = SrasFile(str(path))
check("version flipped to 7", after_dc.version == 7, f"got v{after_dc.version}")
check("dc3 stored for every angle",
all(x is not None for x in after_dc.precomputed_dc3_mv))
check("dc3 values round-trip",
all(np.allclose(after_dc.precomputed_dc3_mv[a], expect_dc3[a], atol=1e-4)
for a in range(after_dc.n_angles)))
check("dc4 values round-trip",
all(np.allclose(after_dc.precomputed_dc4_mv[a], expect_dc4[a], atol=1e-4)
for a in range(after_dc.n_angles)))
check("no fft block yet",
all(x is None for x in after_dc.precomputed_freq_mhz))
check("cached images are native float32",
after_dc.precomputed_dc3_mv[0].dtype == np.float32
and after_dc.precomputed_dc3_mv[0].dtype.byteorder in ("=", "|"),
str(after_dc.precomputed_dc3_mv[0].dtype.byteorder))
check("cached images are writable",
after_dc.precomputed_dc3_mv[0].flags.writeable)
err = cache_file(str(path), "fft", True)
check("fft cache_file succeeded", err == "", err)
both = SrasFile(str(path))
check("fft stored for every angle",
all(x is not None for x in both.precomputed_freq_mhz))
check("fft values round-trip",
all(np.allclose(both.precomputed_freq_mhz[a], expect_fft[a], atol=1e-3)
for a in range(both.n_angles)))
check("DC block carried forward through the FFT write",
all(np.allclose(both.precomputed_dc3_mv[a], expect_dc3[a], atol=1e-4)
for a in range(both.n_angles)))
check("bg_sub flag persisted", both.precomputed_bg_sub is True)
# The fast path must reproduce a fresh compute, and masking must still
# apply on top of a cached (unmasked) image.
fresh = SrasFile(str(path))
fresh.precomputed_freq_mhz = [None] * fresh.n_angles
dc4 = dc_image_mv(both, 0, CH4_IDX)
thr = float(np.median(dc4))
check("cached fast path == fresh compute (unmasked)",
np.allclose(compute_rf_image(both, 0, dc_threshold_mv=None, apply_bg_sub=True),
compute_rf_image(fresh, 0, dc_threshold_mv=None, apply_bg_sub=True),
atol=1e-3))
check("cached fast path == fresh compute (masked)",
np.allclose(compute_rf_image(both, 0, dc_threshold_mv=thr, apply_bg_sub=True),
compute_rf_image(fresh, 0, dc_threshold_mv=thr, apply_bg_sub=True),
atol=1e-3))
# Waveform data must be byte-identical to the pre-cache file.
orig = scratch / "roundtrip_orig.sras"
gen.write(orig, n_angles=3, seed=1, samples_per_frame=64)
o, n = SrasFile(str(orig)), SrasFile(str(path))
check("waveform data untouched by the cache write",
all(np.array_equal(np.asarray(o.data[a]), np.asarray(n.data[a]))
for a in range(o.n_angles)))
def test_partial_v7_cache(scratch: Path):
"""Only some angles cached: uncached angles must compute, not read zeros.
This is the v5 bug the ragged normalisation fixed, checked via v7."""
print("\npartial cache (only some angles stored)")
path = scratch / "partial.sras"
gen.write(path, n_angles=3, seed=2, samples_per_frame=64)
src = SrasFile(str(path))
expected = [compute_rf_image(src, a, dc_threshold_mv=None, apply_bg_sub=True)
for a in range(src.n_angles)]
partial = [expected[0], None, expected[2]] # angle 1 deliberately absent
src.write_v7_cache(new_freq_mhz=partial, new_bg_sub=True)
reread = SrasFile(str(path))
check("angle 1 is not cached", reread.precomputed_freq_mhz[1] is None)
check("angles 0 and 2 are cached",
reread.precomputed_freq_mhz[0] is not None
and reread.precomputed_freq_mhz[2] is not None)
img1 = compute_rf_image(reread, 1, dc_threshold_mv=None, apply_bg_sub=True)
check("uncached angle computes rather than returning zeros",
np.any(img1 != 0) and np.allclose(img1, expected[1], atol=1e-3))
def test_parallel_identity(scratch: Path):
"""Forcing 1 worker vs many must give identical output — catches
chunk-boundary and race bugs."""
print("\nparallel vs serial identity")
path = scratch / "parallel.sras"
# Many rows, so the row loop actually splits into several chunks.
n_rows, n_frames, spf = 48, 9, 256
gen.write(path, n_angles=1, seed=3, samples_per_frame=spf,
geometry=[(n_rows, n_frames)])
sras = SrasFile(str(path))
saved_budget, saved_workers = compute._TOTAL_BYTES_BUDGET, compute._MAX_WORKERS
try:
# Shrink the budget so chunk_rows collapses to 1 and every row is
# its own chunk — the worst case for boundary bugs.
compute._TOTAL_BYTES_BUDGET = 8 * n_frames * spf * 4
compute._MAX_WORKERS = 1
dc_serial = compute_dc_image(sras, 0, CH4_IDX)
rf_serial = compute_rf_image(sras, 0, dc_threshold_mv=None, apply_bg_sub=True)
dc4 = adc_to_mv(dc_serial, *sras.cal(CH4_IDX))
thr = float(np.median(dc4))
rf_masked_serial = compute_rf_image(sras, 0, dc_threshold_mv=thr,
apply_bg_sub=True)
chunk_rows, n_workers = compute._plan_chunks(
n_rows, n_frames, spf, live_multiplier=compute._FFT_LIVE_MULTIPLIER)
check("serial plan uses 1 worker", n_workers == 1, f"chunk_rows={chunk_rows}")
check("work actually splits into multiple chunks", chunk_rows < n_rows,
f"chunk_rows={chunk_rows} of {n_rows} rows")
compute._MAX_WORKERS = 8
chunk_rows, n_workers = compute._plan_chunks(
n_rows, n_frames, spf, live_multiplier=compute._FFT_LIVE_MULTIPLIER)
check("parallel plan uses >1 worker", n_workers > 1,
f"chunk_rows={chunk_rows} workers={n_workers}")
dc_par = compute_dc_image(sras, 0, CH4_IDX)
rf_par = compute_rf_image(sras, 0, dc_threshold_mv=None, apply_bg_sub=True)
rf_masked_par = compute_rf_image(sras, 0, dc_threshold_mv=thr, apply_bg_sub=True)
check("dc image identical", np.array_equal(dc_serial, dc_par))
check("rf image identical (unmasked)", np.array_equal(rf_serial, rf_par))
check("rf image identical (masked)",
np.array_equal(rf_masked_serial, rf_masked_par))
finally:
compute._TOTAL_BYTES_BUDGET, compute._MAX_WORKERS = saved_budget, saved_workers
def test_nomask_equals_low_threshold(scratch: Path):
"""dc_threshold_mv=None must equal a threshold below every pixel, while
skipping the CH4 read."""
print("\nno-mask path")
path = scratch / "nomask.sras"
gen.write(path, n_angles=2, seed=4, samples_per_frame=128)
sras = SrasFile(str(path))
for a in range(sras.n_angles):
none_img = compute_rf_image(sras, a, dc_threshold_mv=None, apply_bg_sub=True)
low_img = compute_rf_image(sras, a, dc_threshold_mv=-1e9, apply_bg_sub=True)
check(f"angle {a}: None == -1e9 threshold",
np.array_equal(none_img, low_img))
check(f"angle {a}: image is non-degenerate",
len(np.unique(none_img)) > 1, f"{len(np.unique(none_img))} unique")
def test_roi_mask():
"""The bbox-restricted mask must equal a full-grid point-in-polygon test."""
print("\nROI mask (bbox fast path vs full grid)")
from matplotlib.path import Path as MplPath
from sras_viewer import RoiQuad
rng = np.random.default_rng(0)
x = np.linspace(-2.0, 3.0, 137)
y = np.linspace(1.0, 4.0, 91)
cases = {
"axis-aligned rect": np.array([[0.0, 1.5], [1.0, 1.5], [1.0, 3.0], [0.0, 3.0]]),
"skewed quad": np.array([[-0.5, 1.2], [1.7, 1.9], [1.2, 3.4], [-1.0, 2.6]]),
"entirely outside": np.array([[8.0, 8.0], [9.0, 8.0], [9.0, 9.0], [8.0, 9.0]]),
"covers whole grid": np.array([[-9.0, -9.0], [9.0, -9.0], [9.0, 9.0], [-9.0, 9.0]]),
"straddles left edge": np.array([[-4.0, 2.0], [0.5, 2.0], [0.5, 3.0], [-4.0, 3.0]]),
}
for _ in range(5):
cases[f"random {_}"] = rng.uniform([-2.5, 0.5], [3.5, 4.5], size=(4, 2))
for name, pts in cases.items():
roi = RoiQuad(pts)
fast = roi.mask_for_grid(x, y)
X, Y = np.meshgrid(x.astype(np.float64), y.astype(np.float64))
slow = MplPath(pts).contains_points(
np.column_stack([X.ravel(), Y.ravel()])).reshape(X.shape)
check(f"{name} ({int(slow.sum())} px inside)", np.array_equal(fast, slow))
# Descending y axis (images are stored top-down in some scans).
roi = RoiQuad(cases["skewed quad"])
y_desc = y[::-1]
fast = roi.mask_for_grid(x, y_desc)
X, Y = np.meshgrid(x.astype(np.float64), y_desc.astype(np.float64))
slow = MplPath(cases["skewed quad"]).contains_points(
np.column_stack([X.ravel(), Y.ravel()])).reshape(X.shape)
check("descending y axis", np.array_equal(fast, slow))
def test_legacy_parse_and_average(scratch: Path):
"""v2-v4 parsing plus the sras_average.py rewrite (which now streams via
SrasFile rather than slurping the whole file)."""
import subprocess
print("\nlegacy formats (v2-v4) and sras_average")
repo = Path(__file__).resolve().parent.parent
for version in (2, 3, 4):
path = scratch / f"legacy_v{version}.sras"
meta = gen.write_legacy(path, version=version, n_angles=2, n_rows=4,
n_frames=12, samples_per_frame=32, seed=version)
s = SrasFile(str(path))
check(f"v{version} parses", s.version == version, f"got v{s.version}")
check(f"v{version} geometry uniform across angles",
list(s.n_rows) == [4, 4] and list(s.n_frames) == [12, 12],
f"rows={list(s.n_rows)} frames={list(s.n_frames)}")
check(f"v{version} waveform data matches what was written",
all(np.array_equal(np.asarray(s.data[a]), meta["data"][a])
for a in range(s.n_angles)))
check(f"v{version} background {'present' if version >= 4 else 'absent'}",
(s.background is not None) == (version >= 4))
check(f"v{version} precomputed stores are ragged lists",
isinstance(s.precomputed_freq_mhz, list)
and len(s.precomputed_freq_mhz) == s.n_angles)
# DC image must equal a direct mean of the known input.
expect = meta["data"][0][:, CH3_IDX, :, :].astype(np.float64).mean(axis=-1)
check(f"v{version} DC image equals a direct mean",
np.allclose(compute_dc_image(s, 0, CH3_IDX), expect, atol=1e-3))
src = scratch / "legacy_v4.sras"
meta = gen.write_legacy(src, version=4, n_angles=2, n_rows=4, n_frames=12,
samples_per_frame=32, seed=4)
dst = scratch / "legacy_v4_avg.sras"
if dst.exists():
dst.unlink()
proc = subprocess.run(
[sys.executable, str(repo / "sras_average.py"), str(src), str(dst), "--n", "4"],
capture_output=True, text=True, cwd=repo)
check("sras_average ran", proc.returncode == 0,
(proc.stderr or proc.stdout).strip()[-200:])
if dst.exists():
avg = SrasFile(str(dst))
check("averaged file parses", avg.version == 4)
check("frame count divided by 4",
list(avg.n_frames) == [3, 3], f"{list(avg.n_frames)}")
check("angles/rows/channels unchanged",
avg.n_angles == 2 and list(avg.n_rows) == [4, 4]
and avg.n_channels == meta["n_channels"])
check("calibration preserved",
np.allclose(avg.ch_ymult_mv, SrasFile(str(src)).ch_ymult_mv))
check("background preserved",
np.array_equal(avg.background, SrasFile(str(src)).background))
src_data = meta["data"]
expect0 = src_data[0][:, :, 0:4, :].astype(np.float32).mean(axis=2).astype(np.int16)
check("first averaged group equals the mean of its 4 source frames",
np.array_equal(np.asarray(avg.data[0])[:, :, 0, :], expect0))
# Remainder handling: 12 frames / 5 -> 2 full groups + 1 partial.
dst2 = scratch / "legacy_v4_avg5.sras"
subprocess.run([sys.executable, str(repo / "sras_average.py"),
str(src), str(dst2), "--n", "5"],
capture_output=True, text=True, cwd=repo)
if dst2.exists():
check("partial trailing group kept by default",
list(SrasFile(str(dst2)).n_frames) == [3, 3],
f"{list(SrasFile(str(dst2)).n_frames)}")
dst3 = scratch / "legacy_v4_avg5d.sras"
subprocess.run([sys.executable, str(repo / "sras_average.py"),
str(src), str(dst3), "--n", "5", "--discard-remainder"],
capture_output=True, text=True, cwd=repo)
if dst3.exists():
check("--discard-remainder drops the partial group",
list(SrasFile(str(dst3)).n_frames) == [2, 2],
f"{list(SrasFile(str(dst3)).n_frames)}")
def test_unsupported_version_reported(scratch: Path):
"""cache_file must report, not raise, for a file it can't handle."""
print("\nerror reporting")
bogus = scratch / "bogus.sras"
bogus.write_bytes(b"SRAS" + bytes([99]) + b"\x00" * 200)
err = cache_file(str(bogus), "dc", True)
check("bad version returns an error string", bool(err), err)
missing = cache_file(str(scratch / "does_not_exist.sras"), "dc", True)
check("missing file returns an error string", bool(missing), missing)
def main():
p = argparse.ArgumentParser(description=__doc__)
p.add_argument("--scratch")
args = p.parse_args()
tmp = None
if args.scratch:
scratch = Path(args.scratch)
scratch.mkdir(parents=True, exist_ok=True)
else:
tmp = tempfile.mkdtemp(prefix="sras_test_")
scratch = Path(tmp)
try:
test_cache_roundtrip(scratch)
test_partial_v7_cache(scratch)
test_parallel_identity(scratch)
test_nomask_equals_low_threshold(scratch)
test_roi_mask()
test_legacy_parse_and_average(scratch)
test_unsupported_version_reported(scratch)
finally:
if tmp:
shutil.rmtree(tmp, ignore_errors=True)
print()
if _failures:
print(f"{len(_failures)} FAILURE(S): " + ", ".join(_failures))
sys.exit(1)
print("All checks passed.")
if __name__ == "__main__":
main()