Merge remote-tracking branch 'origin/main' into main
# Conflicts: # sras_compute.py # sras_viewer.py # sras_workers.py # tools/make_test_sras.py # tools/test_refactor.py
This commit is contained in:
Thomas Ales [M S E]
@@ -0,0 +1,106 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Benchmark the FFT peak-search path: exact vs zoom, serial vs pooled.
|
||||
|
||||
Reports wall time, waveforms/s, CPU utilization (utime+stime over wall, in
|
||||
cores), and verifies every variant against the exact reference image.
|
||||
|
||||
Usage:
|
||||
python tools/bench_fft.py # synthetic, pads 1/8/40
|
||||
python tools/bench_fft.py --pads 40 --spf 2500 --rows 8 --frames 1024
|
||||
python tools/bench_fft.py --real /path/big.sras --real-rows 32 --pads 40
|
||||
"""
|
||||
|
||||
import argparse
|
||||
import resource
|
||||
import sys
|
||||
import tempfile
|
||||
import time
|
||||
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 compute_rf_image, set_fft_backend # noqa: E402
|
||||
from sras_format import SrasFile # noqa: E402
|
||||
import tools.make_test_sras as gen # noqa: E402
|
||||
from tools.check_equivalence import row_slice # noqa: E402
|
||||
|
||||
|
||||
def _timed(fn):
|
||||
r0 = resource.getrusage(resource.RUSAGE_SELF)
|
||||
t0 = time.perf_counter()
|
||||
out = fn()
|
||||
wall = time.perf_counter() - t0
|
||||
r1 = resource.getrusage(resource.RUSAGE_SELF)
|
||||
cpu = (r1.ru_utime - r0.ru_utime) + (r1.ru_stime - r0.ru_stime)
|
||||
return out, wall, cpu / max(wall, 1e-9)
|
||||
|
||||
|
||||
def bench(sras, pads, backends):
|
||||
n_wf = sum(int(sras.n_rows[a]) * int(sras.n_frames[a])
|
||||
for a in range(sras.n_angles))
|
||||
spf = sras.samples_per_frame
|
||||
print(f"{n_wf} waveforms x {spf} samples, {sras.n_angles} angle(s)")
|
||||
print(f"{'pad':>4} {'backend':>8} {'variant':>16} {'wall':>9} "
|
||||
f"{'wf/s':>10} {'util':>6} match")
|
||||
|
||||
for pad in pads:
|
||||
n_fft = spf * pad if pad > 1 else None
|
||||
for backend in backends:
|
||||
set_fft_backend(backend)
|
||||
|
||||
def run(**kw):
|
||||
imgs = [compute_rf_image(sras, a, dc_threshold_mv=None,
|
||||
apply_bg_sub=True, n_fft=n_fft, **kw)
|
||||
for a in range(sras.n_angles)]
|
||||
return np.concatenate([i.ravel() for i in imgs])
|
||||
|
||||
ref, wall, util = _timed(lambda: run(exact=True))
|
||||
rows = [("exact(serial)", ref, wall, util, True)]
|
||||
for label, kw in (("zoom(serial)", dict(max_workers=1)),
|
||||
("zoom(pool)", {})):
|
||||
img, wall, util = _timed(lambda: run(**kw))
|
||||
rows.append((label, img, wall, util, bool(np.array_equal(img, ref))))
|
||||
for label, img, wall, util, ok in rows:
|
||||
print(f"{pad:>4} {backend:>8} {label:>16} {wall:>8.2f}s "
|
||||
f"{n_wf / wall:>10.0f} {util:>5.1f}x "
|
||||
f"{'OK' if ok else 'MISMATCH'}")
|
||||
|
||||
|
||||
def main():
|
||||
p = argparse.ArgumentParser(description=__doc__)
|
||||
p.add_argument("--pads", default="1,8,40",
|
||||
help="comma-separated pad factors (default 1,8,40)")
|
||||
p.add_argument("--spf", type=int, default=2500)
|
||||
p.add_argument("--rows", type=int, default=8)
|
||||
p.add_argument("--frames", type=int, default=1024)
|
||||
p.add_argument("--backends", default=None,
|
||||
help="comma-separated (default: scipy,pyfftw if available)")
|
||||
p.add_argument("--real", help="path to a real .sras file")
|
||||
p.add_argument("--real-rows", type=int, default=32,
|
||||
help="rows of angle 0 to use from the real file")
|
||||
args = p.parse_args()
|
||||
|
||||
pads = [int(x) for x in args.pads.split(",")]
|
||||
if args.backends:
|
||||
backends = args.backends.split(",")
|
||||
else:
|
||||
backends = ["scipy"] + (["pyfftw"] if compute.PYFFTW_AVAILABLE else [])
|
||||
|
||||
if args.real:
|
||||
sras = row_slice(SrasFile(args.real), 0, args.real_rows)
|
||||
sras.data = [sras.data[0]]
|
||||
sras.n_angles = 1
|
||||
bench(sras, pads, backends)
|
||||
else:
|
||||
with tempfile.TemporaryDirectory(prefix="sras_bench_") as tmp:
|
||||
path = Path(tmp) / "bench.sras"
|
||||
gen.write(path, n_angles=1, seed=0, samples_per_frame=args.spf,
|
||||
geometry=[(args.rows, args.frames)])
|
||||
bench(SrasFile(str(path)), pads, backends)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
+26
-29
@@ -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:
|
||||
@@ -63,8 +47,8 @@ def row_slice(sras: SrasFile, angle_idx: int, n_rows: int) -> SrasFile:
|
||||
view.n_rows[angle_idx] = n
|
||||
view.data = list(sras.data)
|
||||
view.data[angle_idx] = sras.data[angle_idx][:n]
|
||||
view._y_pos_per_angle = list(sras._y_pos_per_angle)
|
||||
view._y_pos_per_angle[angle_idx] = sras._y_pos_per_angle[angle_idx][:n]
|
||||
view.y_pos_per_angle = list(sras.y_pos_per_angle)
|
||||
view.y_pos_per_angle[angle_idx] = sras.y_pos_per_angle[angle_idx][:n]
|
||||
return view
|
||||
|
||||
|
||||
@@ -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():
|
||||
|
||||
+140
-8
@@ -18,9 +18,12 @@ import numpy as np
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
||||
|
||||
# Packed straight from sras_format.py's own constants (not a hand-copy) so
|
||||
# this generator can never silently drift from what the real parser expects.
|
||||
from sras_format import GEO_FMT_V6, HDR_FMT as HDR_FMT_LEGACY, HDR_FMT_V6 # noqa: E402
|
||||
# Single source of truth for the byte layout: the reader's own constants.
|
||||
# The byte *assembly* below stays independent, so a writer bug can't be
|
||||
# masked by a matching reader bug.
|
||||
from sras_format import HDR_FMT as HDR_FMT_LEGACY # noqa: E402
|
||||
from sras_format import GEO_FMT_V6, HDR_FMT_V6 # noqa: E402
|
||||
from sras_compute import _rotation_matrix as _rot # noqa: E402
|
||||
|
||||
# Per-angle (n_rows, n_frames) — deliberately different per angle so ragged
|
||||
# geometry handling is actually exercised.
|
||||
@@ -46,12 +49,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.
|
||||
@@ -104,7 +107,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,
|
||||
@@ -123,12 +128,139 @@ 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
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Rotating-sample scan: one shape, imaged at several known rotations
|
||||
# ---------------------------------------------------------------------------
|
||||
#
|
||||
# The scan the angle-alignment path actually has to solve: every angle images
|
||||
# the *same* sample at a different known rotation and offset, and a correct
|
||||
# alignment stacks them all back into one shape. Two properties are
|
||||
# deliberately hostile:
|
||||
#
|
||||
# * every angle gets a different window size and a different, meaningless
|
||||
# stage x_start / y0 — alignment must ignore per-angle stage coordinates
|
||||
# entirely, so any code that reads them will visibly fail here;
|
||||
# * the pixel grid is strongly anisotropic (5 µm along x, 50 µm along y),
|
||||
# like the real instrument, so any registration that rotates raw indices
|
||||
# instead of millimetres shears the image and cannot converge.
|
||||
|
||||
_ROT_DX_MM = 0.005 # x pitch, from velocity/laser_freq below
|
||||
_ROT_DY_MM = 0.05 # row spacing
|
||||
_ROT_BG_MV = 4.0
|
||||
_ROT_FG_MV = 160.0
|
||||
|
||||
|
||||
# How far the sample sits from the rotation axis. Non-zero on purpose: on the
|
||||
# real instrument every angle's scan window is centred on the rotation axis
|
||||
# while the sample is not, so each scan sees the sample somewhere else along a
|
||||
# circle. That offset is exactly what a wrong rotation pivot turns into a ring
|
||||
# of scans instead of a stack, so a centred test sample would hide the bug.
|
||||
_ROT_SAMPLE_OFFSET_MM = (0.55, 0.40)
|
||||
|
||||
|
||||
def _sample_shape_mv(u: np.ndarray, v: np.ndarray) -> np.ndarray:
|
||||
"""An asymmetric test sample in its own mm frame, chirally distinct at
|
||||
every rotation (no 180° ambiguity) and with structure at several radii so
|
||||
rotation is well determined."""
|
||||
u = u - _ROT_SAMPLE_OFFSET_MM[0]
|
||||
v = v - _ROT_SAMPLE_OFFSET_MM[1]
|
||||
img = np.full(u.shape, _ROT_BG_MV, dtype=np.float32)
|
||||
img[((u / 0.85) ** 2 + (v / 0.40) ** 2) <= 1.0] = _ROT_FG_MV # bar
|
||||
img[(np.abs(u - 0.55) <= 0.22) & (np.abs(v - 0.62) <= 0.22)] = _ROT_FG_MV # nub
|
||||
img[((u + 0.75) ** 2 + (v + 0.30) ** 2) <= 0.20 ** 2] = _ROT_FG_MV # dot
|
||||
return img
|
||||
|
||||
|
||||
def write_rotating(path: Path, n_angles: int = 5, samples_per_frame: int = 4,
|
||||
seed: int = 0) -> dict:
|
||||
"""Write a v6 file whose CH4 DC image is one sample seen at n_angles known
|
||||
rotations, and return the ground truth each angle should register to.
|
||||
|
||||
``truth[a] = (rotation_deg, (shift_x_mm, shift_y_mm))`` is the rigid map
|
||||
from angle *a*'s local mm (origin at its own array center) to angle 0's —
|
||||
exactly what ``register_angle_to_reference`` is supposed to recover.
|
||||
"""
|
||||
rng = np.random.default_rng(seed)
|
||||
n_ch, bps = 3, 1
|
||||
cal = [(1.5625e-3, -87.04, 0.0), (2.0e-3, -60.0, 1.0e-3), (2.5e-3, -40.0, -2.0e-3)]
|
||||
ymult_mv, yoff, yzero_mv = cal[2][0] * 1000, cal[2][1], cal[2][2] * 1000
|
||||
|
||||
stage_angles, geom, x_starts, y_starts, thetas, offsets = [], [], [], [], [], []
|
||||
for a in range(n_angles):
|
||||
stage = -37.0 * a # what the rotation stage reports
|
||||
stage_angles.append(stage)
|
||||
# The true image rotation is the negative of the stage's reported
|
||||
# angle: the stage's positive sense is the opposite of math-positive
|
||||
# (x toward y) in scan mm. Nothing may depend on knowing that — the
|
||||
# registration search tries both signs.
|
||||
thetas.append(-stage)
|
||||
offsets.append((0.0, 0.0) if a == 0
|
||||
else (float(rng.uniform(-0.3, 0.3)), float(rng.uniform(-0.3, 0.3))))
|
||||
# A different window per angle, all centred on the same array center —
|
||||
# the real instrument grows each angle's axis-aligned bounding box to
|
||||
# cover the rotated ROI. Sized so the off-axis sample stays inside every
|
||||
# window at every angle, keeping the expected result unambiguous.
|
||||
geom.append((88 + 8 * a, 780 + 60 * a))
|
||||
# Meaningless per-angle stage positions: correct alignment never reads
|
||||
# them, so scattering them proves it.
|
||||
x_starts.append(float(20.0 + rng.uniform(-6.0, 6.0)))
|
||||
y_starts.append(float(30.0 + rng.uniform(-6.0, 6.0)))
|
||||
|
||||
out = bytearray()
|
||||
out += struct.pack(
|
||||
HDR_FMT_V6, b"SRAS", 6, n_angles,
|
||||
x_starts[0], y_starts[0], 1.0, 1.0, _ROT_DY_MM,
|
||||
_VELOCITY_MM_S, _VELOCITY_MM_S / _ROT_DX_MM, # velocity/freq -> 5 µm pitch
|
||||
samples_per_frame, _SAMPLE_RATE_HZ, bps, n_ch,
|
||||
)
|
||||
out += np.array(stage_angles, dtype=">f4").tobytes()
|
||||
for a, (n_rows, n_frames) in enumerate(geom):
|
||||
out += struct.pack(GEO_FMT_V6, x_starts[a], 1.0, n_frames, n_rows)
|
||||
for a, (n_rows, _) in enumerate(geom):
|
||||
out += (y_starts[a] + np.arange(n_rows) * _ROT_DY_MM).astype(">f4").tobytes()
|
||||
for ymult_v, yoff_a, yzero_v in cal:
|
||||
p = _preamble(ymult_v, yoff_a, yzero_v)
|
||||
out += struct.pack(">H", len(p)) + p
|
||||
background = rng.integers(-8, 9, size=samples_per_frame, dtype=np.int8)
|
||||
out += struct.pack(">I", samples_per_frame) + background.tobytes()
|
||||
|
||||
truth, dc4_images = {}, []
|
||||
for a, (n_rows, n_frames) in enumerate(geom):
|
||||
# Local mm of every pixel, measured from this angle's own array center.
|
||||
lx = (np.arange(n_frames) - (n_frames - 1) / 2.0) * _ROT_DX_MM
|
||||
ly = (np.arange(n_rows) - (n_rows - 1) / 2.0) * _ROT_DY_MM
|
||||
gx, gy = np.meshgrid(lx, ly)
|
||||
# local = R(theta) @ sample + offset, so sample = R(theta)^T @ (local - offset)
|
||||
rel = np.stack([gx - offsets[a][0], gy - offsets[a][1]], axis=-1)
|
||||
s = rel @ _rot(thetas[a]) # == rel @ R^T.T == R^T @ rel
|
||||
dc4 = _sample_shape_mv(s[..., 0], s[..., 1])
|
||||
dc4_images.append(dc4)
|
||||
|
||||
inv = _rot(-thetas[a])
|
||||
truth[a] = (-thetas[a],
|
||||
tuple(float(v) for v in -(inv @ np.array(offsets[a]))))
|
||||
|
||||
adc4 = np.clip(np.round((dc4 - yzero_mv) / ymult_mv + yoff), -128, 127).astype(np.int8)
|
||||
block = np.zeros((n_rows, n_ch, n_frames, samples_per_frame), dtype=np.int8)
|
||||
block[:, 2] = adc4[:, :, None] # CH4 carries the sample
|
||||
block[:, 1] = 10 # CH3 flat
|
||||
block[:, 0] = rng.integers(-40, 41, size=(n_rows, n_frames, samples_per_frame),
|
||||
dtype=np.int8) # CH1 noise
|
||||
out += block.tobytes()
|
||||
|
||||
path.write_bytes(bytes(out))
|
||||
return {"n_angles": n_angles, "geometry": geom, "stage_angles_deg": stage_angles,
|
||||
"truth": truth, "dc4_mv": dc4_images, "x_starts": x_starts,
|
||||
"y_starts": y_starts, "dx_mm": _ROT_DX_MM, "dy_mm": _ROT_DY_MM}
|
||||
|
||||
|
||||
def write_legacy(path: Path, version: int = 4, n_angles: int = 2,
|
||||
n_rows: int = 4, n_frames: int = 10,
|
||||
samples_per_frame: int = 32, seed: int = 0) -> dict:
|
||||
|
||||
@@ -1,485 +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 # 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())
|
||||
|
||||
# --- Alignment pivot is a signal-weighted centroid, not the raw bbox --
|
||||
# center, and is independent of any DC threshold (so a threshold that
|
||||
# happens to leave a real angle's binary mask empty can't silently
|
||||
# degrade the pivot back to the bbox center).
|
||||
corner_signal = np.zeros(s.image_shape(0), dtype=np.float32)
|
||||
corner_signal[0, 0] = 1.0 # single spike -> weighted centroid is exact
|
||||
expected_corner = (float(s.x_axis_mm(0)[0]), float(s.y_positions_mm(0)[0]))
|
||||
centroid = compute._signal_centroid_mm(s, 0, corner_signal)
|
||||
check("signal-weighted centroid of a single spike pixel is that pixel exactly",
|
||||
np.allclose(centroid, expected_corner), f"{centroid} vs {expected_corner}")
|
||||
bbox_center = compute._bbox_center_mm(s, 0)
|
||||
check("signal centroid differs from the raw scan-window bbox center",
|
||||
not np.allclose(centroid, bbox_center),
|
||||
f"centroid {centroid} vs bbox center {bbox_center}")
|
||||
|
||||
# compute_pivot_points_mm should reuse a pre-computed dc4_mv dict rather
|
||||
# than recomputing from the real DC4 image (which has no such spike and
|
||||
# would give a different answer if silently recomputed).
|
||||
reused_pivot = compute.compute_pivot_points_mm(s, dc4_mv={0: corner_signal})[0]
|
||||
check("compute_pivot_points_mm reuses a pre-computed dc4_mv dict",
|
||||
np.allclose(reused_pivot, expected_corner))
|
||||
|
||||
# A perfectly flat signal carries no information to weight by, so it
|
||||
# falls back to the bbox center rather than producing a NaN/degenerate
|
||||
# centroid.
|
||||
flat_signal = np.full(s.image_shape(0), 5.0, dtype=np.float32)
|
||||
flat_centroid = compute._signal_centroid_mm(s, 0, flat_signal)
|
||||
check("a perfectly flat signal falls back to the bbox center",
|
||||
np.allclose(flat_centroid, bbox_center))
|
||||
|
||||
# --- Rotation sign convention: negative of the raw angles_deg delta ----
|
||||
check("_theta_deg negates the raw angles_deg delta (GR stage's positive "
|
||||
"angle is the opposite rotational sense from this module's CCW "
|
||||
"math convention)",
|
||||
all(np.isclose(compute._theta_deg(s, a, 0),
|
||||
-(float(s.angles_deg[a]) - float(s.angles_deg[0])))
|
||||
for a in range(s.n_angles)))
|
||||
|
||||
# --- FFT phase correlation recovers a known synthetic pixel shift ------
|
||||
rng = np.random.default_rng(0)
|
||||
corr_ref = np.zeros((40, 50), dtype=np.float32)
|
||||
corr_ref[10:25, 15:35] = 1.0
|
||||
corr_ref += 0.05 * rng.standard_normal(corr_ref.shape).astype(np.float32)
|
||||
corr_mov = np.roll(corr_ref, shift=(4, -7), axis=(0, 1))
|
||||
dr, dc = compute._phase_correlate_shift(corr_ref, corr_mov)
|
||||
check("phase correlation recovers the shift that aligns mov onto ref",
|
||||
(dr, dc) == (-4, 7), f"got (dr, dc)={(dr, dc)}")
|
||||
|
||||
# --- Open: must NOT seed from the still-live automatic AlignmentResult --
|
||||
# The automatic result's translation comes from FFT phase correlation --
|
||||
# the very thing manual mode exists to work around -- so manual mode
|
||||
# must start from identity (centroids coincide, zero shift) 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: rotation only, translation untouched ---------------
|
||||
shift_before_derotate = dlg._angle_params[active].shift_mm
|
||||
dlg._on_auto_derotate()
|
||||
expected_theta = compute._theta_deg(s, active, dlg._ref_angle_idx)
|
||||
check("auto de-rotate set the known analytic angle",
|
||||
abs(dlg._angle_params[active].rotation_deg - expected_theta) < 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)
|
||||
|
||||
# --- Auto Cross-Correlate: rotation + FFT-correlated shift, backgrounded -
|
||||
check("cross-correlate action enabled once masks are ready",
|
||||
dlg.btn_auto_correlate.isEnabled())
|
||||
dlg._on_auto_correlate()
|
||||
check("auto cross-correlate completed", wait_until(
|
||||
lambda: not win._job_running("manual_align_correlate"), timeout_ms=30000))
|
||||
check("auto cross-correlate set the known analytic angle for every angle",
|
||||
all(abs(dlg._angle_params[a].rotation_deg
|
||||
- compute._theta_deg(s, a, dlg._ref_angle_idx)) < 1e-6
|
||||
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)
|
||||
|
||||
# The thresholded-mask option should also work end to end.
|
||||
dlg.combo_correlate_source.setCurrentIndex(1) # thresholded mask
|
||||
dlg._on_auto_correlate()
|
||||
check("auto cross-correlate (thresholded-mask option) completed", wait_until(
|
||||
lambda: not win._job_running("manual_align_correlate"), timeout_ms=30000))
|
||||
|
||||
# --- 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())
|
||||
@@ -1,365 +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"]
|
||||
# int16 (not float32) before .mean(): matches average_rows' own
|
||||
# float64-accumulator behavior for integer input, so this doesn't
|
||||
# drift from what average_rows actually guarantees.
|
||||
expect0 = src_data[0][:, :, 0:4, :].astype(np.int16).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"
|
||||
proc2 = subprocess.run([sys.executable, str(repo / "sras_average.py"),
|
||||
str(src), str(dst2), "--n", "5"],
|
||||
capture_output=True, text=True, cwd=repo)
|
||||
check("sras_average ran (--n 5)", proc2.returncode == 0,
|
||||
(proc2.stderr or proc2.stdout).strip()[-200:])
|
||||
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"
|
||||
proc3 = subprocess.run([sys.executable, str(repo / "sras_average.py"),
|
||||
str(src), str(dst3), "--n", "5", "--discard-remainder"],
|
||||
capture_output=True, text=True, cwd=repo)
|
||||
check("sras_average ran (--n 5 --discard-remainder)", proc3.returncode == 0,
|
||||
(proc3.stderr or proc3.stdout).strip()[-200:])
|
||||
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()
|
||||
Reference in New Issue
Block a user