diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..ca708ca --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,35 @@ +[build-system] +requires = ["setuptools>=68"] +build-backend = "setuptools.build_meta" + +[project] +name = "sras-viewer" +version = "0.1.0" +description = "Viewer and processing tools for SRAS .sras scan files" +requires-python = ">=3.12" +dependencies = [ + "PyQt6==6.10.2", + "numpy==2.4.1", + "matplotlib==3.10.8", + "scipy==1.18.0", + # Angle alignment only: masked FFT phase correlation (skimage.registration). + "scikit-image==0.26.0", + # Faster rfft backend; the viewer falls back to scipy.fft without it. + "pyFFTW==0.15.1", +] + +[project.optional-dependencies] +dev = ["pytest"] + +[tool.setuptools] +py-modules = [ + "sras_format", + "sras_compute", + "sras_workers", + "sras_viewer", + "sras_average", + "sras_edit_scans", +] + +[tool.pytest.ini_options] +testpaths = ["tests"] diff --git a/sras_compute.py b/sras_compute.py index 28c8b3d..bf48c6b 100644 --- a/sras_compute.py +++ b/sras_compute.py @@ -1188,10 +1188,6 @@ def compute_angle_alignment(sras: SrasFile, ref_angle_idx: int, return result -# Back-compat alias for the pre-split private name (used by tooling). -_compute_angle_alignment = compute_angle_alignment - - # --------------------------------------------------------------------------- # Manual alignment (Fusion menu -> Manual Alignment... dialog) # diff --git a/sras_viewer_requirements.txt b/sras_viewer_requirements.txt deleted file mode 100644 index 1448342..0000000 --- a/sras_viewer_requirements.txt +++ /dev/null @@ -1,11 +0,0 @@ -PyQt6==6.10.2 -numpy==2.4.1 -matplotlib==3.10.8 -scipy==1.18.0 -# Angle alignment only: masked FFT phase correlation, which registers scans -# whose valid (scanned) regions differ in shape — see sras_compute's -# _masked_shift. -scikit-image==0.26.0 -# Optional: a faster rfft backend for the RF/FFT images (FFT Options -> pyFFTW). -# The viewer falls back to scipy.fft when it is not installed. -pyFFTW==0.15.1 diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..c9df201 --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,8 @@ +"""Shared test setup: repo-root imports and the offscreen Qt platform.""" + +import os +import sys +from pathlib import Path + +os.environ.setdefault("QT_QPA_PLATFORM", "offscreen") +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) diff --git a/tests/test_alignment.py b/tests/test_alignment.py new file mode 100644 index 0000000..18be785 --- /dev/null +++ b/tests/test_alignment.py @@ -0,0 +1,212 @@ +"""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 tests/test_gui.py for the +dialog and Aligned-View plumbing. +""" + +from types import SimpleNamespace + +import numpy as np +import pytest + +import sras_compute as compute +from sras_format import CH4_IDX, SrasFile, adc_to_mv +import tools.make_test_sras as gen + +# 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 + + +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) + + +@pytest.fixture(scope="module") +def rig(tmp_path_factory): + """The rotating-sample scan plus everything computed from it once.""" + tmpdir = tmp_path_factory.mktemp("sras_align") + path = tmpdir / "rotating.sras" + meta = gen.write_rotating(path, n_angles=5) + sras = SrasFile(str(path)) + 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)} + result = compute.compute_angle_alignment(sras, 0, _THRESHOLD_MV) + return SimpleNamespace(path=path, sras=sras, truth=meta["truth"], + dc4=dc4, fits=fits, result=result) + + +def test_registration_recovers_truth(rig): + """Per-angle rigid registration (rotation + translation, no scale).""" + for a, fit in rig.fits.items(): + t_rot, t_shift = rig.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])) + assert rot_err <= _ROT_TOL_DEG, \ + (f"angle {a}: got {fit.rotation_deg:.3f}°, truth {t_rot:.3f}° " + f"(err {rot_err:.3f}°)") + assert shift_err <= _SHIFT_TOL_MM, f"angle {a}: err {shift_err:.4f} mm" + assert rig.fits[0] == compute.RigidFit(0.0, (0.0, 0.0), 1.0, "reference"), \ + "reference angle registers as exact identity" + + +def test_stage_angle_sign_is_not_trusted(rig): + # 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(rig.path)) + flipped.angles_deg = -flipped.angles_deg + flipped_fits = {a: compute.register_angle_to_reference( + flipped, a, 0, rig.dc4, dc_threshold_mv=_THRESHOLD_MV) + for a in range(1, flipped.n_angles)} + mismatches = {a: (flipped_fits[a].rotation_deg, rig.fits[a].rotation_deg) + for a in flipped_fits if flipped_fits[a] != rig.fits[a]} + assert not mismatches, \ + f"negating every reported stage angle changed fits: {mismatches}" + + +def test_stage_coordinates_are_not_consulted(rig): + # 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(rig.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)} + mismatches = {a: (round(moved_fits[a].rotation_deg, 4), rig.fits[a].rotation_deg) + for a in moved_fits if moved_fits[a] != rig.fits[a]} + assert not mismatches, \ + f"relocating every other angle's scan window changed fits: {mismatches}" + + +def test_canvas_is_reference_grid_extended(rig): + sras, result = rig.sras, rig.result + t0 = result.per_angle[0] + assert np.allclose(t0.matrix, np.eye(2)), \ + f"angle 0's transform has rotation/scale/shear: {t0.matrix}" + assert np.allclose(t0.offset, np.round(t0.offset)), \ + f"angle 0 does not land on whole canvas pixels: {t0.offset}" + assert ((result.canvas_dx_mm, result.canvas_dy_mm) + == compute._pixel_pitch_mm(sras, 0)), \ + "canvas pitch is angle 0's own pitch" + + 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) + assert np.allclose(x_axis[col0:col0 + a0_cols], sras.x_axis_mm(0)), \ + "canvas X axis reproduces angle 0's own X coordinates" + assert np.allclose(y_axis[row0:row0 + a0_rows], sras.y_positions_mm(0)), \ + "canvas Y axis reproduces angle 0's own Y coordinates" + assert (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))), \ + f"canvas does not cover every angle's footprint: {result.canvas_shape}" + + +def test_transforms_are_pure_rotations(rig): + """No scaling anywhere in the per-angle transforms.""" + for a in range(rig.sras.n_angles): + R = mm_transform(rig.sras, rig.result, a) + assert (np.allclose(R @ R.T, np.eye(2), atol=1e-9) + and abs(abs(np.linalg.det(R)) - 1.0) < 1e-9), \ + f"angle {a}: det={np.linalg.det(R):.6f}" + + +def test_all_angles_stack(rig): + aligned = {a: compute.apply_alignment(rig.result, a, rig.dc4[a]) + for a in range(rig.sras.n_angles)} + base = aligned[0] >= _THRESHOLD_MV + for a in range(1, rig.sras.n_angles): + other = aligned[a] >= _THRESHOLD_MV + iou = float((base & other).sum()) / max(1, int((base | other).sum())) + assert iou >= _STACK_IOU_MIN, f"angle {a}: IoU {iou:.4f}" + + +def test_downsampled_preview_lands_with_full_res(rig): + # 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. + sras, result = rig.sras, rig.result + pitch = (result.canvas_dx_mm, result.canvas_dy_mm) + a = sras.n_angles - 1 + p = result.per_angle[a] + full_mask = (rig.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) + assert (abs(d[0]) <= abs(pitch[0] * fx) and abs(d[1]) <= abs(pitch[1] * fy)), \ + f"downsampled preview offset {d[0]:+.4f}, {d[1]:+.4f} mm" + + +def test_manual_path_reproduces_geometry(rig): + sras, result = rig.sras, rig.result + 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) + assert (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))), \ + "build_manual_alignment matches compute_angle_alignment for the same params" + + +def test_sidecar_roundtrip(rig): + sras, result = rig.sras, rig.result + params = {a: compute.ManualAngleParams(t.rotation_deg, t.shift_mm) + for a, t in result.per_angle.items()} + compute.save_manual_alignment(sras, 0, _THRESHOLD_MV, params) + loaded = compute.load_manual_alignment(sras) + assert (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))), \ + "sidecar reloads every angle's params" + assert compute.delete_manual_alignment(sras), "sidecar deletes cleanly" diff --git a/tests/test_compute.py b/tests/test_compute.py new file mode 100644 index 0000000..e08af39 --- /dev/null +++ b/tests/test_compute.py @@ -0,0 +1,277 @@ +"""Behavioural tests for the compute/format layer. + +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. +""" + +import subprocess +import sys +from pathlib import Path + +import numpy as np + +import sras_compute as compute +from sras_compute import ( + cache_file, compute_dc_image, compute_rf_image, dc_image_mv, +) +from sras_format import CH3_IDX, CH4_IDX, SrasFile, adc_to_mv +import tools.make_test_sras as gen + +REPO = Path(__file__).resolve().parent.parent + + +def test_cache_roundtrip(tmp_path): + """v6 -> v7 for DC, then FFT, asserting the first block survives the + second write (the carry-forward path in write_v7_cache).""" + path = tmp_path / "roundtrip.sras" + gen.write(path, n_angles=3, seed=1, samples_per_frame=64) + + src = SrasFile(str(path)) + assert 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) + assert err == "", err + + after_dc = SrasFile(str(path)) + assert after_dc.version == 7, f"got v{after_dc.version}" + assert all(x is not None for x in after_dc.precomputed_dc3_mv) + assert all(np.allclose(after_dc.precomputed_dc3_mv[a], expect_dc3[a], atol=1e-4) + for a in range(after_dc.n_angles)) + assert all(np.allclose(after_dc.precomputed_dc4_mv[a], expect_dc4[a], atol=1e-4) + for a in range(after_dc.n_angles)) + assert all(x is None for x in after_dc.precomputed_freq_mhz), "no fft block yet" + assert (after_dc.precomputed_dc3_mv[0].dtype == np.float32 + and after_dc.precomputed_dc3_mv[0].dtype.byteorder in ("=", "|")), \ + "cached images are native float32" + assert after_dc.precomputed_dc3_mv[0].flags.writeable + + err = cache_file(str(path), "fft", True) + assert err == "", err + + both = SrasFile(str(path)) + assert all(x is not None for x in both.precomputed_freq_mhz) + assert all(np.allclose(both.precomputed_freq_mhz[a], expect_fft[a], atol=1e-3) + for a in range(both.n_angles)) + assert all(np.allclose(both.precomputed_dc3_mv[a], expect_dc3[a], atol=1e-4) + for a in range(both.n_angles)), \ + "DC block carried forward through the FFT write" + assert 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)) + assert 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), "cached fast path == fresh compute (unmasked)" + assert 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), "cached fast path == fresh compute (masked)" + + # Waveform data must be byte-identical to the pre-cache file. + orig = tmp_path / "roundtrip_orig.sras" + gen.write(orig, n_angles=3, seed=1, samples_per_frame=64) + o, n = SrasFile(str(orig)), SrasFile(str(path)) + assert all(np.array_equal(np.asarray(o.data[a]), np.asarray(n.data[a])) + for a in range(o.n_angles)), \ + "waveform data untouched by the cache write" + + +def test_partial_v7_cache(tmp_path): + """Only some angles cached: uncached angles must compute, not read zeros. + This is the v5 bug the ragged normalisation fixed, checked via v7.""" + path = tmp_path / "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)) + assert reread.precomputed_freq_mhz[1] is None + assert (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) + assert np.any(img1 != 0) and np.allclose(img1, expected[1], atol=1e-3), \ + "uncached angle computes rather than returning zeros" + + +def test_parallel_identity(tmp_path, monkeypatch): + """Forcing 1 worker vs many must give identical output — catches + chunk-boundary and race bugs.""" + path = tmp_path / "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)) + + # Shrink the budget so chunk_rows collapses to 1 and every row is + # its own chunk — the worst case for boundary bugs. + monkeypatch.setattr(compute, "_TOTAL_BYTES_BUDGET", 8 * n_frames * spf * 4) + + monkeypatch.setattr(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) + assert n_workers == 1, f"serial plan uses 1 worker (chunk_rows={chunk_rows})" + assert chunk_rows < n_rows, \ + f"work actually splits into multiple chunks ({chunk_rows} of {n_rows} rows)" + + monkeypatch.setattr(compute, "_MAX_WORKERS", 8) + chunk_rows, n_workers = compute._plan_chunks( + n_rows, n_frames, spf, live_multiplier=compute._FFT_LIVE_MULTIPLIER) + assert n_workers > 1, \ + f"parallel plan uses >1 worker (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) + + assert np.array_equal(dc_serial, dc_par), "dc image identical" + assert np.array_equal(rf_serial, rf_par), "rf image identical (unmasked)" + assert np.array_equal(rf_masked_serial, rf_masked_par), \ + "rf image identical (masked)" + + +def test_nomask_equals_low_threshold(tmp_path): + """dc_threshold_mv=None must equal a threshold below every pixel, while + skipping the CH4 read.""" + path = tmp_path / "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) + assert np.array_equal(none_img, low_img), \ + f"angle {a}: None == -1e9 threshold" + assert len(np.unique(none_img)) > 1, \ + f"angle {a}: image is degenerate ({len(np.unique(none_img))} unique)" + + +def test_roi_mask(): + """The bbox-restricted mask must equal a full-grid point-in-polygon test.""" + 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) + assert np.array_equal(fast, slow), f"{name} ({int(slow.sum())} px inside)" + + # 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) + assert np.array_equal(fast, slow), "descending y axis" + + +def test_legacy_parse(tmp_path): + """v2-v4 parsing against known written data.""" + for version in (2, 3, 4): + path = tmp_path / 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)) + assert s.version == version, f"got v{s.version}" + assert list(s.n_rows) == [4, 4] and list(s.n_frames) == [12, 12], \ + f"rows={list(s.n_rows)} frames={list(s.n_frames)}" + assert all(np.array_equal(np.asarray(s.data[a]), meta["data"][a]) + for a in range(s.n_angles)), \ + f"v{version} waveform data matches what was written" + assert (s.background is not None) == (version >= 4), \ + f"v{version} background {'present' if version >= 4 else 'absent'}" + assert (isinstance(s.precomputed_freq_mhz, list) + and len(s.precomputed_freq_mhz) == s.n_angles), \ + f"v{version} precomputed stores are ragged lists" + # DC image must equal a direct mean of the known input. + expect = meta["data"][0][:, CH3_IDX, :, :].astype(np.float64).mean(axis=-1) + assert np.allclose(compute_dc_image(s, 0, CH3_IDX), expect, atol=1e-3), \ + f"v{version} DC image equals a direct mean" + + +def test_sras_average(tmp_path): + """The sras_average.py CLI: frame averaging with remainder handling.""" + src = tmp_path / "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 = tmp_path / "legacy_v4_avg.sras" + proc = subprocess.run( + [sys.executable, str(REPO / "sras_average.py"), str(src), str(dst), "--n", "4"], + capture_output=True, text=True, cwd=REPO) + assert proc.returncode == 0, (proc.stderr or proc.stdout).strip()[-200:] + + avg = SrasFile(str(dst)) + assert avg.version == 4 + assert list(avg.n_frames) == [3, 3], f"{list(avg.n_frames)}" + assert (avg.n_angles == 2 and list(avg.n_rows) == [4, 4] + and avg.n_channels == meta["n_channels"]) + assert np.allclose(avg.ch_ymult_mv, SrasFile(str(src)).ch_ymult_mv), \ + "calibration preserved" + assert np.array_equal(avg.background, SrasFile(str(src)).background), \ + "background preserved" + src_data = meta["data"] + expect0 = src_data[0][:, :, 0:4, :].astype(np.float32).mean(axis=2).astype(np.int16) + assert np.array_equal(np.asarray(avg.data[0])[:, :, 0, :], expect0), \ + "first averaged group equals the mean of its 4 source frames" + + # Remainder handling: 12 frames / 5 -> 2 full groups + 1 partial. + dst2 = tmp_path / "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) + assert list(SrasFile(str(dst2)).n_frames) == [3, 3], \ + "partial trailing group kept by default" + dst3 = tmp_path / "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) + assert list(SrasFile(str(dst3)).n_frames) == [2, 2], \ + "--discard-remainder drops the partial group" + + +def test_unsupported_version_reported(tmp_path): + """cache_file must report, not raise, for a file it can't handle.""" + bogus = tmp_path / "bogus.sras" + bogus.write_bytes(b"SRAS" + bytes([99]) + b"\x00" * 200) + err = cache_file(str(bogus), "dc", True) + assert err, "bad version returns an error string" + missing = cache_file(str(tmp_path / "does_not_exist.sras"), "dc", True) + assert missing, "missing file returns an error string" diff --git a/tests/test_gui.py b/tests/test_gui.py new file mode 100644 index 0000000..7b1ebec --- /dev/null +++ b/tests/test_gui.py @@ -0,0 +1,501 @@ +"""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. + +NOTE: this module is one ordered integration sequence over a single shared +window — the tests build on each other's state and must run in definition +order (pytest's default within a module). Run the whole module, not single +tests. +""" + +import json +from types import SimpleNamespace +from unittest.mock import patch + +import numpy as np +import pytest +from PyQt6.QtCore import QEventLoop, Qt, QTimer +from PyQt6.QtTest import QTest +from PyQt6.QtWidgets import QApplication, QMessageBox + +import sras_compute as compute +from sras_format import CH1_IDX, CH3_IDX, CH4_IDX, SrasFile +from sras_viewer import RoiQuad, SrasViewerWindow, VELOCITY_MODE_IDX +import tools.make_test_sras as gen + + +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() + + +@pytest.fixture(scope="module") +def ctx(tmp_path_factory): + """The shared window, test file, and cross-test state for the sequence.""" + app = QApplication.instance() or QApplication([]) + tmpdir = tmp_path_factory.mktemp("sras_gui") + path = tmpdir / "gui.sras" + gen.write(path, n_angles=4, seed=11, samples_per_frame=256) + + win = SrasViewerWindow() + win.show() + errors: list[str] = [] + # 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) + + c = SimpleNamespace(app=app, win=win, path=path, tmpdir=tmpdir, + errors=errors, s=None) + yield c + if win.isVisible(): + win.close() + pump(400) + + +def test_load(ctx): + win = ctx.win + win._load_file(str(ctx.path)) + assert wait_until(lambda: win._sras is not None), "file loaded" + ctx.s = s = win._sras + assert s.version == 6, f"v{s.version}" + assert win.combo_channel.currentIndex() == CH4_IDX, "defaults to CH4" + assert win._current_image is not None, "image displayed" + assert win.spin_angle.maximum() == s.n_angles - 1, \ + "angle spinbox ranges over all angles" + assert win._info["Angles"].text() == f"Angles: {s.n_angles}", \ + win._info["Angles"].text() + + +def test_dc_precompute_all_angles(ctx): + win, s = ctx.win, ctx.s + 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))) + assert ok, f"every angle cached for CH3 and CH4 ({len(win._dc_cache)} entries)" + assert "ready for all angles" in win.lbl_dc_precompute.text(), \ + win.lbl_dc_precompute.text() + + +def test_angle_switching_from_cache(ctx): + win, s = ctx.win, ctx.s + for a in range(s.n_angles): + win.spin_angle.setValue(a) + win._on_view_changed() + pump(60) + expected = win._sras.image_shape(a) + assert win._current_image.shape == expected, \ + f"angle {a} shows its own geometry {expected}, got {win._current_image.shape}" + assert not win._job_running("compute"), \ + "no compute job needed for cached DC angles" + + +def test_channel_switching(ctx): + win = ctx.win + win.spin_angle.setValue(0) + win._on_view_changed() + pump(60) + win.combo_channel.setCurrentIndex(CH3_IDX) + assert wait_until(lambda: win._current_ch == CH3_IDX), "CH3 displayed" + + win.combo_channel.setCurrentIndex(CH1_IDX) + assert wait_until( + lambda: win._current_ch == CH1_IDX and not win._job_running("compute")), \ + "CH1 (FFT) computed" + assert len(win._fft_cache) > 0, "FFT result cached" + ctx.rf_img = win._current_image + assert len(np.unique(ctx.rf_img)) > 1, \ + f"FFT image is degenerate ({len(np.unique(ctx.rf_img))} unique values)" + + +def test_velocity_mode(ctx): + """Velocity mode is a pure post-multiply, no recompute.""" + win = ctx.win + ctx.n_fft_before = len(win._fft_cache) + win.combo_channel.setCurrentIndex(VELOCITY_MODE_IDX) + assert wait_until( + lambda: win._current_ch == VELOCITY_MODE_IDX + and not win._job_running("compute")), "velocity displayed" + grating = win.spin_grating_um.value() + assert np.allclose(win._current_image, ctx.rf_img * grating, atol=1e-3), \ + "velocity == freq x grating" + assert len(win._fft_cache) == ctx.n_fft_before, \ + f"velocity reused the cached FFT ({ctx.n_fft_before} -> {len(win._fft_cache)})" + assert win.grp_velocity.isVisible(), "grating spinbox visible in velocity mode" + + +def test_threshold_change_recomputes(ctx): + """A threshold change is a genuine cache-key change.""" + win = ctx.win + 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() + assert wait_until( + lambda: not win._job_running("compute") + and len(win._fft_cache) > ctx.n_fft_before), "recomputed at new threshold" + n_zero = int((win._current_image == 0).sum()) + assert n_zero > 0, \ + f"masking zeroed some pixels ({n_zero} of {win._current_image.size})" + + +def test_bg_sub_toggle(ctx): + win = ctx.win + n_before = len(win._fft_cache) + win.chk_bg_sub.setChecked(False) + assert wait_until( + lambda: not win._job_running("compute") and len(win._fft_cache) > n_before), \ + "recomputed without bg-sub" + win.chk_bg_sub.setChecked(True) + pump(200) + assert not win._job_running("compute"), \ + "returning to bg-sub was a cache hit (no recompute)" + + +def test_roi_and_csv_export(ctx): + win, s = ctx.win, ctx.s + 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) + assert win.image_canvas.get_roi() is not None, "ROI registered" + assert ("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]) + assert 0 < npix <= win._current_image.size, f"{npix}" + assert win.btn_export_roi.isEnabled(), "Export ROI enabled" + + csv_path = ctx.tmpdir / "roi.csv" + with patch("sras_viewer.QFileDialog.getSaveFileName", + return_value=(str(csv_path), "")): + win._on_export_roi_csv() + assert csv_path.exists(), "ROI CSV written" + body = [l for l in csv_path.read_text().splitlines() if not l.startswith("#")] + assert len(body) == npix + 1, \ + f"ROI CSV has {len(body)} lines for {npix} pixels (want header + one per pixel)" + + img_csv = ctx.tmpdir / "img.csv" + with patch("sras_viewer.QFileDialog.getSaveFileName", + return_value=(str(img_csv), "")): + win._on_export_csv() + assert img_csv.exists(), "image CSV written" + arr = np.loadtxt(img_csv, delimiter=",") + assert (arr.shape == win._current_image.shape + and np.allclose(arr, win._current_image, rtol=1e-5, atol=1e-4)), \ + "image CSV round-trips the displayed image" + + +def test_roi_survives_switches(ctx): + win = ctx.win + win.spin_angle.setValue(1) + win._on_view_changed() + wait_until(lambda: not win._job_running("compute")) + assert win.image_canvas.get_roi() is not None, \ + "ROI still present after angle switch" + win.combo_channel.setCurrentIndex(CH4_IDX) + wait_until(lambda: win._current_ch == CH4_IDX) + assert win.image_canvas.get_roi() is not None, \ + "ROI still present after channel switch" + + +def test_angle_alignment(ctx): + win, s = ctx.win, ctx.s + win.spin_angle.setValue(0) + win._on_view_changed() + wait_until(lambda: not win._job_running("compute")) + assert win._alignment_act.isEnabled(), "alignment action enabled" + win._on_angle_alignment() + assert wait_until( + lambda: win._alignment_result is not None and not win._job_running("align"), + timeout_ms=60000), "alignment completed" + + r = win._alignment_result + assert len(r.per_angle) == s.n_angles, "transform for every angle" + assert 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)), \ + f"canvas is at least as large as any single angle: {r.canvas_shape}" + assert r.per_angle[r.ref_angle_idx].shift_mm == (0.0, 0.0), \ + "reference angle has zero shift" + assert win.chk_aligned_view.isEnabled() and win.chk_aligned_view.isChecked(), \ + "Aligned View auto-enabled and checked" + pump(200) + assert 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) + assert win.image_canvas._img_shape == s.image_shape(0), \ + f"unchecking returns to the raw per-angle grid: {win.image_canvas._img_shape}" + + +def test_manual_alignment_geometry(ctx): + """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 + tests/test_alignment.py, which has a synthetic sample to register.)""" + win, s = ctx.win, ctx.s + assert win._manual_align_act.isEnabled(), "manual alignment action enabled" + + n_rows, n_frames = s.image_shape(0) + assert np.allclose(compute._center_idx(s, 0), + [(n_rows - 1) / 2, (n_frames - 1) / 2]), \ + "array center is the geometric center of the pixel grid" + dx0, dy0 = compute._pixel_pitch_mm(s, 0) + assert np.allclose(compute._local_half_extent_mm(s, 0), + [(n_frames - 1) / 2 * abs(dx0), (n_rows - 1) / 2 * abs(dy0)]), \ + "local half-extent is derived from shape and pitch alone" + 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(ctx.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) + assert shape_a == shape_b and np.allclose(origin_a, origin_b), \ + ("moving every non-reference angle's scan window must leave the canvas " + f"unchanged: {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) + assert 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 + assert compute._shift_into(arr, -1, -1).sum() == 0.0, \ + "_shift_into zero-fills rather than wrapping" + assert compute._shift_into(arr, 2, 3)[2, 3] == 1.0, \ + "_shift_into moves content by exactly the requested offset" + + +def test_manual_dialog_opens_at_identity(ctx): + """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, s = ctx.win, ctx.s + win._on_manual_alignment() + assert win._manual_align_dialog is not None, "dialog opened" + ctx.dlg = dlg = win._manual_align_dialog + assert not win._job_running("manual_align_masks"), \ + "mask prep needed no background worker (already DC-cached)" + assert all(dlg._angle_params[a] == compute.ManualAngleParams() + for a in range(s.n_angles)), \ + "no manual sidecar yet -> dialog starts at identity, not the automatic result" + + +def test_reference_angle_is_locked(ctx): + dlg = ctx.dlg + 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) + assert not dlg.grp_manual_adjust.isEnabled(), "reference angle group disabled" + assert dlg._angle_params[dlg._ref_angle_idx] == before_ref, \ + "reference angle untouched by nudge attempts" + + +def test_nudges(ctx): + """Nudging a real angle (fine + coarse, translate + rotate).""" + dlg, s = ctx.dlg, ctx.s + ctx.active = 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() + assert abs(dlg._angle_params[active].shift_mm[0] - (before[0] + fine_step)) < 1e-9, \ + "fine translate nudge moved shift_x by exactly one fine step" + + 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() + assert abs(dlg._angle_params[active].shift_mm[1] - (before[1] - coarse_step)) < 1e-9, \ + "coarse translate nudge uses the multiplier" + + before_rot = dlg._angle_params[active].rotation_deg + dlg._on_nudge_rotate(1, False) + assert dlg._angle_params[active].rotation_deg != before_rot, \ + "rotate nudge changed rotation_deg" + assert len(dlg._preview_layers) == s.n_angles, \ + "preview canvas rebuilt for every angle after a rotation nudge" + + # Real key-event wiring (proves keyPressEvent -> signal -> slot). + before = dlg._angle_params[active].shift_mm + QTest.keyClick(dlg.canvas, Qt.Key.Key_Right) + assert dlg._angle_params[active].shift_mm[0] > before[0], \ + "a real Right-arrow key event nudged shift_x" + + +def test_auto_derotate(ctx): + """Auto De-rotate: seeds rotation from the stage angle, no translation.""" + dlg, s, active = ctx.dlg, ctx.s, ctx.active + shift_before_derotate = dlg._angle_params[active].shift_mm + dlg._on_auto_derotate() + nominal = compute._nominal_delta_deg(s, active, dlg._ref_angle_idx) + assert abs(dlg._angle_params[active].rotation_deg - nominal) < 1e-6, \ + "auto de-rotate seeded rotation from the stage's reported angle" + assert dlg._angle_params[active].shift_mm == shift_before_derotate, \ + "auto de-rotate left translation untouched" + assert dlg._angle_params[dlg._ref_angle_idx].rotation_deg == 0.0, \ + "reference angle stays identity after auto de-rotate" + # Clicking again offers the other sign, since which one lines the scans up + # is not knowable from the file. + dlg._on_auto_derotate() + assert abs(dlg._angle_params[active].rotation_deg + nominal) < 1e-6, \ + "auto de-rotate offers the opposite sign on a second click" + + +def test_auto_cross_correlate(ctx): + """Auto Cross-Correlate: searches rotation *and* translation.""" + win, dlg, s = ctx.win, ctx.dlg, ctx.s + assert dlg.btn_auto_correlate.isEnabled(), \ + "cross-correlate action enabled once masks are ready" + for label_idx, (label, _sources) in enumerate(dlg._CORRELATE_SOURCES): + dlg.combo_correlate_source.setCurrentIndex(label_idx) + dlg._on_auto_correlate() + assert wait_until( + lambda: not win._job_running("manual_align_correlate"), + timeout_ms=60000), f"auto cross-correlate completed ({label})" + assert all(a in dlg._fit_notes for a in range(s.n_angles) + if a != dlg._ref_angle_idx), \ + f"every non-reference angle got a fit ({label})" + assert dlg._angle_params[dlg._ref_angle_idx] == compute.ManualAngleParams(), \ + "auto cross-correlate reference angle stays identity" + assert dlg.grp_correlate.isEnabled() and dlg.btn_save.isEnabled(), \ + "auto cross-correlate re-enabled controls when done" + assert len(dlg._preview_layers) == s.n_angles, \ + "preview canvas rebuilt after cross-correlate" + assert dlg._fit_report(), "fit quality is reported per angle" + + +def test_save_sidecar(ctx): + win, dlg, s, active = ctx.win, ctx.dlg, ctx.s, ctx.active + dlg._on_save() + sidecar = compute.sidecar_path(s.path) + assert sidecar.exists(), "sidecar file written" + ctx.sidecar = sidecar + ctx.sidecar_raw = raw = json.loads(sidecar.read_text()) + assert raw.get("schema_version") == compute._SIDECAR_SCHEMA_VERSION, \ + "sidecar schema_version is current" + assert all(raw.get("per_angle", {}).get(str(a), {}).get("rotation_deg") + == dlg._angle_params[a].rotation_deg for a in range(s.n_angles)), \ + "sidecar per_angle round-trips the dialog's resolved params" + assert (win._alignment_result is not None + and win._alignment_result.per_angle[active].rotation_deg + == dlg._angle_params[active].rotation_deg), \ + "main window's alignment_result replaced by the manual build" + assert win.chk_aligned_view.isEnabled() and win.chk_aligned_view.isChecked(), \ + "Aligned View auto-enabled after Save" + + +def test_stale_schema_sidecar_ignored(ctx): + """An old-schema sidecar (pre-pivot/sign fix) is treated as absent.""" + s, raw, sidecar = ctx.s, ctx.sidecar_raw, ctx.sidecar + stale = dict(raw) + stale["schema_version"] = compute._SIDECAR_SCHEMA_VERSION - 1 + sidecar.write_text(json.dumps(stale)) + assert compute.load_manual_alignment(s) is None, \ + "a sidecar with an old schema_version is not loaded" + sidecar.write_text(json.dumps(raw)) # restore for the rest of the sequence + + +def test_clear_with_confirmation(ctx): + win, dlg, s = ctx.win, ctx.dlg, ctx.s + with patch("sras_viewer.QMessageBox.question", + return_value=QMessageBox.StandardButton.Yes): + dlg._on_clear() + assert not ctx.sidecar.exists(), "sidecar file deleted" + assert all(dlg._angle_params[a] == compute.ManualAngleParams() + for a in range(s.n_angles)), "dialog params reset to identity" + assert win._alignment_result is None, "main window alignment_result cleared" + assert (not win.chk_aligned_view.isEnabled() + and not win.chk_aligned_view.isChecked()), \ + "Aligned View disabled after Clear" + + dlg.close() + pump(150) + assert win._manual_align_dialog is None, "dialog reference released on close" + + +def test_sidecar_restored_on_reload(ctx): + win, active = ctx.win, ctx.active + 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(ctx.path)) # reload the same file fresh + assert wait_until( + lambda: win._sras is not None and id(win._sras) != old_sras_id), \ + "file reloaded" + ctx.s = win._sras + assert win._manual_align_dialog is None, \ + "manual dialog force-closed by a reload" + assert win._alignment_result is not None, \ + "reload restores the saved manual alignment automatically" + assert abs(win._alignment_result.per_angle[active].rotation_deg + - saved_rotation) < 1e-9, "restored rotation matches what was saved" + assert win._alignment_result.per_angle[active].shift_mm == saved_shift, \ + "restored shift matches what was saved" + assert win.chk_aligned_view.isChecked(), \ + "Aligned View auto-checked after restoring a saved alignment" + + +def test_pixel_inspector(ctx): + win = ctx.win + win.chk_aligned_view.setChecked(False) + pump(100) + win._on_pixel_clicked(0, 0) + pump(150) + assert win.lbl_wave_hint.isHidden(), "waveform hint hidden after a click" + win.combo_channel.setCurrentIndex(CH1_IDX) + wait_until(lambda: not win._job_running("compute")) + win._on_pixel_clicked(1, 1) + pump(150) + assert len(win.wave_canvas.ax_wave.lines) > 0, \ + f"RF waveform panel rendered ({len(win.wave_canvas.ax_wave.lines)} lines)" + + +def test_shutdown(ctx): + win = ctx.win + win.close() + pump(400) + assert len(win._jobs) == 0, f"all background jobs released: {list(win._jobs)}" + + +def test_no_status_bar_errors(ctx): + unexpected = [e for e in ctx.errors if e] + assert not unexpected, f"status-bar errors seen: {unexpected}" diff --git a/tools/check_equivalence.py b/tools/check_equivalence.py index fa11968..37934dc 100644 --- a/tools/check_equivalence.py +++ b/tools/check_equivalence.py @@ -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(): diff --git a/tools/make_test_sras.py b/tools/make_test_sras.py index 71e93ef..92b0b54 100644 --- a/tools/make_test_sras.py +++ b/tools/make_test_sras.py @@ -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 diff --git a/tools/test_alignment.py b/tools/test_alignment.py deleted file mode 100644 index f78f9fa..0000000 --- a/tools/test_alignment.py +++ /dev/null @@ -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()) diff --git a/tools/test_gui.py b/tools/test_gui.py deleted file mode 100644 index 36bca72..0000000 --- a/tools/test_gui.py +++ /dev/null @@ -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()) diff --git a/tools/test_refactor.py b/tools/test_refactor.py deleted file mode 100644 index 0cc50f6..0000000 --- a/tools/test_refactor.py +++ /dev/null @@ -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()