ykcheungab
c377c14474
2. added unwarp map export in current cli, don't need to generate by precompute in sensor running
400 lines
14 KiB
Python
400 lines
14 KiB
Python
"""Tests for the noise model: real idle calibration, injected noise, retention.
|
|
|
|
These cover the correction that matters most in this bench. The render-seed floor was
|
|
being used as a noise model, its ordering reverses under additive noise, and it
|
|
understates the real idle floor measured from hardware. Everything here needs only
|
|
numpy and cv2 — no Blender — so the noise path is testable on its own.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
|
|
import numpy as np
|
|
import pytest
|
|
|
|
from orisys_sensor_design import compare, floorcal, flow, sensitivity
|
|
|
|
cv2 = pytest.importorskip("cv2")
|
|
|
|
|
|
def _pattern(size: int = 96, cell: int = 8, seed: int = 3) -> np.ndarray:
|
|
rng = np.random.RandomState(seed)
|
|
grid = rng.rand(size // cell + 1, size // cell + 1) * 255.0
|
|
return cv2.resize(grid, (size, size), interpolation=cv2.INTER_NEAREST).astype(np.uint8)
|
|
|
|
|
|
def _write_frames(directory: Path, frames: list[np.ndarray]) -> Path:
|
|
directory.mkdir(parents=True, exist_ok=True)
|
|
for index, frame in enumerate(frames):
|
|
assert cv2.imwrite(str(directory / f"{index:04d}.png"), frame)
|
|
return directory
|
|
|
|
|
|
def _noisy_clip(directory: Path, sigma: float, count: int = 8, seed: int = 11) -> Path:
|
|
base = _pattern()
|
|
rng = np.random.RandomState(seed)
|
|
frames = []
|
|
for _ in range(count):
|
|
noisy = base.astype(np.float32) + rng.normal(0.0, sigma, base.shape)
|
|
frames.append(np.clip(noisy, 0, 255).astype(np.uint8))
|
|
return _write_frames(directory, frames)
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Real idle calibration
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
def test_idle_floor_grows_with_noise(tmp_path: Path) -> None:
|
|
quiet = floorcal.measure_idle_floor(_noisy_clip(tmp_path / "q", 1.0), max_frames=8)
|
|
loud = floorcal.measure_idle_floor(_noisy_clip(tmp_path / "l", 8.0), max_frames=8)
|
|
assert quiet.floor_px_short >= 0.0
|
|
assert loud.floor_px_short > quiet.floor_px_short
|
|
assert quiet.frames == 8
|
|
|
|
|
|
def test_idle_floor_reports_a_dominant_period(tmp_path: Path) -> None:
|
|
measured = floorcal.measure_idle_floor(_noisy_clip(tmp_path / "p", 2.0), max_frames=8)
|
|
assert measured.dominant_period_px > 0
|
|
assert measured.floor_per_period_short > 0
|
|
payload = measured.to_json()
|
|
assert payload["roi"]["pixels"] > 0
|
|
assert "short = consecutive frames" in payload["basis"]
|
|
|
|
|
|
def test_drift_shows_as_a_long_over_short_ratio(tmp_path: Path) -> None:
|
|
"""A drifting scene must show long > short; white noise must not."""
|
|
base = _pattern()
|
|
drifting = []
|
|
for index in range(8):
|
|
matrix = np.float32([[1, 0, 0.35 * index], [0, 1, 0]])
|
|
drifting.append(
|
|
cv2.warpAffine(base, matrix, (base.shape[1], base.shape[0]),
|
|
flags=cv2.INTER_LINEAR, borderMode=cv2.BORDER_REPLICATE)
|
|
)
|
|
drift = floorcal.measure_idle_floor(_write_frames(tmp_path / "d", drifting), max_frames=8)
|
|
white = floorcal.measure_idle_floor(_noisy_clip(tmp_path / "w", 4.0), max_frames=8)
|
|
assert drift.drift_ratio > 1.0
|
|
assert white.drift_ratio < drift.drift_ratio
|
|
|
|
|
|
def test_measure_idle_floor_rejects_missing_and_tiny_sources(tmp_path: Path) -> None:
|
|
with pytest.raises(floorcal.FloorCalibrationError):
|
|
floorcal.measure_idle_floor(tmp_path / "nope.mp4")
|
|
single = _write_frames(tmp_path / "one", [_pattern()])
|
|
with pytest.raises(floorcal.FloorCalibrationError):
|
|
floorcal.measure_idle_floor(single)
|
|
|
|
|
|
def test_resolve_idle_source_tolerates_absence(tmp_path: Path) -> None:
|
|
assert floorcal.resolve_idle_source(tmp_path, None) is None
|
|
assert floorcal.resolve_idle_source(tmp_path, "missing.mp4") is None
|
|
present = tmp_path / "clip.mp4"
|
|
present.write_bytes(b"not really a video")
|
|
assert floorcal.resolve_idle_source(tmp_path, "clip.mp4") == present.resolve()
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Injected noise
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
def test_inject_noise_is_deterministic_and_zero_sigma_is_identity() -> None:
|
|
base = _pattern()
|
|
assert np.array_equal(flow.inject_noise(base, 0.0, 1), base)
|
|
first = flow.inject_noise(base, 5.0, 7)
|
|
second = flow.inject_noise(base, 5.0, 7)
|
|
assert np.array_equal(first, second)
|
|
assert not np.array_equal(first, flow.inject_noise(base, 5.0, 8))
|
|
assert first.dtype == np.uint8
|
|
|
|
|
|
def _axis_frames(size: int = 96) -> dict:
|
|
base = _pattern(size)
|
|
frames = {}
|
|
for axis, matrix in (
|
|
("x", np.float32([[1, 0, 0.8], [0, 1, 0]])),
|
|
("y", np.float32([[1, 0, 0], [0, 1, 0.8]])),
|
|
):
|
|
steps = []
|
|
for step_mm in (0.01, 0.05, 0.10):
|
|
shifted = cv2.warpAffine(
|
|
base, np.float32(matrix * (1, 1, step_mm / 0.10)), (size, size),
|
|
flags=cv2.INTER_LINEAR, borderMode=cv2.BORDER_REPLICATE,
|
|
)
|
|
steps.append((step_mm, shifted))
|
|
frames[axis] = steps
|
|
return {"base": base, "frames": frames}
|
|
|
|
|
|
def test_noise_model_pass_reports_gain_floor_and_samples() -> None:
|
|
data = _axis_frames()
|
|
reference = data["base"]
|
|
mask, _ = flow.region_masks(reference.shape, (48.0, 48.0), 20.0)
|
|
result = flow.noise_model_pass(
|
|
reference=reference,
|
|
axis_frames=data["frames"],
|
|
directions={"x": (1.0, 0.0), "y": (0.0, 1.0)},
|
|
roi_mask=mask,
|
|
config={"dis_preset": "medium", "dis_mode": 1},
|
|
sigmas=(2.0, 8.0),
|
|
draws=1,
|
|
)
|
|
assert result["sigmas"] == [2.0, 8.0]
|
|
for axis in ("x", "y"):
|
|
entries = result["axes"][axis]
|
|
assert [entry["sigma"] for entry in entries] == [2.0, 8.0]
|
|
for entry in entries:
|
|
assert entry["gain_px_per_mm"] is not None
|
|
assert entry["floor_px"] is not None and entry["floor_px"] > 0.0
|
|
assert entry["floor_samples"] == 1
|
|
assert entry["points"] == 3
|
|
# A larger sigma must not produce a quieter floor.
|
|
x_entries = {entry["sigma"]: entry for entry in result["axes"]["x"]}
|
|
assert x_entries[8.0]["floor_px"] >= x_entries[2.0]["floor_px"]
|
|
|
|
|
|
def test_noise_model_pass_is_deterministic() -> None:
|
|
data = _axis_frames()
|
|
mask, _ = flow.region_masks(data["base"].shape, (48.0, 48.0), 20.0)
|
|
kwargs = dict(
|
|
reference=data["base"],
|
|
axis_frames=data["frames"],
|
|
directions={"x": (1.0, 0.0)},
|
|
roi_mask=mask,
|
|
config={"dis_preset": "medium", "dis_mode": 1},
|
|
sigmas=(4.0,),
|
|
draws=1,
|
|
)
|
|
first = flow.noise_model_pass(**kwargs)
|
|
second = flow.noise_model_pass(**kwargs)
|
|
assert json.dumps(first, sort_keys=True) == json.dumps(second, sort_keys=True)
|
|
|
|
|
|
def test_resolve_noise_settings_defaults_and_overrides() -> None:
|
|
default = flow._resolve_noise_settings({})
|
|
assert default["enabled"] is True
|
|
assert default["scales"] == ["render"]
|
|
assert default["sigmas"] == list(flow.DEFAULT_NOISE_SIGMAS)
|
|
|
|
off = flow._resolve_noise_settings(
|
|
{"settings": {"noise": {"models": ["render_seed"], "sigmas": [4.0]}}}
|
|
)
|
|
assert off["enabled"] is False
|
|
|
|
bad = flow._resolve_noise_settings(
|
|
{"settings": {"noise": {"sigmas": [-1.0, 0.0, 999.0], "draws": 0}}}
|
|
)
|
|
assert bad["sigmas"] == list(flow.DEFAULT_NOISE_SIGMAS)
|
|
assert bad["draws"] == 1
|
|
|
|
|
|
def test_track_job_carries_the_noise_block(tmp_path: Path) -> None:
|
|
"""Regression: `build_track_job` omitted `noise`, so every --noise-* flag and the
|
|
design.yaml block were silently ignored and the pass ran on built-in defaults (which
|
|
happened to match, hiding it). The job must carry what the caller asked for."""
|
|
settings = SimpleNamespace(
|
|
resolution=(64, 64),
|
|
process_resolution=(32, 32),
|
|
axes=("x",),
|
|
step_mm=0.01,
|
|
max_mm=0.10,
|
|
radius_mm=0.5,
|
|
tracker={"dis_mode": 1, "dis_preset": "medium"},
|
|
noise={"sigmas": [3.0], "draws": 7, "scales": ["process"], "models": ["injected"]},
|
|
)
|
|
job = flow.build_track_job(settings, tmp_path)
|
|
assert job["settings"]["noise"] == {
|
|
"sigmas": [3.0],
|
|
"draws": 7,
|
|
"scales": ["process"],
|
|
"models": ["injected"],
|
|
}
|
|
resolved = flow._resolve_noise_settings(job)
|
|
assert resolved["draws"] == 7
|
|
assert resolved["sigmas"] == [3.0]
|
|
assert resolved["scales"] == ["process"]
|
|
|
|
|
|
def test_floor_basis_is_labelled_render_seed() -> None:
|
|
"""The reclassification must be machine-readable, not just prose."""
|
|
source = Path(flow.__file__).read_text(encoding="utf-8")
|
|
assert '"floor_basis": "render_seed_reproducibility"' in source
|
|
assert "NOT a sensor-noise model" in source
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Retention and disagreement, in the sensitivity report
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
def _scored_scale(gain: float = 10.0):
|
|
from orisys_sensor_design import sensitivity as sens
|
|
|
|
axis = sens.AxisScore(
|
|
axis="x",
|
|
direction_px=(1.0, 0.0),
|
|
direction_source="test",
|
|
floor_px=0.02,
|
|
floor_source="null_projection_p95",
|
|
points=(),
|
|
gain_px_per_mm=gain,
|
|
gain_mm_per_mm=None,
|
|
scale_px_per_mm=13.0,
|
|
gain_gate="no_gain",
|
|
snr=1.0,
|
|
spearman=1.0,
|
|
resolvable_step_mm=None,
|
|
saturation_mm=None,
|
|
cross_axis_leak=0.0,
|
|
score=1.0,
|
|
verdict="resolvable",
|
|
px_per_mm_axis=13.0,
|
|
)
|
|
return sens.ScaleScore(
|
|
name="render",
|
|
width=500,
|
|
height=500,
|
|
px_per_mm=13.0,
|
|
displacement_px_per_step=0.13,
|
|
cell_px=6.25,
|
|
cell_over_patch_px=0.78,
|
|
null_floor_px=0.02,
|
|
null_floor_reported_px=0.02,
|
|
axes=(axis,),
|
|
score=1.0,
|
|
verdict="resolvable",
|
|
)
|
|
|
|
|
|
def test_attach_noise_models_computes_retention_and_snr() -> None:
|
|
scale = _scored_scale(gain=10.0)
|
|
report = {
|
|
"scales": [
|
|
{
|
|
"name": "render",
|
|
"noise_models": {
|
|
"basis": "injected",
|
|
"draws": 1,
|
|
"sigmas": [4.0],
|
|
"axes": {
|
|
"x": [
|
|
{
|
|
"sigma": 4.0,
|
|
"gain_px_per_mm": 8.0,
|
|
"floor_px": 0.04,
|
|
"points": 10,
|
|
"floor_samples": 1,
|
|
}
|
|
]
|
|
},
|
|
},
|
|
}
|
|
]
|
|
}
|
|
attached = sensitivity.attach_noise_models([scale], report, max_mm=0.10)
|
|
models = attached[0].noise_models
|
|
assert models is not None
|
|
row = models["axes"]["x"][0]
|
|
assert row["response_retention"] == pytest.approx(0.8)
|
|
# snr under the same model: |gain| * max_mm / floor
|
|
assert row["snr_noise"] == pytest.approx(8.0 * 0.10 / 0.04)
|
|
|
|
|
|
def test_noise_disagreement_fires_only_past_the_threshold() -> None:
|
|
scale = _scored_scale(gain=10.0)
|
|
noisy = {
|
|
"axes": {"x": [{"sigma": 16.0, "gain_px_per_mm": 6.0, "floor_px": 0.05}]},
|
|
}
|
|
import dataclasses
|
|
|
|
loud = dataclasses.replace(scale, noise_models=noisy)
|
|
message = sensitivity.noise_disagreement(loud)
|
|
assert message is not None and "60%" in message
|
|
|
|
quiet = dataclasses.replace(
|
|
scale,
|
|
noise_models={"axes": {"x": [{"sigma": 16.0, "gain_px_per_mm": 9.5, "floor_px": 0.05}]}},
|
|
)
|
|
assert sensitivity.noise_disagreement(quiet) is None
|
|
assert sensitivity.noise_disagreement(scale) is None
|
|
|
|
|
|
def test_calibration_note_states_the_factor() -> None:
|
|
note = sensitivity.calibration_note(
|
|
{
|
|
"floor_per_period_short": 0.0089,
|
|
"floor_px_short": 0.2267,
|
|
"width": 1280,
|
|
"height": 720,
|
|
"dominant_period_px": 25.5,
|
|
"drift_ratio": 0.98,
|
|
}
|
|
)
|
|
assert "0.00890" in note
|
|
assert "white noise" in note
|
|
|
|
|
|
def test_noise_calibration_is_cached_and_skippable(tmp_path: Path) -> None:
|
|
frames = _noisy_clip(tmp_path / "footage", 3.0, count=6)
|
|
settings = SimpleNamespace(noise={"idle_footage": str(frames), "max_frames": 6})
|
|
first = sensitivity.noise_calibration(tmp_path, settings)
|
|
assert first is not None and first["source"] == "measured"
|
|
assert (tmp_path / sensitivity.IDLE_CALIBRATION_FILENAME).is_file()
|
|
second = sensitivity.noise_calibration(tmp_path, settings)
|
|
assert second is not None and second["source"] == "cached"
|
|
assert sensitivity.noise_calibration(tmp_path, settings, enabled=False) is None
|
|
|
|
unset = sensitivity.noise_calibration(tmp_path, SimpleNamespace(noise={}))
|
|
assert unset is not None and unset["source"] == "unavailable"
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Figure selection: the duplicate-label regression
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
def _twelve_runs() -> list[dict]:
|
|
runs = []
|
|
for nx in (31, 56, 80):
|
|
for radius in (0.5, 1.5):
|
|
for dis_mode in (1, 0):
|
|
runs.append(
|
|
{
|
|
"id": f"nx{nx}@r{radius}@dm{dis_mode}",
|
|
"nx": nx,
|
|
"radius_mm": radius,
|
|
"dis_mode": dis_mode,
|
|
"score": float(nx + radius + dis_mode),
|
|
}
|
|
)
|
|
return runs
|
|
|
|
|
|
def test_select_runs_is_unique_and_represents_every_dis_mode() -> None:
|
|
runs = _twelve_runs()
|
|
chosen = compare.select_runs(runs, limit=4, reference_radius=0.5)
|
|
ids = [run["id"] for run in chosen]
|
|
assert len(ids) == len(set(ids)), "a run must not appear twice"
|
|
assert len(ids) == 4
|
|
# Global best and worst are always present.
|
|
ranked = sorted(runs, key=lambda run: -run["score"])
|
|
assert ranked[0]["id"] in ids and ranked[-1]["id"] in ids
|
|
# And the shipped setting (dm1) is represented at the reference radius: the old
|
|
# best+worst-of-keys[:1] rule picked dm0 only and dropped it.
|
|
assert any(
|
|
run["id"] in ids and run["dis_mode"] == 1 and run["radius_mm"] == 0.5
|
|
for run in runs
|
|
)
|
|
|
|
|
|
def test_select_runs_never_duplicates_when_radii_differ() -> None:
|
|
"""The exact shape that produced two identical `nx=80 dm0` legend entries."""
|
|
runs = _twelve_runs()
|
|
chosen = compare.select_runs(runs, limit=6, reference_radius=0.5)
|
|
labels = [f"nx={r['nx']} r={r['radius_mm']:g} mm dm={r['dis_mode']}" for r in chosen]
|
|
assert len(labels) == len(set(labels))
|