Thomas Ales
afe33249d1
The headline of the refactor. Scan orchestration no longer lives inside a QObject that reaches through Qt workers for its hardware handles. core/scan_engine.py — ScanEngine(stage, scope, rotator, plan, out_path, resume, callbacks). Takes the concrete drivers, blocks in run(), reports via plain callables, and prompts through an injected blocking callable. No Qt import anywhere in the path (test_engine_imports_without_qt proves it), so a simpler GUI or a CLI can drive the identical acquisition. Supporting extractions, all Qt-free: - core/scope_sras.py — SCPI policy: channel profiles, trigger programming, background average, per-row FastFrame transfer - core/rotation.py — RotationAxis + RotationSettings (the GR_* constants) - core/scan_resume.py — frontier contiguity rule + settings compatibility - gui/scan_bridge.py — QtScanController, exposing exactly the signal surface the old ScanWorker had, so MainWindow's connections are unchanged hardware/t3r_driver.py is now Qt-free: a plain Signal class, a threading reader, and a polling thread instead of QObject/QThread/QTimer. gui/qt_t3r.py re-emits its callbacks as queued Qt signals for the panels. Fixes carried by the extraction: - rotation waits on the driver's MOTION_DONE event instead of time.sleep(estimate + 0.5) - abort during an operator prompt now takes effect; the old _prompt_event.wait() had no timeout and could not be interrupted - the poll timer is a thread, so an I/O error tearing down the driver no longer calls QTimer.stop() from the wrong thread - T3RDriver.disconnect() renamed close(); it shadowed QObject.disconnect() - per-frame DC means use np.frombuffer over the joined block instead of struct.unpack per frame (~16k tuple allocations per row) tests/fakes.py + test_scan_engine.py (14 tests) assert the exact command sequence, file layout, resume seeking, abort/pause, and geometry rejection before any hardware call; test_scan_resume.py covers the frontier rule. 58 passing. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
78 lines
2.8 KiB
Python
78 lines
2.8 KiB
Python
"""core.scan_resume: the frontier contiguity rule, over real fixture files."""
|
|
from pathlib import Path
|
|
|
|
from core.scan_resume import is_compatible, plan_resume
|
|
from core.sras_format import SrasFile
|
|
|
|
GOLDEN = Path(__file__).parent / "golden"
|
|
|
|
|
|
def _statuses(name):
|
|
return SrasFile(GOLDEN / name).angle_status()
|
|
|
|
|
|
def test_complete_file_frontier_is_past_the_end():
|
|
st = _statuses("complete.sras")
|
|
plan = plan_resume(st, selected={0})
|
|
assert plan.frontier_idx == len(st)
|
|
assert [t.angle_idx for t in plan.targets] == [0]
|
|
assert plan.auto_added == []
|
|
|
|
|
|
def test_selecting_past_frontier_backfills_the_gap():
|
|
# angle 0 complete, angle 1 truncated → frontier = 1
|
|
st = _statuses("trunc_rowboundary_a1.sras")
|
|
assert st[0].complete and not st[1].complete
|
|
|
|
plan = plan_resume(st, selected={1})
|
|
assert plan.frontier_idx == 1
|
|
assert [t.angle_idx for t in plan.targets] == [1]
|
|
assert plan.auto_added == []
|
|
|
|
|
|
def test_selection_before_frontier_is_untouched():
|
|
st = _statuses("trunc_rowboundary_a1.sras")
|
|
plan = plan_resume(st, selected={0})
|
|
assert [t.angle_idx for t in plan.targets] == [0]
|
|
assert plan.auto_added == []
|
|
|
|
|
|
def test_selecting_only_a_later_angle_pulls_in_the_frontier():
|
|
"""Data is one contiguous stream, so angle 1 can't be skipped to reach 2."""
|
|
st = _statuses("header_only.sras") # nothing written: frontier = 0
|
|
assert plan_resume(st, selected={0}).frontier_idx == 0
|
|
|
|
plan = plan_resume(st, selected={1})
|
|
assert [t.angle_idx for t in plan.targets] == [0, 1]
|
|
assert plan.auto_added == [0]
|
|
|
|
|
|
def test_targets_carry_offsets_and_rows():
|
|
st = _statuses("complete.sras")
|
|
plan = plan_resume(st, selected={0, 1})
|
|
for target, status in zip(plan.targets, st, strict=True):
|
|
assert target.data_offset == status.data_offset
|
|
assert target.n_rows == status.n_rows
|
|
assert target.angle_deg == status.angle_deg
|
|
assert plan.total_rows == sum(s.n_rows for s in st)
|
|
|
|
|
|
def test_to_state_carries_samples_per_frame():
|
|
sras = SrasFile(GOLDEN / "complete.sras")
|
|
state = plan_resume(sras.angle_status(), selected={0}).to_state(sras)
|
|
assert state.path == sras.path
|
|
assert state.samples_per_frame == sras.header.samples_per_frame
|
|
assert state.target_indices == {0}
|
|
|
|
|
|
def test_is_compatible_checks_acquisition_settings():
|
|
sras = SrasFile(GOLDEN / "complete.sras")
|
|
h = sras.header
|
|
ok = dict(velocity=h.velocity, laser_freq=h.laser_freq,
|
|
sample_rate=h.sample_rate, n_channels=h.n_channels)
|
|
assert is_compatible(sras, **ok)
|
|
assert not is_compatible(sras, **{**ok, "velocity": h.velocity + 1})
|
|
assert not is_compatible(sras, **{**ok, "laser_freq": h.laser_freq * 2})
|
|
assert not is_compatible(sras, **{**ok, "sample_rate": h.sample_rate * 2})
|
|
assert not is_compatible(sras, **{**ok, "n_channels": h.n_channels + 1})
|