Thomas Ales
aa06fa1460
A multi-angle scan runs for hours, but every angle was referenced against
one background captured before the first row of the first angle. That
reference has drifted by the last angle, and comparing angles — the whole
point of a multi-angle scan — was comparing each one against a noise floor
measured at whichever angle came first.
Every angle now captures its own. Before each angle's rows, the operator is
prompted to switch the Genesis laser off, the engine averages a fresh CH1
record, and the operator switches it back on. The data block therefore reads
[background][scan][background][scan] …, one pair per angle.
Format v7 (scan) and v11 (SAW check) carry the background inside the data
block, one length-prefixed block ahead of each angle's rows; the single
block that sat between the preambles and the data is gone. Per-angle offsets
now come from a walk of the data block at parse time rather than arithmetic
over the geometry table, and an angle whose background is not fully on disk
is the frontier — nothing of it was written yet.
v6/v10 files still read: SrasFile hands their one background to every angle,
so readers never branch on the version. Nothing writes them, and a resume
refuses them, since a re-acquired angle writes a block the old layout has no
room for. A resumed v7 angle rewrites its background in place, and the
engine checks the new block fits the room the file has before writing it —
anything else would shift every row behind it.
Two fixes made along the way:
* QtScanController never accepted file_version, so every scan launched
from the app raised TypeError at construction.
* angle_status() left its cursor parked at the frontier, so every angle
past it reported the frontier's own data_offset — which handed a resumed
scan the same write position for several angles. Two recorded offsets in
tests/golden/sras_expected.json are corrected accordingly.
The v6 goldens stay as parser fixtures; the writer is now locked against
bytes the test lays out from scan_format.md itself.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
251 lines
10 KiB
Python
251 lines
10 KiB
Python
"""core.sras_format: the current writer against the spec, and the parser
|
|
against the pre-refactor golden fixtures.
|
|
|
|
The goldens are legacy v6 files produced by the original sc3_aui_app
|
|
implementation — one background for the whole file. Nothing writes that
|
|
layout any more, so they lock the parser (field-for-field, including the
|
|
frontier walk) rather than the writer. The writer is locked instead against
|
|
bytes this test lays out from scan_format.md itself.
|
|
"""
|
|
import json
|
|
import struct
|
|
from dataclasses import asdict
|
|
from pathlib import Path
|
|
|
|
import numpy as np
|
|
import pytest
|
|
|
|
from core.sras_format import (
|
|
BG_LEN_FMT, GEOM_FMT, HDR_FMT, MAGIC, VERSION, VERSION_SAW_CHECK,
|
|
SrasFile, create_scan_file,
|
|
)
|
|
from golden_util import (
|
|
BACKGROUND, CHANNELS, PREAMBLES, SAMPLE_RATE, SPF, angle_background,
|
|
synthetic_frame, tiny_plan, write_v7,
|
|
)
|
|
|
|
GOLDEN = Path(__file__).parent / "golden"
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def expected():
|
|
with open(GOLDEN / "sras_expected.json") as f:
|
|
return json.load(f)
|
|
|
|
|
|
def spec_bytes(plan):
|
|
"""The v7 layout spelled out from scan_format.md, writer not involved."""
|
|
buf = bytearray()
|
|
buf += struct.pack(HDR_FMT, MAGIC, VERSION, plan.n_angles,
|
|
plan.x_start_nominal, plan.y_start_nominal,
|
|
plan.x_delta_nominal, plan.y_delta_nominal,
|
|
plan.row_spacing, plan.velocity_mm_s, plan.laser_freq_hz,
|
|
SPF, SAMPLE_RATE, 1, len(CHANNELS))
|
|
buf += struct.pack(f">{plan.n_angles}f", *plan.angles)
|
|
for pa in plan.per_angle:
|
|
buf += struct.pack(GEOM_FMT, pa.x_start, pa.x_delta, pa.n_frames, pa.n_rows)
|
|
for pa in plan.per_angle:
|
|
buf += struct.pack(f">{pa.n_rows}f", *pa.y_positions)
|
|
for pre in PREAMBLES:
|
|
buf += struct.pack(">H", len(pre)) + pre.encode("utf-8")
|
|
for ai, pa in enumerate(plan.per_angle):
|
|
bg = angle_background(ai)
|
|
buf += struct.pack(BG_LEN_FMT, len(bg)) + bg
|
|
for ri in range(pa.n_rows):
|
|
for ci in range(len(CHANNELS)):
|
|
for fi in range(pa.n_frames):
|
|
buf += synthetic_frame(ai, ri, ci, fi)
|
|
return bytes(buf)
|
|
|
|
|
|
def test_writer_matches_the_spec_byte_for_byte(tmp_path):
|
|
out = tmp_path / "v7.sras"
|
|
plan = write_v7(out)
|
|
assert out.read_bytes() == spec_bytes(plan)
|
|
|
|
|
|
def test_writer_refuses_the_legacy_versions(tmp_path):
|
|
for version in (6, 10):
|
|
with pytest.raises(ValueError, match=f"version {version}"):
|
|
create_scan_file(tmp_path / "bad.sras", tiny_plan(), SPF,
|
|
SAMPLE_RATE, PREAMBLES, version=version)
|
|
assert not (tmp_path / "bad.sras").exists()
|
|
|
|
|
|
def test_header_matches_golden(expected):
|
|
sras = SrasFile(GOLDEN / "complete.sras")
|
|
h = expected["header"]
|
|
assert asdict(sras.header) == {
|
|
"n_angles": h["n_angles"],
|
|
"x_start_nominal": h["x_start_nominal"], "y_start_nominal": h["y_start_nominal"],
|
|
"x_delta_nominal": h["x_delta_nominal"], "y_delta_nominal": h["y_delta_nominal"],
|
|
"row_spacing": h["row_spacing"], "velocity": h["velocity"],
|
|
"laser_freq": h["laser_freq"],
|
|
"samples_per_frame": h["samples_per_frame"], "sample_rate": h["sample_rate"],
|
|
"bytes_per_sample": h["bytes_per_sample"], "n_channels": h["n_channels"],
|
|
}
|
|
assert sras.data_start_offset == h["data_start_offset"]
|
|
assert [pa.angle_deg for pa in sras.per_angle] == h["angles"]
|
|
for pa, exp in zip(sras.per_angle, h["per_angle"], strict=True):
|
|
assert pa.angle_deg == exp["angle"]
|
|
assert pa.x_start == exp["x_start"]
|
|
assert pa.x_delta == exp["x_delta"]
|
|
assert pa.n_frames == exp["n_frames"]
|
|
assert pa.n_rows == exp["n_rows"]
|
|
assert pa.y_positions == exp["y_positions"]
|
|
|
|
|
|
def test_frontier_all_truncation_variants(expected):
|
|
"""Every field the goldens recorded, plus the one added since.
|
|
|
|
The expectations predate AngleStatus.bg_offset, so they are compared key
|
|
by key; a v6 angle has no background block of its own, which is exactly
|
|
what bg_offset == data_offset says.
|
|
|
|
Two of the recorded data_offsets were corrected when the walk moved into
|
|
the parser: an angle past the frontier used to report the frontier's own
|
|
offset, because the old walk stopped advancing its cursor there, which
|
|
handed a resumed scan the same write position for every missing angle.
|
|
They are now the declared position each angle will be written at.
|
|
"""
|
|
for name, exp_statuses in expected["statuses"].items():
|
|
statuses = SrasFile(GOLDEN / name).angle_status()
|
|
assert len(statuses) == len(exp_statuses), f"mismatch for {name}"
|
|
for status, exp in zip(statuses, exp_statuses, strict=True):
|
|
got = asdict(status)
|
|
assert {k: got[k] for k in exp} == exp, f"mismatch for {name}"
|
|
assert status.bg_offset == status.data_offset
|
|
assert status.bg_bytes == 0
|
|
|
|
|
|
def test_legacy_preambles_and_shared_background():
|
|
"""A v6 file's one background stands in for every angle's."""
|
|
sras = SrasFile(GOLDEN / "complete.sras")
|
|
assert sras.preambles == PREAMBLES
|
|
assert sras.is_legacy_layout
|
|
assert sras.backgrounds == [BACKGROUND] * sras.header.n_angles
|
|
assert np.array_equal(sras.background_array(1),
|
|
np.frombuffer(BACKGROUND, dtype=np.int8))
|
|
|
|
|
|
# ── Per-angle backgrounds (v7/v11) ───────────────────────────────────────────
|
|
|
|
def test_each_angle_keeps_its_own_background(tmp_path):
|
|
out = tmp_path / "v7.sras"
|
|
plan = write_v7(out)
|
|
|
|
sras = SrasFile(out)
|
|
assert not sras.is_legacy_layout
|
|
assert sras.backgrounds == [angle_background(ai)
|
|
for ai in range(plan.n_angles)]
|
|
assert [s.status for s in sras.angle_status()] == ["OK"] * plan.n_angles
|
|
# Every angle's rows start just past its own background block …
|
|
for st in sras.angle_status():
|
|
assert st.bg_bytes == 4 + SPF
|
|
assert st.data_offset == st.bg_offset + st.bg_bytes
|
|
# … and the data itself still reads back frame for frame.
|
|
assert sras.load_row(1, 2, 0).tobytes() == b"".join(
|
|
synthetic_frame(1, 2, 0, fi) for fi in range(plan.per_angle[1].n_frames))
|
|
sras.close()
|
|
|
|
|
|
def test_saw_check_version_also_carries_per_angle_backgrounds(tmp_path):
|
|
from core.saw_check import middle_row_plan
|
|
out = tmp_path / "check.sras"
|
|
plan = write_v7(out, plan=middle_row_plan(tiny_plan()),
|
|
version=VERSION_SAW_CHECK)
|
|
|
|
sras = SrasFile(out)
|
|
assert sras.is_saw_check and not sras.is_legacy_layout
|
|
assert sras.backgrounds == [angle_background(ai)
|
|
for ai in range(plan.n_angles)]
|
|
sras.close()
|
|
|
|
|
|
def test_angle_missing_its_background_is_the_frontier(tmp_path):
|
|
"""A file cut inside a background block stops at that angle.
|
|
|
|
Nothing of that angle is on disk yet — not even the reference its rows
|
|
would be read against — so it is MISSING rather than TRUNCATED, and its
|
|
predicted offsets are where a resumed scan would write.
|
|
"""
|
|
out = tmp_path / "v7.sras"
|
|
write_v7(out)
|
|
whole = out.read_bytes()
|
|
bg1 = SrasFile(out).angle_status()[1].bg_offset
|
|
|
|
for cut, expected_status in ((bg1, "MISSING"), (bg1 + 4 + SPF // 2, "MISSING")):
|
|
out.write_bytes(whole[:cut])
|
|
statuses = SrasFile(out).angle_status()
|
|
assert [s.status for s in statuses] == ["OK", expected_status]
|
|
assert statuses[1].n_rows_available == 0
|
|
assert statuses[1].bg_offset == bg1
|
|
# The absent block is predicted at a full record's worth of bytes,
|
|
# which is what the writer will produce when the scan resumes.
|
|
assert statuses[1].data_offset == bg1 + 4 + SPF
|
|
|
|
|
|
def test_rows_after_a_background_still_truncate_by_row(tmp_path):
|
|
out = tmp_path / "v7.sras"
|
|
plan = write_v7(out)
|
|
whole = out.read_bytes()
|
|
st1 = SrasFile(out).angle_status()[1]
|
|
|
|
out.write_bytes(whole[:st1.data_offset + 2 * st1.row_bytes])
|
|
statuses = SrasFile(out).angle_status()
|
|
assert [s.status for s in statuses] == ["OK", "TRUNCATED"]
|
|
assert statuses[1].n_rows_available == 2
|
|
assert SrasFile(out).load_angle(1, n_rows=2).shape[0] == 2
|
|
assert plan.per_angle[1].n_rows == 3
|
|
|
|
|
|
def test_load_angle_memmap_equals_eager():
|
|
with SrasFile(GOLDEN / "complete.sras") as sras:
|
|
raw = (GOLDEN / "complete.sras").read_bytes()
|
|
for ai, pa in enumerate(sras.per_angle):
|
|
view = sras.load_angle(ai)
|
|
h = sras.header
|
|
assert view.shape == (pa.n_rows, h.n_channels, pa.n_frames, h.samples_per_frame)
|
|
start = sras.angle_data_offset(ai)
|
|
eager = np.frombuffer(
|
|
raw, dtype=np.int8, offset=start, count=view.size
|
|
).reshape(view.shape)
|
|
assert np.array_equal(view, eager)
|
|
assert not view.flags.writeable
|
|
|
|
|
|
def test_load_row_matches_synthetic_pattern():
|
|
with SrasFile(GOLDEN / "complete.sras") as sras:
|
|
for ai in range(2):
|
|
for ri in range(3):
|
|
for ci in range(3):
|
|
row = sras.load_row(ai, ri, ci)
|
|
expected_bytes = b"".join(
|
|
synthetic_frame(ai, ri, ci, fi)
|
|
for fi in range(sras.per_angle[ai].n_frames)
|
|
)
|
|
assert row.tobytes() == expected_bytes
|
|
|
|
|
|
def test_truncated_load_angle_partial_rows():
|
|
sras = SrasFile(GOLDEN / "trunc_rowboundary_a1.sras")
|
|
st = sras.angle_status()[1]
|
|
assert st.status == "TRUNCATED" and st.n_rows_available == 2
|
|
view = sras.load_angle(1, n_rows=st.n_rows_available)
|
|
assert view.shape[0] == 2
|
|
sras.close()
|
|
|
|
|
|
def test_bad_magic_and_version_rejected(tmp_path):
|
|
bad = tmp_path / "bad.sras"
|
|
bad.write_bytes(b"XXXX" + bytes(60))
|
|
with pytest.raises(ValueError, match="bad magic"):
|
|
SrasFile(bad)
|
|
|
|
data = bytearray((GOLDEN / "complete.sras").read_bytes())
|
|
data[4] = 5 # version byte
|
|
v5 = tmp_path / "v5.sras"
|
|
v5.write_bytes(bytes(data))
|
|
with pytest.raises(ValueError, match="version 5"):
|
|
SrasFile(v5)
|