diff --git a/sras_average.py b/sras_average.py index d997a15..42aeaca 100644 --- a/sras_average.py +++ b/sras_average.py @@ -47,7 +47,7 @@ def read_header_sections(sras: SrasFile) -> bytes: lost in a re-encode.""" with open(sras.path, "rb") as f: f.seek(HDR_SIZE) - return f.read(sras._data_offset - HDR_SIZE) + return f.read(sras.data_offset - HDR_SIZE) def average_rows(block: np.ndarray, n: int, discard_remainder: bool) -> np.ndarray: diff --git a/sras_compute.py b/sras_compute.py index fc7b30e..0305c54 100644 --- a/sras_compute.py +++ b/sras_compute.py @@ -768,7 +768,7 @@ def _skimage_phase_cross_correlation(): # ---- Geometry: local mm, ref mm, and the one affine builder --------------- -def _pixel_pitch_mm(sras: SrasFile, angle_idx: int) -> tuple[float, float]: +def pixel_pitch_mm(sras: SrasFile, angle_idx: int) -> tuple[float, float]: """(dx, dy) mm/pixel for one angle: dx is the file-wide constant pixel_x_mm; dy is this angle's own row spacing (assumed uniform, the same assumption _redraw_image makes when it builds the display extent). dy keeps @@ -803,7 +803,7 @@ def _local_half_extent_mm(sras: SrasFile, angle_idx: int) -> tuple[float, float] """(half width, half height) in mm from this angle's array center to the center of its outermost pixel.""" n_rows, n_frames = sras.image_shape(angle_idx) - dx, dy = _pixel_pitch_mm(sras, angle_idx) + dx, dy = pixel_pitch_mm(sras, angle_idx) return (n_frames - 1) / 2.0 * abs(dx), (n_rows - 1) / 2.0 * abs(dy) @@ -813,7 +813,7 @@ def _rotation_matrix(theta_deg: float) -> np.ndarray: return np.array([[c, -s], [s, c]]) # CCW rotation acting on (x, y) -def _nominal_delta_deg(sras: SrasFile, angle_idx: int, ref_idx: int) -> float: +def nominal_delta_deg(sras: SrasFile, angle_idx: int, ref_idx: int) -> float: """The rotation-stage's own reported angle change between two angles. Used only to *seed* the rotation search, never as the answer: the stage's @@ -888,7 +888,7 @@ def apply_alignment(result: AlignmentResult, angle_idx: int, img: np.ndarray, mode="constant", cval=0.0) -def _block_mean_2d(img: np.ndarray, fy: int, fx: int) -> np.ndarray: +def block_mean_2d(img: np.ndarray, fy: int, fx: int) -> np.ndarray: """Block-mean by independent row/column factors. Independent factors matter because the raw grid is strongly anisotropic (5 µm along x, 50 µm along y): a single square factor would either alias along x or throw away rows.""" @@ -944,10 +944,10 @@ def _prepare_reg_image(sras: SrasFile, angle_idx: int, img: np.ndarray, resampled onto the registration grid. Pre-averaging matters: the raw grid is 10x finer along x than along y, so sampling it directly at the (much coarser) isotropic registration pitch would alias badly along x.""" - dx, dy = _pixel_pitch_mm(sras, angle_idx) + dx, dy = pixel_pitch_mm(sras, angle_idx) fx = max(1, int(pitch_mm / abs(dx))) fy = max(1, int(pitch_mm / abs(dy))) - small = _block_mean_2d(np.asarray(img, dtype=np.float32), fy, fx) + small = block_mean_2d(np.asarray(img, dtype=np.float32), fy, fx) n_rows, n_frames = img.shape return _RegImage( small, dx * fx, dy * fy, @@ -1087,7 +1087,7 @@ def _reg_pitch_and_size(sras: SrasFile, max_dim: int, span = diag * margin native = float(np.sqrt( abs(sras.pixel_x_mm) - * max(abs(_pixel_pitch_mm(sras, a)[1]) for a in range(sras.n_angles)))) + * max(abs(pixel_pitch_mm(sras, a)[1]) for a in range(sras.n_angles)))) pitch = max(span / max_dim, native) n = int(scipy_fft.next_fast_len(max(16, int(np.ceil(span / pitch))))) return pitch, n @@ -1111,6 +1111,16 @@ def _registration_workers(sras: SrasFile, fine_dim: int) -> int: _TOTAL_BYTES_BUDGET // max(1, per_worker)))) +def registration_workers(sras: SrasFile) -> int: + """Public: concurrent-registration cap at the default fine grid.""" + return _registration_workers(sras, _DEFAULT_FINE_DIM) + + +def default_max_workers() -> int: + """Public: the module-wide worker cap (SRAS_MAX_WORKERS or cpu count).""" + return _MAX_WORKERS + + def _rotation_candidates(nominal_deg: float, search_deg: float, step_deg: float) -> list[float]: """Coarse rotation candidates: a window around *both* signs of the stage's @@ -1254,7 +1264,7 @@ def register_angle_to_reference( if not candidates: return RigidFit(0.0, (0.0, 0.0), -1.0, "none") - nominal = _nominal_delta_deg(sras, angle_idx, ref_angle_idx) + nominal = nominal_delta_deg(sras, angle_idx, ref_angle_idx) thetas = _rotation_candidates(nominal, search_deg, coarse_step_deg) # ---- Stage 1: coarse sweep, every source ------------------------------ @@ -1363,7 +1373,7 @@ def build_canvas_affine(sras: SrasFile, angle_idx: int, ref_angle_idx: int, (rows, cols) block-mean factor already applied to the image the caller will resample — 1:1 for the raw image, coarser for the manual-alignment preview's downsampled masks.""" - dx_a, dy_a = _pixel_pitch_mm(sras, angle_idx) + dx_a, dy_a = pixel_pitch_mm(sras, angle_idx) n_rows, n_frames = sras.image_shape(angle_idx) fy, fx = (max(1, int(v)) for v in src_downsample) if (fy, fx) != (1, 1): @@ -1398,7 +1408,7 @@ def _result_from_params(sras: SrasFile, ref_angle_idx: int, the shared canvas, then build each angle's canvas->raw affine. Pure matrix and bbox math, so it is cheap enough to call synchronously on the GUI thread on every manual edit.""" - pitch = _pixel_pitch_mm(sras, ref_angle_idx) + pitch = pixel_pitch_mm(sras, ref_angle_idx) canvas_origin_mm, canvas_shape = canvas_for_params( sras, ref_angle_idx, pitch, params, snap=True) diff --git a/sras_edit_scans.py b/sras_edit_scans.py index 8774daf..040e0d5 100644 --- a/sras_edit_scans.py +++ b/sras_edit_scans.py @@ -25,15 +25,17 @@ import struct import sys from pathlib import Path -from sras_format import ( - GEO_FMT_V6, GEO_SIZE_V6, HDR_FMT, HDR_FMT_V6, HDR_SIZE, HDR_SIZE_V6, - SrasFile, -) +from sras_format import GEO_FMT_V6, HDR_FMT, HDR_FMT_V6, HDR_SIZE, SrasFile _LEGACY_VERSIONS = (2, 3, 4, 5) _V6_VERSIONS = (6, 7) +def _die(msg: str): + print(f"Error: {msg}", file=sys.stderr) + sys.exit(1) + + def parse_args(): p = argparse.ArgumentParser( description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) @@ -106,7 +108,7 @@ def _write_legacy(sras: SrasFile, keep: list[int], out_path: Path): angle_table_size = sras.n_angles * 4 with open(sras.path, "rb") as f: f.seek(HDR_SIZE + angle_table_size) - shared_mid = f.read(sras._data_offset - (HDR_SIZE + angle_table_size)) + shared_mid = f.read(sras.data_offset - (HDR_SIZE + angle_table_size)) angle_bytes = n_rows * n_ch * n_frames * spf * bps @@ -115,128 +117,55 @@ def _write_legacy(sras: SrasFile, keep: list[int], out_path: Path): fout.write(sras.angles_deg[keep].astype(">f4").tobytes()) fout.write(shared_mid) for a in keep: - _copy_range(fin, fout, sras._data_offset + a * angle_bytes, angle_bytes) + _copy_range(fin, fout, sras.data_offset + a * angle_bytes, angle_bytes) # --------------------------------------------------------------------------- # v6/v7: per-angle geometry, ragged waveform blocks # --------------------------------------------------------------------------- -def _read_v6_sections(path: Path) -> dict: - """Raw, low-level read of everything before the waveform data. - - SrasFile._parse_v6 reads and discards each angle's x_delta (it's not - part of the display-facing geometry it exposes), so a round-trip through - SrasFile would silently drop that field. Re-parsing here keeps every - byte of the Per-Angle Geometry Table intact. - """ - with open(path, "rb") as f: - hdr_raw = f.read(HDR_SIZE_V6) - (magic, ver, n_angles_declared, x_start_nom, y_start_nom, x_delta_nom, - y_delta_nom, row_spacing, vel, freq, spf, sr, bps, - n_ch) = struct.unpack(HDR_FMT_V6, hdr_raw) - - angles_deg = list(struct.unpack(f">{n_angles_declared}f", - f.read(n_angles_declared * 4))) - - geo = [struct.unpack(GEO_FMT_V6, f.read(GEO_SIZE_V6)) - for _ in range(n_angles_declared)] # (x_start, x_delta, n_frames, n_rows) - - row_table = [f.read(geo[a][3] * 4) for a in range(n_angles_declared)] - - preamble_start = f.tell() - for _ in range(n_ch): - (length,) = struct.unpack(">H", f.read(2)) - f.read(length) - preambles_raw = _reread_span(f, preamble_start) - - bg_start = f.tell() - (n_bg,) = struct.unpack(">I", f.read(4)) - f.read(n_bg) - background_raw = _reread_span(f, bg_start) - - data_offset = f.tell() - - return { - "version": ver, "n_angles_declared": n_angles_declared, - "x_start_nom": x_start_nom, "y_start_nom": y_start_nom, - "x_delta_nom": x_delta_nom, "y_delta_nom": y_delta_nom, - "row_spacing": row_spacing, "vel": vel, "freq": freq, - "spf": spf, "sr": sr, "bps": bps, "n_ch": n_ch, - "angles_deg": angles_deg, "geo": geo, "row_table": row_table, - "preambles_raw": preambles_raw, "background_raw": background_raw, - "data_offset": data_offset, - } - - -def _reread_span(f, start: int) -> bytes: - end = f.tell() - f.seek(start) - span = f.read(end - start) - f.seek(end) - return span - - -def _v6_angle_offsets(sections: dict, file_size: int) -> list[tuple[int, int]]: - """(offset, nbytes) of each declared angle's waveform block, stopping at - the first angle whose data isn't fully on disk (aborted scan).""" - n_ch, spf, bps = sections["n_ch"], sections["spf"], sections["bps"] - offsets = [] - offset = sections["data_offset"] - for xs, xd, nf, nr in sections["geo"]: - nbytes = nr * n_ch * nf * spf * bps - if offset + nbytes > file_size: - break - offsets.append((offset, nbytes)) - offset += nbytes - return offsets - - -def _write_v6(in_path: Path, sections: dict, keep: list[int], out_path: Path): - file_size = in_path.stat().st_size - offsets = _v6_angle_offsets(sections, file_size) - geo = sections["geo"] - +def _write_v6(sras: SrasFile, keep: list[int], out_path: Path): header = struct.pack( - HDR_FMT_V6, b"SRAS", sections["version"], len(keep), - sections["x_start_nom"], sections["y_start_nom"], - sections["x_delta_nom"], sections["y_delta_nom"], - sections["row_spacing"], sections["vel"], sections["freq"], - sections["spf"], sections["sr"], sections["bps"], sections["n_ch"], + HDR_FMT_V6, b"SRAS", sras.version, len(keep), + sras.x_start_nominal_mm, sras.y_start_nominal_mm, + sras.x_delta_nominal_mm, sras.y_delta_nominal_mm, + sras.row_spacing_mm, sras.velocity_mm_s, sras.laser_freq_hz, + sras.samples_per_frame, sras.sample_rate_hz, + sras.bytes_per_sample, sras.n_channels, ) - with open(in_path, "rb") as fin, open(out_path, "wb") as fout: + blocks = {a: (offset, nbytes) for a, offset, nbytes in sras.iter_angle_blocks()} + with open(sras.path, "rb") as fin, open(out_path, "wb") as fout: fout.write(header) - fout.write(struct.pack(f">{len(keep)}f", - *[sections["angles_deg"][i] for i in keep])) + fout.write(sras.angles_deg[keep].astype(">f4").tobytes()) for i in keep: - fout.write(struct.pack(GEO_FMT_V6, *geo[i])) + fout.write(struct.pack( + GEO_FMT_V6, float(sras.x_start_mm[i]), + float(sras.x_delta_mm_per_angle[i]), + int(sras.n_frames[i]), int(sras.n_rows[i]))) for i in keep: - fout.write(sections["row_table"][i]) - fout.write(sections["preambles_raw"]) - fout.write(sections["background_raw"]) + fout.write(sras.y_pos_per_angle[i].astype(">f4").tobytes()) + fout.write(sras.preambles_raw) + fout.write(sras.background_raw) for i in keep: - off, nbytes = offsets[i] - _copy_range(fin, fout, off, nbytes) + offset, nbytes = blocks[i] + _copy_range(fin, fout, offset, nbytes) def main(): args = parse_args() in_path = Path(args.input) if not in_path.exists(): - print(f"Error: input file not found: {in_path}", file=sys.stderr) - sys.exit(1) + _die(f"input file not found: {in_path}") print(f"Reading {in_path} ...", flush=True) try: sras = SrasFile(str(in_path)) except ValueError as e: - print(f"Error: {e}", file=sys.stderr) - sys.exit(1) + _die(str(e)) if sras.version not in (*_LEGACY_VERSIONS, *_V6_VERSIONS): - print(f"Error: unsupported .sras version: {sras.version}", file=sys.stderr) - sys.exit(1) + _die(f"unsupported .sras version: {sras.version}") aborted_note = " (scan aborted; trailing angle(s) already excluded)" if sras.scan_aborted else "" print(f" Version : v{sras.version}", flush=True) @@ -247,16 +176,13 @@ def main(): return if not args.output: - print("Error: output path required unless --list is given.", file=sys.stderr) - sys.exit(1) + _die("output path required unless --list is given.") if not (args.drop or args.keep): - print("Error: specify --drop or --keep (see --list for indices).", file=sys.stderr) - sys.exit(1) + _die("specify --drop or --keep (see --list for indices).") out_path = Path(args.output) if out_path.resolve() == in_path.resolve(): - print("Error: output path must differ from input path.", file=sys.stderr) - sys.exit(1) + _die("output path must differ from input path.") try: if args.drop: @@ -265,12 +191,10 @@ def main(): else: keep = sorted(_parse_index_list(args.keep, sras.n_angles)) except ValueError as e: - print(f"Error: {e}", file=sys.stderr) - sys.exit(1) + _die(str(e)) if not keep: - print("Error: at least one angle must remain.", file=sys.stderr) - sys.exit(1) + _die("at least one angle must remain.") dropped = [a for a in range(sras.n_angles) if a not in keep] print(f"\nDropping angle(s): {dropped}") @@ -280,7 +204,7 @@ def main(): if sras.version in _LEGACY_VERSIONS: _write_legacy(sras, keep, out_path) else: - _write_v6(in_path, _read_v6_sections(in_path), keep, out_path) + _write_v6(sras, keep, out_path) in_mb = in_path.stat().st_size / 1024**2 out_mb = out_path.stat().st_size / 1024**2 diff --git a/sras_format.py b/sras_format.py index 0b2715d..46d8003 100644 --- a/sras_format.py +++ b/sras_format.py @@ -375,21 +375,34 @@ class SrasFile: angles = np.frombuffer(f.read(n_angles * 4), dtype=">f4").astype(np.float32) x_start = np.empty(n_angles, dtype=np.float64) + x_delta = np.empty(n_angles, dtype=np.float64) n_frames = np.empty(n_angles, dtype=np.int64) n_rows = np.empty(n_angles, dtype=np.int64) for a in range(n_angles): xs, xd, nf, nr = _read_struct(f, GEO_FMT_V6) - x_start[a], n_frames[a], n_rows[a] = xs, nf, nr + x_start[a], x_delta[a], n_frames[a], n_rows[a] = xs, xd, nf, nr y_pos_per_angle = [ np.frombuffer(f.read(int(n_rows[a]) * 4), dtype=">f4").astype(np.float32) for a in range(n_angles) ] + # Verbatim on-disk spans of the preamble and background sections, + # kept so file-rewriting tools (sras_edit_scans) can carry them + # over byte-for-byte without re-parsing. + span_start = f.tell() self._set_calibration(_read_preambles(f, n_ch), n_ch) - self.background = _read_background(f) + span_end = f.tell() + f.seek(span_start) + self.preambles_raw = f.read(span_end - span_start) - data_offset = f.tell() + span_start = span_end + self.background = _read_background(f) + span_end = f.tell() + f.seek(span_start) + self.background_raw = f.read(span_end - span_start) + + data_offset = span_end self._data_offset = data_offset @@ -428,6 +441,7 @@ class SrasFile: self.scan_aborted = n_complete < n_angles_declared self.angles_deg = angles[:n_complete] self.x_start_mm = x_start[:n_complete] + self.x_delta_mm_per_angle = x_delta[:n_complete] self.n_frames = n_frames[:n_complete] self.n_rows = n_rows[:n_complete] self._y_pos_per_angle = y_pos_per_angle[:n_complete] @@ -436,6 +450,37 @@ class SrasFile: if self.version == 7 and offset < file_size: self._parse_cach_section(offset) + # ------------------------------------------------------------------ + # Byte-layout accessors (public: used by file-rewriting tools) + # ------------------------------------------------------------------ + + @property + def data_offset(self) -> int: + """File offset where the waveform data begins (headers end).""" + return self._data_offset + + @property + def y_pos_per_angle(self) -> list[np.ndarray]: + """Per-angle Y row positions (mm). The list and its arrays are the + live parsed state — tools that reproject may replace entries.""" + return self._y_pos_per_angle + + @y_pos_per_angle.setter + def y_pos_per_angle(self, value: list[np.ndarray]): + self._y_pos_per_angle = value + + def iter_angle_blocks(self): + """Yields (angle_idx, byte_offset, byte_count) for each complete + angle's waveform block. Works for every version: legacy files have + uniform per-angle geometry, so the same walk applies.""" + offset = self._data_offset + for a in range(self.n_angles): + nbytes = (int(self.n_rows[a]) * self.n_channels + * int(self.n_frames[a]) * self.samples_per_frame + * self.bytes_per_sample) + yield a, offset, nbytes + offset += nbytes + # ------------------------------------------------------------------ # v7 cache tail (CACH section: precomputed DC / FFT images) # ------------------------------------------------------------------ @@ -445,12 +490,10 @@ class SrasFile: start), derived purely from the header + Per-Angle Geometry Table — independent of whether a cache tail is actually present. Used by both the parser and the in-place writer.""" - waveform_bytes = sum( - int(self.n_rows[a]) * self.n_channels * int(self.n_frames[a]) - * self.samples_per_frame * self.bytes_per_sample - for a in range(self.n_angles) - ) - return self._data_offset + int(waveform_bytes) + end = self._data_offset + for _, offset, nbytes in self.iter_angle_blocks(): + end = offset + nbytes + return end def _read_cache_block(self, f, hdr_fmt: str, magic: bytes, stores: list[list]) -> int | None: diff --git a/sras_viewer.py b/sras_viewer.py index c1c232f..23f9344 100644 --- a/sras_viewer.py +++ b/sras_viewer.py @@ -1127,7 +1127,7 @@ class ManualAlignmentDialog(QDialog): threshold = self.spin_mask_threshold_mv.value() fy, fx = self._downsample self._masks_small = { - a: compute._block_mean_2d((img >= threshold).astype(np.float32), fy, fx) + a: compute.block_mean_2d((img >= threshold).astype(np.float32), fy, fx) for a, img in self._dc4_mv.items() } @@ -1143,7 +1143,7 @@ class ManualAlignmentDialog(QDialog): mask-threshold change, Auto De-rotate, a rotation nudge/edit of the active angle. NOT triggered by a translation-only nudge — see _refresh_active_preview_layer.""" - dx_ref, dy_ref = compute._pixel_pitch_mm(self._sras, self._ref_angle_idx) + dx_ref, dy_ref = compute.pixel_pitch_mm(self._sras, self._ref_angle_idx) fy, fx = self._downsample pitch = (dx_ref * fx, dy_ref * fy) origin, shape = compute.canvas_for_params( @@ -1288,7 +1288,7 @@ class ManualAlignmentDialog(QDialog): for a in range(self._sras.n_angles): if a == self._ref_angle_idx: continue - self._angle_params[a].rotation_deg = sign * compute._nominal_delta_deg( + self._angle_params[a].rotation_deg = sign * compute.nominal_delta_deg( self._sras, a, self._ref_angle_idx) n_changed += 1 self._sync_active_spinboxes() @@ -1364,7 +1364,7 @@ class ManualAlignmentDialog(QDialog): f"{worst[1][1]})."] drifted = [] for a, _note in rows: - nominal = compute._nominal_delta_deg(self._sras, a, self._ref_angle_idx) + nominal = compute.nominal_delta_deg(self._sras, a, self._ref_angle_idx) got = self._angle_params[a].rotation_deg dev = min(abs(got - nominal), abs(got + nominal)) if dev > 1.0: diff --git a/sras_workers.py b/sras_workers.py index 5919b79..4a75dd8 100644 --- a/sras_workers.py +++ b/sras_workers.py @@ -253,7 +253,8 @@ class BatchCacheWorker(QObject): for path in paths: try: err = cache_file(path, self._mode, self._apply_bg_sub, - compute.get_fft_backend(), compute._MAX_WORKERS) + compute.get_fft_backend(), + compute.default_max_workers()) except Exception as exc: err = str(exc) done += 1 @@ -382,7 +383,7 @@ class CrossCorrelateWorker(_PooledWorker): self._search_deg = search_deg def _plan(self) -> int: - return compute._registration_workers(self._sras, compute._DEFAULT_FINE_DIM) + return compute.registration_workers(self._sras) def _items(self): return self._angles diff --git a/tests/test_alignment.py b/tests/test_alignment.py index 18be785..04bb85f 100644 --- a/tests/test_alignment.py +++ b/tests/test_alignment.py @@ -40,7 +40,7 @@ def mm_transform(sras: SrasFile, result, angle_idx: int) -> np.ndarray: 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) + 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) @@ -99,7 +99,7 @@ def test_stage_coordinates_are_not_consulted(rig): 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.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) @@ -118,7 +118,7 @@ def test_canvas_is_reference_grid_extended(rig): 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)), \ + == compute.pixel_pitch_mm(sras, 0)), \ "canvas pitch is angle 0's own pitch" n_rows, n_cols = result.canvas_shape @@ -169,7 +169,7 @@ def test_downsampled_preview_lands_with_full_res(rig): 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), + 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), diff --git a/tests/test_gui.py b/tests/test_gui.py index 7b1ebec..cb0bc28 100644 --- a/tests/test_gui.py +++ b/tests/test_gui.py @@ -261,7 +261,7 @@ def test_manual_alignment_geometry(ctx): 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) + 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" @@ -270,7 +270,7 @@ def test_manual_alignment_geometry(ctx): 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 + 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 " @@ -356,7 +356,7 @@ def test_auto_derotate(ctx): 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) + 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, \ diff --git a/tools/check_equivalence.py b/tools/check_equivalence.py index 37934dc..9488040 100644 --- a/tools/check_equivalence.py +++ b/tools/check_equivalence.py @@ -47,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 diff --git a/tools/make_test_sras.py b/tools/make_test_sras.py index 92b0b54..0a3651b 100644 --- a/tools/make_test_sras.py +++ b/tools/make_test_sras.py @@ -11,12 +11,19 @@ Usage: import argparse import struct +import sys from pathlib import Path import numpy as np -HDR_FMT_V6 = ">4sBHfffffffIdBB" -GEO_FMT_V6 = ">ffIH" +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +# 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. @@ -171,12 +178,6 @@ def _sample_shape_mv(u: np.ndarray, v: np.ndarray) -> np.ndarray: return img -def _rot(theta_deg: float) -> np.ndarray: - t = np.radians(theta_deg) - c, s = np.cos(t), np.sin(t) - return np.array([[c, -s], [s, c]]) - - 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 @@ -260,7 +261,6 @@ def write_rotating(path: Path, n_angles: int = 5, samples_per_frame: int = 4, "y_starts": y_starts, "dx_mm": _ROT_DX_MM, "dy_mm": _ROT_DY_MM} -HDR_FMT_LEGACY = ">4sBHHffffIIdBB" def write_legacy(path: Path, version: int = 4, n_angles: int = 2,