ykcheungab
c377c14474
2. added unwarp map export in current cli, don't need to generate by precompute in sensor running
290 lines
12 KiB
Python
290 lines
12 KiB
Python
"""Tests for `compare`: the settings sweep and its caching.
|
|
|
|
Every test injects a fake `stage_runner`, so nothing here needs Blender, cv2 or
|
|
matplotlib. The caching behaviour is the part worth testing hardest: if a `dis_mode`
|
|
variant re-rendered instead of reusing the frames, the sweep's runtime claim would be
|
|
false and the comparison would be comparing different pixels.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import dataclasses
|
|
import json
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
|
|
from orisys_sensor_design import compare
|
|
|
|
|
|
class RecordingRunner:
|
|
"""A fake stage runner that records calls and writes minimal artifacts."""
|
|
|
|
def __init__(self, *, fail_on: str | None = None, scores: dict | None = None) -> None:
|
|
self.calls: list[tuple[str, Path, dict]] = []
|
|
self.settings_by_kind: dict[str, list] = {}
|
|
self.fail_on = fail_on
|
|
self.scores = scores or {"render": 10.0, "process": 4.0}
|
|
|
|
def __call__(self, kind: str, *, settings=None, output_dir=None, **kwargs) -> int:
|
|
output_dir = Path(output_dir)
|
|
output_dir.mkdir(parents=True, exist_ok=True)
|
|
self.calls.append((kind, output_dir, dict(kwargs)))
|
|
self.settings_by_kind.setdefault(kind, []).append(settings)
|
|
if self.fail_on and kind == self.fail_on:
|
|
return 1
|
|
if kind == "texture":
|
|
(output_dir / "patch_texture_compute.png").write_bytes(b"png")
|
|
if kind == "deform":
|
|
(output_dir / "probe_frames.npz").write_bytes(b"npz")
|
|
if kind == "render":
|
|
(output_dir / "frames").mkdir(exist_ok=True)
|
|
manifest = {
|
|
"width": 500,
|
|
"height": 500,
|
|
"frames": [
|
|
{"name": "000_reference", "axis": None, "step_mm": 0.0,
|
|
"png": "frames/000_reference.png"},
|
|
],
|
|
"null": {"name": "reference_null", "seed": 1,
|
|
"png": "frames/reference_null.png"},
|
|
}
|
|
(output_dir / "render_manifest.json").write_text(json.dumps(manifest), encoding="utf-8")
|
|
if kind == "sensitivity":
|
|
report = {
|
|
"overall": {"score": 12.0, "verdict": "resolvable"},
|
|
"scales": [
|
|
{"name": name, "score": score, "verdict": "resolvable",
|
|
"cell_px": 16.13, "effective_patch_px": 16.0}
|
|
for name, score in self.scores.items()
|
|
],
|
|
}
|
|
(output_dir / "sensitivity_report.json").write_text(json.dumps(report), encoding="utf-8")
|
|
return 0
|
|
|
|
def kinds(self) -> list[str]:
|
|
return [kind for kind, _directory, _kwargs in self.calls]
|
|
|
|
def dirs_for(self, kind: str) -> list[Path]:
|
|
return [directory for name, directory, _kwargs in self.calls if name == kind]
|
|
|
|
|
|
@dataclasses.dataclass(frozen=True)
|
|
class FakePattern:
|
|
"""Mirrors ``deform.PatternInfo``, which the sweep overrides per candidate."""
|
|
|
|
nx: int = 80
|
|
ny: int = 80
|
|
seed: int = 7
|
|
vector_width_mm: float = 18.5
|
|
vector_height_mm: float = 16.4
|
|
|
|
|
|
@dataclasses.dataclass(frozen=True)
|
|
class FakeSettings:
|
|
"""A frozen dataclass on purpose: ``run_compare`` uses ``dataclasses.replace``,
|
|
so a ``SimpleNamespace`` fixture would pass tests that the real
|
|
``deform.ProbeSettings`` path fails."""
|
|
|
|
radius_mm: float = 0.5
|
|
texture: str = "patch_texture_compute.png"
|
|
tracker: dict = dataclasses.field(
|
|
default_factory=lambda: {"dis_mode": 1, "dis_preset": "medium"}
|
|
)
|
|
resolution: tuple = (500, 500)
|
|
process_resolution: tuple = (250, 250)
|
|
pattern: FakePattern = dataclasses.field(default_factory=FakePattern)
|
|
|
|
|
|
def _settings() -> FakeSettings:
|
|
return FakeSettings()
|
|
|
|
|
|
def _run(tmp_path: Path, runner, **kwargs):
|
|
defaults = dict(
|
|
stage_runner=runner,
|
|
grid=(31, 80),
|
|
seeds=(7,),
|
|
dis_modes=(1, 0),
|
|
radii=(0.5,),
|
|
output_dir=tmp_path,
|
|
progress=lambda _message: None,
|
|
)
|
|
defaults.update(kwargs)
|
|
return compare.run_compare(SimpleNamespace(root=tmp_path), _settings(), **defaults)
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Enumeration
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
def test_enumerate_is_deterministic_and_sorted() -> None:
|
|
first = compare.enumerate_sweep(
|
|
grid=(80, 31), seeds=(7,), dis_modes=(0, 1), radii=(1.5, 0.5)
|
|
)
|
|
second = compare.enumerate_sweep(
|
|
grid=(31, 80), seeds=(7,), dis_modes=(1, 0), radii=(0.5, 1.5)
|
|
)
|
|
assert [entry.id for entry in first] == [entry.id for entry in second]
|
|
assert [entry.nx for entry in first] == sorted(entry.nx for entry in first)
|
|
# dis_mode 1 (the shipped setting) sorts before 0 within every n, rx block.
|
|
assert [entry.dis_mode for entry in first if entry.nx == 31 and entry.radius_mm == 0.5] == [1, 0]
|
|
|
|
|
|
def test_enumerate_rejects_bad_values() -> None:
|
|
with pytest.raises(compare.CompareError):
|
|
compare.enumerate_sweep(grid=(0,), seeds=(7,), dis_modes=(1,), radii=(0.5,))
|
|
with pytest.raises(compare.CompareError):
|
|
compare.enumerate_sweep(grid=(31,), seeds=(7,), dis_modes=(1,), radii=(0.0,))
|
|
with pytest.raises(compare.CompareError):
|
|
compare.enumerate_sweep(grid=(31,), seeds=(7,), dis_modes=(-1,), radii=(0.5,))
|
|
|
|
|
|
def test_dry_run_enumerates_and_writes_nothing(tmp_path: Path) -> None:
|
|
runner = RecordingRunner()
|
|
report, code = _run(tmp_path, runner, dry_run=True)
|
|
assert code == 0
|
|
assert runner.calls == []
|
|
assert len(report["runs"]) == 4 # ids, not outcomes
|
|
assert not (tmp_path / compare.COMPARISON_DIRNAME / compare.FILENAME).exists()
|
|
assert any("dry-run" in note for note in report["notes"])
|
|
|
|
|
|
def test_max_runs_refuses_a_large_matrix(tmp_path: Path) -> None:
|
|
with pytest.raises(compare.CompareError):
|
|
_run(tmp_path, RecordingRunner(), max_runs=2)
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Caching: what each setting actually invalidates
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
def test_texture_runs_once_per_grid_and_frames_once_per_radius(tmp_path: Path) -> None:
|
|
runner = RecordingRunner()
|
|
report, code = _run(tmp_path, runner)
|
|
assert code == 0
|
|
# 2 grids -> 2 textures; 2 (grid x radius) -> 2 deform + 2 render;
|
|
# 4 entries -> 4 track + 4 sensitivity.
|
|
assert runner.kinds().count("texture") == 2
|
|
assert runner.kinds().count("deform") == 2
|
|
assert runner.kinds().count("render") == 2
|
|
assert runner.kinds().count("track") == 4
|
|
assert runner.kinds().count("sensitivity") == 4
|
|
assert len(report["runs"]) == 4
|
|
|
|
|
|
def test_dis_mode_variant_reuses_frames_and_rewrites_the_manifest(tmp_path: Path) -> None:
|
|
"""The decisive experiment must not re-render: same pixels, different estimator."""
|
|
runner = RecordingRunner()
|
|
_run(tmp_path, runner)
|
|
render_dirs = runner.dirs_for("render")
|
|
assert len(render_dirs) == 2, "each (nx, radius) is rendered exactly once"
|
|
|
|
# The dm0 variant directory must exist, hold no frames of its own, and carry a
|
|
# manifest whose png entries are absolute and point into the rendered directory.
|
|
variants = [d for d in runner.dirs_for("track") if d.name.endswith("_dm0")]
|
|
assert variants, "expected dis_mode=0 variant directories"
|
|
for variant in variants:
|
|
assert not (variant / "frames").exists(), "the variant must not re-render"
|
|
manifest = json.loads((variant / "render_manifest.json").read_text(encoding="utf-8"))
|
|
for entry in list(manifest["frames"]) + [manifest["null"]]:
|
|
path = Path(entry["png"])
|
|
assert path.is_absolute()
|
|
assert path.parent.name == "frames"
|
|
# `png` is relative to the run directory, so the subdirectory must appear
|
|
# exactly once. Doubling it (`frames/frames/...`) failed every variant run
|
|
# with a missing-artifact exit 3.
|
|
assert "frames" + "\\" + "frames" not in str(path)
|
|
assert "frames/frames" not in str(path).replace("\\", "/")
|
|
|
|
|
|
def test_radius_change_reuses_the_texture_but_not_the_frames(tmp_path: Path) -> None:
|
|
runner = RecordingRunner()
|
|
_run(tmp_path, runner, radii=(0.5, 1.5))
|
|
# Two radii and two grids: still one texture per grid, but frames per radius.
|
|
assert runner.kinds().count("texture") == 2
|
|
assert runner.kinds().count("deform") == 4
|
|
assert runner.kinds().count("render") == 4
|
|
# Each deformation must carry its own radius, and the estimator its own dis_mode.
|
|
radii = {round(float(s.radius_mm), 3) for s in runner.settings_by_kind["deform"]}
|
|
assert radii == {0.5, 1.5}
|
|
dis_modes = {int(s.tracker["dis_mode"]) for s in runner.settings_by_kind["track"]}
|
|
assert dis_modes == {0, 1}
|
|
# A deform call must never be asked to redo a radius it already did.
|
|
assert runner.kinds().count("deform") == len({(s.radius_mm, s.texture) for s in runner.settings_by_kind["deform"]})
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Report shape, robustness, determinism
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
def test_report_records_per_setting_bests(tmp_path: Path) -> None:
|
|
report, code = _run(tmp_path, RecordingRunner())
|
|
assert code == 0
|
|
assert set(report) >= {
|
|
"schema", "search", "runs", "ranked", "best_per_dis_mode", "best_per_radius", "notes",
|
|
}
|
|
assert report["schema"] == compare.SCHEMA
|
|
assert set(report["best_per_dis_mode"]) == {"0", "1"}
|
|
assert set(report["best_per_radius"]) == {"0.5"}
|
|
assert report["ranked"]
|
|
# Notes must say the effective window is what dis_mode changes.
|
|
joined = " ".join(report["notes"])
|
|
assert "finest pyramid level" in joined
|
|
assert "patch_px * 2**dis_mode" in joined
|
|
|
|
|
|
def test_report_is_written_and_parses(tmp_path: Path) -> None:
|
|
_run(tmp_path, RecordingRunner())
|
|
path = tmp_path / compare.COMPARISON_DIRNAME / compare.FILENAME
|
|
assert path.is_file()
|
|
parsed = json.loads(path.read_text(encoding="utf-8"))
|
|
assert parsed["schema"] == compare.SCHEMA
|
|
|
|
|
|
def test_a_failing_run_is_recorded_without_aborting_the_sweep(tmp_path: Path) -> None:
|
|
runner = RecordingRunner(fail_on="sensitivity")
|
|
report, code = _run(tmp_path, runner)
|
|
assert all(run["error"] for run in report["runs"])
|
|
assert code == 1
|
|
assert any("failed and are recorded with an error" in note for note in report["notes"])
|
|
|
|
|
|
def test_two_runs_give_identical_reports(tmp_path: Path) -> None:
|
|
first, _ = _run(tmp_path, RecordingRunner())
|
|
second, _ = _run(tmp_path, RecordingRunner())
|
|
first.pop("revision", None)
|
|
second.pop("revision", None)
|
|
assert json.dumps(first, sort_keys=True) == json.dumps(second, sort_keys=True)
|
|
|
|
|
|
def test_build_figures_writes_three_figures(tmp_path: Path) -> None:
|
|
pytest.importorskip("matplotlib")
|
|
runner = RecordingRunner()
|
|
report, _code = _run(tmp_path, runner)
|
|
# Give one run real flow data so a panel can be drawn rather than placeholdered.
|
|
import numpy as np
|
|
|
|
for directory in runner.dirs_for("track")[:1]:
|
|
flow_dir = directory / "flow" / "render"
|
|
flow_dir.mkdir(parents=True, exist_ok=True)
|
|
np.save(flow_dir / "000_reference.npy", np.zeros((4, 4, 2), dtype=np.float32))
|
|
figures = compare.build_figures(
|
|
report, tmp_path / compare.COMPARISON_DIRNAME, progress=lambda _m: None
|
|
)
|
|
assert "tuning" in figures
|
|
assert figures["tuning"].is_file()
|
|
assert figures["tuning"].stat().st_size > 0
|
|
|
|
|
|
def test_build_figures_without_scored_runs_writes_nothing(tmp_path: Path) -> None:
|
|
pytest.importorskip("matplotlib")
|
|
figures = compare.build_figures(
|
|
{"runs": []}, tmp_path, progress=lambda _m: None
|
|
)
|
|
assert figures == {}
|