orisys-tech/orisys-sensor-design
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
5bd7dd7a0157e32abd6bdf47bdfa7d1ab604eacf
orisys-sensor-design/tests/test_optimise.py
T
ykcheungab c377c14474 1. added blender deformation probe for marker size picking and optimization 
2. added unwarp map export in current cli, don't need to generate by precompute in sensor running
2026-09-25 15:44:44 +08:00

504 lines
20 KiB
Python
Raw Blame History

"""Tests for the deterministic marker-pattern optimizer.
No Blender and no cv2: the per-candidate pipeline is injected, so these tests
exercise enumeration, ranking, winner selection, deterministic ordering, the
full-resolution re-confirmation and the dry-run path directly.
"""
from __future__ import annotations
import json
from pathlib import Path
from types import SimpleNamespace
import pytest
from orisys_sensor_design import optimise as opt
def _paths(tmp_path: Path) -> SimpleNamespace:
revision = tmp_path / "rev"
revision.mkdir(exist_ok=True)
return SimpleNamespace(root=revision)
def _settings(**overrides: object) -> SimpleNamespace:
values = {
"resolution": [500, 500],
"process_resolution": [250, 250],
"step_mm": 0.01,
}
values.update(overrides)
return SimpleNamespace(**values)
class RecordingRunner:
"""Deterministic fake pipeline: score depends only on the candidate."""
def __init__(
self,
scores: dict[str, float] | None = None,
confirm_factor: float = 0.9,
scale_scores: dict[str, dict[str, float]] | None = None,
):
self.calls: list[tuple[str, tuple[int, int], str, Path]] = []
self.scores = scores or {}
self.confirm_factor = confirm_factor
self.scale_scores = scale_scores or {}
def __call__(self, run: opt.CandidateRun) -> dict:
self.calls.append((run.candidate.id, run.resolution, run.stage, run.directory))
run.directory.mkdir(parents=True, exist_ok=True)
score = self.scores.get(run.candidate.id, float(run.candidate.nx))
if run.stage == "confirm":
score = score * self.confirm_factor
scales = {
name: {
"score": value * (self.confirm_factor if run.stage == "confirm" else 1.0),
"verdict": "marginal",
}
for name, value in (self.scale_scores.get(run.candidate.id) or {}).items()
}
return {
"score": score,
"verdict": "marginal",
"cell_px": None,
"px_per_mm": 6.624,
"report": str(run.directory / "sensitivity_report.json"),
"texture_sha256": f"sha-{run.candidate.id}",
"scales": scales,
}
class ReportWritingRunner(RecordingRunner):
"""Runner that returns only a report path, like cli.py's real runner."""
def __call__(self, run: opt.CandidateRun) -> dict:
result = super().__call__(run)
result.pop("scales", None)
report_path = Path(result["report"])
report_path.parent.mkdir(parents=True, exist_ok=True)
report_path.write_text(
json.dumps(
{
"schema": "orisys-sensor-design/sensitivity-report-v1",
"scales": [
{"name": "render", "score": 30.0 + run.candidate.nx, "verdict": "resolvable"},
{"name": "process", "score": 40.0 - run.candidate.nx, "verdict": "marginal"},
],
}
),
encoding="utf-8",
)
return result
# --------------------------------------------------------------------------- #
# Enumeration
# --------------------------------------------------------------------------- #
def test_enumeration_is_sorted_and_square_by_default() -> None:
candidates = opt.enumerate_candidates([31, 15, 20], [7])
assert [candidate.id for candidate in candidates] == ["15@7", "20@7", "31@7"]
assert all(candidate.nx == candidate.ny for candidate in candidates)
def test_enumeration_is_independent_of_input_order() -> None:
forward = opt.enumerate_candidates([15, 20, 31], [7, 3])
backward = opt.enumerate_candidates([31, 15, 20], [3, 7])
assert forward == backward
assert [candidate.id for candidate in forward] == [
"15@3", "15@7", "20@3", "20@7", "31@3", "31@7"
]
def test_enumeration_deduplicates_and_honours_grid_y() -> None:
candidates = opt.enumerate_candidates([20, 20], [7, 7], grid_y=[20, 30])
assert [candidate.id for candidate in candidates] == ["20@7", "20x30@7"]
def test_enumeration_is_bounded_by_the_grid_and_seed_sets() -> None:
candidates = opt.enumerate_candidates(opt.DEFAULT_GRID, [7])
assert len(candidates) == len(opt.DEFAULT_GRID)
assert candidates[0].nx == 15
assert candidates[-1].nx == 80
# nx ~ 28-36 is inside the default grid: it brackets the DIS patch size.
assert any(28 <= candidate.nx <= 36 for candidate in candidates)
@pytest.mark.parametrize(
"grid,seeds",
[
([], [7]),
([20], []),
([0], [7]),
([-3], [7]),
([20], [-1]),
(["x"], [7]),
(None, [7]),
],
)
def test_invalid_search_bounds_raise(grid: object, seeds: object) -> None:
with pytest.raises(opt.OptimiseError):
opt.enumerate_candidates(grid, seeds) # type: ignore[arg-type]
# --------------------------------------------------------------------------- #
# Ranking
# --------------------------------------------------------------------------- #
def _entries(scores: dict[tuple[int, int, int], float | None]) -> list[dict]:
return [
{"candidate": opt.Candidate(nx, ny, seed), "score": score, "verdict": "marginal"}
for (nx, ny, seed), score in scores.items()
]
def test_ranking_is_stable_and_independent_of_input_order() -> None:
scores = {(31, 31, 7): 0.4, (20, 20, 7): 0.9, (36, 36, 7): 0.9, (48, 48, 7): 0.1}
ordered = _entries(dict(sorted(scores.items())))
shuffled = _entries(dict(reversed(list(scores.items()))))
first = [entry["candidate"].id for entry in opt.rank_entries(ordered)]
second = [entry["candidate"].id for entry in opt.rank_entries(shuffled)]
assert first == second == ["20@7", "36@7", "31@7", "48@7"]
def test_unscored_candidates_sort_last() -> None:
ranked = opt.rank_entries(
_entries({(31, 31, 7): 0.4, (20, 20, 7): None, (36, 36, 7): 0.9})
)
assert [entry["candidate"].id for entry in ranked] == ["36@7", "31@7", "20@7"]
def test_ties_break_on_candidate_order() -> None:
ranked = opt.rank_entries(_entries({(36, 36, 7): 0.5, (20, 20, 7): 0.5, (48, 48, 7): 0.5}))
assert [entry["candidate"].id for entry in ranked] == ["20@7", "36@7", "48@7"]
# --------------------------------------------------------------------------- #
# run_optimise: winner, confirmation, determinism, notes
# --------------------------------------------------------------------------- #
def _run(tmp_path: Path, runner: object, **kwargs: object) -> tuple[dict, int]:
kwargs.setdefault("top", None)
return opt.run_optimise(
_paths(tmp_path),
_settings(),
grid=[20, 31, 48],
seeds=[7],
search_resolution=(250, 250),
output_dir=tmp_path / "out",
progress=lambda message: None,
texture_vector_mm=(18.5, 16.4),
runner=runner, # type: ignore[arg-type]
**kwargs,
)
def test_winner_is_the_top_ranked_candidate_and_confirmed_at_full_resolution(
tmp_path: Path,
) -> None:
runner = RecordingRunner(scores={"20@7": 0.2, "31@7": 0.9, "48@7": 0.5})
report, code = _run(tmp_path, runner)
assert code == 0
assert report["ranked"][0] == "31@7"
assert report["winner"]["nx"] == 31 and report["winner"]["seed"] == 7
assert report["winner"]["score"] == pytest.approx(0.9)
assert report["winner"]["confirmed_at_resolution"] == [500, 500]
assert report["winner"]["confirmed_score"] == pytest.approx(0.9 * 0.9)
assert report["winner"]["report"]
search_calls = [call for call in runner.calls if call[2] == "search"]
confirm_calls = [call for call in runner.calls if call[2] == "confirm"]
assert len(search_calls) == 3
assert all(call[1] == (250, 250) for call in search_calls)
assert len(confirm_calls) == 1
assert confirm_calls[0][0] == "31@7"
assert confirm_calls[0][1] == (500, 500), "re-confirmation must use probe.resolution"
def test_report_never_claims_a_global_optimum(tmp_path: Path) -> None:
runner = RecordingRunner()
report, _ = _run(tmp_path, runner)
notes = report["notes"]
assert any("not a global optimum" in note for note in notes)
for note in notes:
if "global optimum" in note:
assert "not a global optimum" in note, note
assert report["search"]["objective"] == "sensitivity_report.overall.score"
# The grid rationale must say it was chosen around the DIS patch, not nx=80.
assert any("DIS patch" in note for note in notes)
def test_optimizer_notes_record_the_roi_confound(tmp_path: Path) -> None:
report, _ = _run(tmp_path, RecordingRunner())
joined = " ".join(report["notes"])
assert "confounded with the ROI size" in joined
assert "Only nx/ny/seed vary" in joined
assert "radius_mm, the ROI and the scale are held fixed" in joined
def test_notes_no_longer_assert_aliasing_for_the_shipped_pattern(tmp_path: Path) -> None:
report, _ = _run(tmp_path, RecordingRunner())
joined = " ".join(report["notes"])
# The disproved claim must be gone.
assert "aliases a sub-period shift" not in joined
assert "wrong period" not in joined
# What was actually measured must be there instead.
assert "does NOT alias" in joined
# The rationale must now reference the effective window and the camera-space
# cell, not the old millimetre-derived cell or the nominal patch alone.
assert "effective" in joined
assert "width/nx" in joined
assert "setFinestScale" in joined
def test_notes_state_that_best_is_scale_dependent(tmp_path: Path) -> None:
report, _ = _run(tmp_path, RecordingRunner())
joined = " ".join(report["notes"])
assert "'Best' is scale-dependent" in joined
assert "unweighted mean" in joined
assert "scale_scores" in joined
def test_notes_report_the_search_to_confirm_score_gap(tmp_path: Path) -> None:
report, _ = _run(tmp_path, RecordingRunner())
joined = " ".join(report["notes"])
assert "score gap between the search and the confirmation" in joined
assert "ratio" in joined
def test_unset_search_resolution_defaults_to_settings_resolution(tmp_path: Path) -> None:
runner = RecordingRunner()
report, code = opt.run_optimise(
_paths(tmp_path),
_settings(resolution=[500, 500]),
grid=[20, 31],
seeds=[7],
top=None,
output_dir=tmp_path / "out",
progress=lambda message: None,
texture_vector_mm=(18.5, 16.4),
runner=runner,
)
assert code == 0
assert report["search"]["render_resolution"] == [500, 500]
search_resolutions = {call[1] for call in runner.calls if call[2] == "search"}
assert search_resolutions == {(500, 500)}, "search must default to settings.resolution"
# The confirmation still uses settings.resolution too.
assert report["winner"]["confirmed_at_resolution"] == [500, 500]
def test_unset_search_resolution_falls_back_when_settings_carry_none(
tmp_path: Path,
) -> None:
runner = RecordingRunner()
report, code = opt.run_optimise(
_paths(tmp_path),
SimpleNamespace(step_mm=0.01), # no resolution at all
grid=[20],
seeds=[7],
top=None,
output_dir=tmp_path / "out",
progress=lambda message: None,
runner=runner,
)
assert code == 0
assert report["search"]["render_resolution"] == list(opt.DEFAULT_SEARCH_RESOLUTION)
def test_winner_is_confirmed_at_settings_resolution_by_default(tmp_path: Path) -> None:
report, _ = _run(
tmp_path, RecordingRunner(scores={"20@7": 0.2, "31@7": 0.9, "48@7": 0.5})
)
# `_run` passes search_resolution=(250,250) explicitly; the confirmation target
# still comes from settings.resolution and must stay 500x500.
assert report["winner"]["confirmed_at_resolution"] == [500, 500]
assert report["search"]["render_resolution"] == [250, 250]
def test_candidate_rows_carry_per_scale_scores_and_ranks(tmp_path: Path) -> None:
scale_scores = {
"20@7": {"render": 0.1, "process": 0.9},
"31@7": {"render": 0.3, "process": 0.5},
"48@7": {"render": 0.9, "process": 0.1},
}
runner = RecordingRunner(
scores={"20@7": 0.2, "31@7": 0.9, "48@7": 0.5},
scale_scores=scale_scores,
)
report, _ = _run(tmp_path, runner)
assert report["search"]["scales"] == ["process", "render"]
# Overall winner is the mean's winner ...
assert report["ranked"][0] == "31@7"
# ... but it is NOT the render-scale winner, which is the point of the dimension.
assert report["ranked_by_scale"]["render"] == ["48@7", "31@7", "20@7"]
assert report["ranked_by_scale"]["process"] == ["20@7", "31@7", "48@7"]
by_nx = {row["nx"]: row for row in report["candidates"]}
assert by_nx[48]["scale_scores"]["render"] == pytest.approx(0.9)
assert by_nx[48]["scale_ranks"]["render"] == 1
assert by_nx[31]["scale_ranks"]["render"] == 2
assert by_nx[20]["scale_ranks"]["process"] == 1
assert by_nx[20]["scale_verdicts"]["render"] == "marginal"
assert by_nx[31]["scale_ranks"]["process"] == 2
assert report["winner"]["scale_scores"]["render"] == pytest.approx(0.3)
assert report["search"]["render_resolution"] == [250, 250]
def test_per_scale_scores_are_read_from_the_candidate_report_path(tmp_path: Path) -> None:
runner = ReportWritingRunner()
report, code = _run(tmp_path, runner)
assert code == 0
by_nx = {row["nx"]: row for row in report["candidates"]}
assert by_nx[31]["scale_scores"]["render"] == pytest.approx(61.0)
assert by_nx[31]["scale_scores"]["process"] == pytest.approx(9.0)
assert report["ranked_by_scale"]["render"][0] == "48@7"
assert report["ranked_by_scale"]["process"][0] == "20@7"
def test_candidate_table_is_sorted_and_carries_measured_fields(tmp_path: Path) -> None:
runner = RecordingRunner(scores={"20@7": 0.2, "31@7": 0.9, "48@7": 0.5})
report, _ = _run(tmp_path, runner)
rows = report["candidates"]
assert [row["nx"] for row in rows] == [31, 48, 20]
assert report["ranked"] == ["31@7", "48@7", "20@7"]
for row in rows:
assert set(row) >= {"nx", "ny", "seed", "score", "verdict", "cell_px",
"report", "texture_sha256", "scale_scores",
"scale_verdicts", "scale_ranks"}
assert report["search"]["candidates"] == 3
assert report["search"]["render_resolution"] == [250, 250]
def test_cell_px_is_the_camera_space_cell(tmp_path: Path) -> None:
"""The pattern grid is defined across the camera image, so cell = width / nx.
This replaces an earlier expectation of `(18.5/31) * px_per_mm`, which derived
the cell from the *physical* page cell times the probe-local scale. That is a
different (and ~2x smaller) number: the probe-local scale is 13.2 px/mm while
the camera-grid scale is 500/18.5 = 27.0 px/mm. Using the millimetre cell made
a 16.1 px cell look like 7.9 px and hid the effective-window rule.
"""
runner = RecordingRunner()
report, _ = _run(tmp_path, runner)
row = next(item for item in report["candidates"] if item["nx"] == 31)
assert row["cell_px"] == pytest.approx(500.0 / 31.0, rel=1e-9)
# And it must NOT be the millimetre-derived value.
assert row["cell_px"] != pytest.approx((18.5 / 31) * 6.624 * 2.0, rel=1e-3)
def test_two_runs_give_byte_identical_rankings(tmp_path: Path) -> None:
first, _ = _run(tmp_path, RecordingRunner(scores={"20@7": 0.2, "31@7": 0.9, "48@7": 0.5}))
second, _ = _run(tmp_path, RecordingRunner(scores={"20@7": 0.2, "31@7": 0.9, "48@7": 0.5}))
first.pop("generated_at")
second.pop("generated_at")
assert first["ranked"] == second["ranked"]
assert first["candidates"] == second["candidates"]
assert json.dumps(first["candidates"], sort_keys=True) == json.dumps(
second["candidates"], sort_keys=True
)
def test_top_truncates_ranked_but_keeps_the_winner(tmp_path: Path) -> None:
runner = RecordingRunner(scores={"20@7": 0.2, "31@7": 0.9, "48@7": 0.5})
report, _ = _run(tmp_path, runner, top=2)
assert report["ranked"] == ["31@7", "48@7"]
assert report["winner"]["nx"] == 31
def test_runner_failure_records_error_without_aborting(tmp_path: Path) -> None:
class Flaky(RecordingRunner):
def __call__(self, run: opt.CandidateRun) -> dict:
if run.candidate.nx == 31 and run.stage == "search":
raise RuntimeError("blender exploded")
return super().__call__(run)
runner = Flaky(scores={"20@7": 0.2, "48@7": 0.5})
report, code = _run(tmp_path, runner)
assert code == 0
assert report["winner"]["nx"] == 48, "a failed candidate must not win"
bad = next(item for item in report["candidates"] if item["nx"] == 31)
assert bad["score"] is None
assert "blender exploded" in bad["error"]
def test_all_candidates_failing_returns_nonzero(tmp_path: Path) -> None:
class Broken:
def __call__(self, run: opt.CandidateRun) -> dict:
raise RuntimeError("no pipeline")
report, code = _run(tmp_path, Broken())
assert code == 1
assert report["winner"] is None
assert all(row["score"] is None for row in report["candidates"])
def test_missing_runner_writes_a_degraded_report_and_returns_2(tmp_path: Path) -> None:
report, code = opt.run_optimise(
_paths(tmp_path),
_settings(),
grid=[20, 31],
seeds=[7],
search_resolution=(250, 250),
top=None,
output_dir=tmp_path / "out",
progress=lambda message: None,
runner=None,
)
assert code == 2
assert "runner" in report["error"]["message"]
assert (tmp_path / "out" / opt.OPTIMIZER_REPORT_FILENAME).is_file()
assert any("not a global optimum" in note for note in report["notes"])
# --------------------------------------------------------------------------- #
# Dry run
# --------------------------------------------------------------------------- #
def test_dry_run_enumerates_without_running_or_writing_candidate_dirs(
tmp_path: Path,
) -> None:
runner = RecordingRunner()
report, code = opt.run_optimise(
_paths(tmp_path),
_settings(),
grid=[15, 20, 31],
seeds=[7, 3],
search_resolution=(250, 250),
top=None,
dry_run=True,
output_dir=tmp_path / "out",
progress=lambda message: None,
runner=runner,
)
assert code == 0
assert runner.calls == [], "dry-run must not run the pipeline"
assert not (tmp_path / "out" / "candidates").exists()
assert report["search"]["dry_run"] is True
assert report["search"]["candidates"] == 6
assert [row["nx"] for row in report["candidates"]] == [15, 15, 20, 20, 31, 31]
assert report["ranked"] == []
assert report["ranked_by_scale"] == {}
assert report["winner"] is None
assert all(row["score"] is None for row in report["candidates"])
assert (tmp_path / "out" / opt.OPTIMIZER_REPORT_FILENAME).is_file()
assert any("dry-run" in note for note in report["notes"])
assert any("scale-dependent" in note for note in report["notes"])
def test_optimizer_report_schema_keys(tmp_path: Path) -> None:
report, _ = _run(tmp_path, RecordingRunner())
assert report["schema"] == opt.OPTIMIZER_REPORT_SCHEMA
assert set(report) >= {"schema", "generated_at", "search", "candidates", "ranked",
"ranked_by_scale", "winner", "notes"}
assert set(report["search"]) >= {"grid", "seeds", "render_resolution", "candidates",
"objective", "scales"}
assert set(report["winner"]) >= {"nx", "ny", "seed", "score",
"confirmed_at_resolution", "confirmed_score", "report"}
Reference in New Issue View Git Blame Copy Permalink