Files
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

972 lines
39 KiB
Python

"""Tests for the probe sensitivity scoring and report.
The strongest evidence is the synthetic case with a *known injected gain*: the
scorer must recover exactly the gain it was handed. The magnitudes used here are
the real measured probe-local scales (x 13.248, y 17.670, z 1.519 px/mm at
500x500; halved at the 250x250 process scale) so the gated and ungated paths are
both exercised at true magnitudes.
"""
from __future__ import annotations
import json
from dataclasses import replace
from pathlib import Path
from types import SimpleNamespace
import numpy as np
import pytest
from orisys_sensor_design import sensitivity as sens
H = W = 48
ROI = sens.Roi(u=24.0, v=24.0, radius=5.0)
PROCESS_H = PROCESS_W = 24
PROCESS_ROI = sens.Roi(
u=(ROI.u + 0.5) / 2.0 - 0.5,
v=(ROI.v + 0.5) / 2.0 - 0.5,
radius=ROI.radius / 2.0,
)
DIRECTION = (0.9995, -0.0310)
DIRECTION_UNIT = tuple(
float(value) / float(np.hypot(*DIRECTION)) for value in DIRECTION
)
STEPS = tuple(index * 0.01 for index in range(1, 11))
RENDER_PX_PER_MM = 13.2480
PROCESS_PX_PER_MM = 6.6240
#: Real measured per-axis probe-local scales (px per mm).
RENDER_LOCAL = {"x": 13.2480, "y": 17.6704, "z": 1.5193}
PROCESS_LOCAL = {"x": 6.6240, "y": 8.8352, "z": 0.75965}
#: Injecting gain == probe-local scale models a rigid probe: all applied
#: displacement shows up in the image, so gain_mm_per_mm ~ 1.
AXIS_GAIN = {"x": RENDER_LOCAL["x"], "y": RENDER_LOCAL["y"], "z": RENDER_LOCAL["z"]}
AXIS_SEED = {"x": 11, "y": 22, "z": 33}
# --------------------------------------------------------------------------- #
# Synthetic flow builders
# --------------------------------------------------------------------------- #
def _injected_flow(
gain_px_per_mm: float,
*,
seed: int,
floor_sigma: float = 0.0255,
perpendicular_px_per_mm: float = 0.0,
shape: tuple[int, int] = (H, W),
roi: sens.Roi = ROI,
) -> tuple[list[sens.FrameFlow], np.ndarray]:
"""Frames whose ROI-mean response is ``gain * step`` (plus noise)."""
height, width = shape
rng = np.random.default_rng(seed)
frames = [
sens.FrameFlow(
index=0,
name="000_reference",
axis=None,
step_mm=0.0,
roi=roi,
flow=np.zeros((height, width, 2), dtype=np.float64),
centroid_px=(roi.u, roi.v),
)
]
perpendicular = (-DIRECTION_UNIT[1], DIRECTION_UNIT[0])
for index, step in enumerate(STEPS, start=1):
signal = gain_px_per_mm * step
frame = np.zeros((height, width, 2), dtype=np.float64)
frame[..., 0] = signal * DIRECTION_UNIT[0]
frame[..., 1] = signal * DIRECTION_UNIT[1]
if perpendicular_px_per_mm:
leak = perpendicular_px_per_mm * step
frame[..., 0] += leak * perpendicular[0]
frame[..., 1] += leak * perpendicular[1]
frame += rng.normal(0.0, floor_sigma, (height, width, 2))
frames.append(
sens.FrameFlow(
index=index,
name=f"{index:03d}_frame",
axis="x",
step_mm=step,
roi=roi,
flow=frame,
centroid_px=(
roi.u + signal * DIRECTION_UNIT[0],
roi.v + signal * DIRECTION_UNIT[1],
),
)
)
null = rng.normal(0.0, floor_sigma, (height, width, 2))
return frames, null
def _axis_frames(
axis: str,
gain_px_per_mm: float,
*,
seed: int,
shape: tuple[int, int] = (H, W),
roi: sens.Roi = ROI,
perpendicular_px_per_mm: float = 0.0,
) -> tuple[list[sens.FrameFlow], np.ndarray]:
frames, null = _injected_flow(
gain_px_per_mm,
seed=seed,
shape=shape,
roi=roi,
perpendicular_px_per_mm=perpendicular_px_per_mm,
)
renamed: list[sens.FrameFlow] = []
for frame in frames:
if frame.step_mm == 0.0:
renamed.append(replace(frame, name="000_reference", axis=None))
else:
renamed.append(replace(frame, name=f"{axis}_{frame.step_mm:.2f}", axis=axis))
return renamed, null
def _score(
gain_px_per_mm: float = RENDER_LOCAL["x"],
*,
axis: str = "x",
seed: int = 0,
scale_px_per_mm: float | None = RENDER_LOCAL["x"],
null_flow: np.ndarray | None = None,
manifest_scale_factor: float = 1.0,
perpendicular_px_per_mm: float = 0.0,
**kwargs: object,
) -> sens.AxisScore:
frames, null = _axis_frames(
axis, gain_px_per_mm, seed=seed, perpendicular_px_per_mm=perpendicular_px_per_mm
)
return sens.score_axis(
frames,
axis,
px_per_mm=RENDER_PX_PER_MM,
null_flow=null if null_flow is None else null_flow,
scale_px_per_mm=scale_px_per_mm,
manifest_scale_factor=manifest_scale_factor,
**kwargs,
)
# --------------------------------------------------------------------------- #
# Synthetic known-gain recovery (primary correctness evidence)
# --------------------------------------------------------------------------- #
def test_known_gain_recovered_in_px_per_mm() -> None:
injected = RENDER_LOCAL["x"]
score = _score(injected)
assert score.gain_px_per_mm is not None
recovered = score.gain_px_per_mm
tolerance = 0.20 # 1.5 % of 13.248
assert abs(recovered - injected) < tolerance, (
f"injected gain {injected} px/mm, recovered {recovered:.6f} px/mm; "
f"tolerance {tolerance}"
)
assert 0.030 < score.floor_px < 0.070, score.floor_px
assert score.floor_source == "null_projection_p95"
assert score.direction_source == "centroid_regression"
assert score.direction_px[0] == pytest.approx(DIRECTION_UNIT[0], abs=5e-3)
assert score.direction_px[1] == pytest.approx(DIRECTION_UNIT[1], abs=5e-3)
expected_snr = abs(recovered) * 0.10 / score.floor_px
assert score.snr == pytest.approx(expected_snr, rel=1e-9)
assert score.gain_mm_per_mm == pytest.approx(recovered / RENDER_LOCAL["x"], rel=1e-9)
assert score.gain_gate == "derived"
assert score.verdict == "resolvable"
print(
f"\n[synthetic] injected={injected} px/mm recovered={recovered:.6f} px/mm "
f"tol={tolerance} floor={score.floor_px:.6f} px snr={score.snr:.4f} "
f"spearman={score.spearman:.4f} leak={score.cross_axis_leak:.4f} "
f"score={score.score:.4f}"
)
def test_known_gain_recovery_is_scale_agnostic() -> None:
frames, null = _injected_flow(4.0, seed=3, floor_sigma=0.01)
score = sens.score_axis(
frames, "x", px_per_mm=RENDER_PX_PER_MM, null_flow=null, scale_px_per_mm=4.0
)
assert score.gain_px_per_mm == pytest.approx(4.0, abs=0.05)
def test_gain_requires_three_points_above_floor_else_null() -> None:
score = _score(0.02)
assert score.gain_px_per_mm is None
assert score.gain_mm_per_mm is None
assert score.gain_gate == "no_gain"
assert score.resolvable_step_mm is None
assert score.verdict == "unresolvable"
# --------------------------------------------------------------------------- #
# The mm/mm gate
# --------------------------------------------------------------------------- #
def test_local_scale_gate_boundary_matches_the_measured_axes() -> None:
derived = _score(RENDER_LOCAL["z"], axis="z", scale_px_per_mm=RENDER_LOCAL["z"])
gated = _score(
PROCESS_LOCAL["z"], axis="z", scale_px_per_mm=PROCESS_LOCAL["z"]
)
# Render z = 1.519 px/mm clears the 1.0 threshold and keeps its ratio.
assert derived.gain_gate == "derived"
assert derived.gain_mm_per_mm is not None
# Process z = 0.760 px/mm is gated to null with no score inflation.
assert gated.gain_mm_per_mm is None, "sub-threshold local scale must not be divided"
assert gated.gain_gate == "below_min_local_scale"
assert gated.score < derived.score or gated.gain_px_per_mm is not None
def test_gate_does_not_inflate_the_score() -> None:
injected = RENDER_LOCAL["x"]
derived = _score(injected, scale_px_per_mm=RENDER_LOCAL["x"])
gated = _score(injected, scale_px_per_mm=0.4)
assert gated.gain_mm_per_mm is None
assert gated.gain_gate == "below_min_local_scale"
assert gated.gain_px_per_mm == pytest.approx(injected, abs=0.2)
assert gated.score == pytest.approx(derived.score, rel=1e-9)
assert gated.score < injected / 0.4
unknown = _score(injected, scale_px_per_mm=None)
assert unknown.gain_mm_per_mm is None
assert unknown.gain_gate == "scale_unknown"
def test_min_local_scale_constant_is_one() -> None:
assert sens.min_local_scale_px_per_mm == 1.0
assert sens.MIN_LOCAL_SCALE_PX_PER_MM == 1.0
# --------------------------------------------------------------------------- #
# applied_px must be per-axis, and per-scale
# --------------------------------------------------------------------------- #
def test_applied_px_uses_the_per_axis_scale() -> None:
frames, null = _axis_frames("z", AXIS_GAIN["z"], seed=AXIS_SEED["z"])
score = sens.score_axis(
frames,
"z",
px_per_mm=RENDER_PX_PER_MM,
null_flow=null,
scale_px_per_mm=RENDER_LOCAL["z"],
)
last = score.points[-1]
assert score.px_per_mm_axis is not None
assert last["applied_px"] == pytest.approx(0.10 * score.px_per_mm_axis, rel=1e-6)
# The isotropic scalar 13.248 would claim 1.3248 px; the true z motion is 0.152 px.
assert last["applied_px"] < 0.2
assert last["applied_px"] == pytest.approx(0.10 * RENDER_LOCAL["z"], abs=5e-3)
def test_px_per_mm_axis_and_applied_px_are_rescaled_to_the_scale() -> None:
frames, null = _axis_frames("z", AXIS_GAIN["z"], seed=AXIS_SEED["z"])
render = sens.score_axis(
frames,
"z",
px_per_mm=RENDER_PX_PER_MM,
null_flow=null,
scale_px_per_mm=RENDER_LOCAL["z"],
manifest_scale_factor=1.0,
)
process = sens.score_axis(
frames,
"z",
px_per_mm=PROCESS_PX_PER_MM,
null_flow=null,
scale_px_per_mm=PROCESS_LOCAL["z"],
manifest_scale_factor=0.5,
)
assert process.px_per_mm_axis == pytest.approx(render.px_per_mm_axis * 0.5, rel=1e-9)
assert process.points[-1]["applied_px"] == pytest.approx(
render.points[-1]["applied_px"] * 0.5, rel=1e-9
)
assert process.px_per_mm_axis == pytest.approx(PROCESS_LOCAL["z"], abs=5e-3)
# --------------------------------------------------------------------------- #
# Ambiguous (periodic-pattern aliasing) verdict
# --------------------------------------------------------------------------- #
def _find_axis(report: dict, axis: str, scale_name: str = "render") -> dict:
scale = next(item for item in report["scales"] if item["name"] == scale_name)
for entry in scale["axes"]:
if entry["axis"] == axis:
return entry
raise AssertionError(f"axis {axis!r} not in scale {scale_name!r}")
def _find_scale(report: dict, scale_name: str) -> dict:
return next(item for item in report["scales"] if item["name"] == scale_name)
def test_high_score_with_failed_reference_texture_known_answer_is_ambiguous(
tmp_path: Path,
) -> None:
report = _run_end_to_end(
tmp_path, pattern={"nx": 80, "ny": 80, "seed": 7}, known_answer_passed=False
)
x_axis = _find_axis(report, "x")
assert x_axis["summary"]["score"] > 0, "the score stays a measured number"
assert report["overall"]["score"] > 0
assert x_axis["summary"]["verdict"] == "ambiguous"
assert report["overall"]["verdict"] == "ambiguous"
assert report["overall"]["verdict"] != "resolvable"
aliasing = [w for w in report["warnings"] if "reference_texture_known_answer FAILED" in w]
assert aliasing, report["warnings"]
assert "measured" in aliasing[0] and "expected" in aliasing[0]
assert "8.17" in aliasing[0]
def test_passing_known_answer_does_not_force_ambiguous(tmp_path: Path) -> None:
report = _run_end_to_end(
tmp_path, pattern={"nx": 31, "ny": 31, "seed": 7}, known_answer_passed=True
)
assert _find_axis(report, "x")["summary"]["verdict"] == "resolvable"
assert report["overall"]["verdict"] != "ambiguous"
def test_ambiguous_outranks_unresolvable_and_marginal() -> None:
assert sens.VERDICT_SEVERITY["ambiguous"] < sens.VERDICT_SEVERITY["unresolvable"]
assert sens.VERDICT_SEVERITY["unresolvable"] < sens.VERDICT_SEVERITY["marginal"]
assert sens.VERDICT_SEVERITY["marginal"] < sens.VERDICT_SEVERITY["resolvable"]
# --------------------------------------------------------------------------- #
# Resolvable-step selection, leak, saturation, spearman
# --------------------------------------------------------------------------- #
def test_resolvable_step_is_first_step_above_floor() -> None:
score = _score()
above = [point for point in score.points if point["above_floor"]]
assert above, score.points
assert score.resolvable_step_mm == pytest.approx(above[0]["step_mm"])
assert score.resolvable_step_mm is not None and score.resolvable_step_mm > 0.0
def test_resolvable_step_none_when_nothing_clears_floor() -> None:
frames, _ = _axis_frames("x", 0.0, seed=0)
null = np.full((H, W, 2), 0.2)
score = sens.score_axis(
frames,
"x",
px_per_mm=RENDER_PX_PER_MM,
null_flow=null,
scale_px_per_mm=6.2,
direction_px=(1.0, 0.0),
)
assert score.floor_px == pytest.approx(0.2)
assert score.resolvable_step_mm is None
assert score.gain_px_per_mm is None
assert score.verdict == "unresolvable"
assert score.score == 0.0
def test_resolvable_step_skips_a_below_floor_first_step() -> None:
frames, null = _axis_frames("x", AXIS_GAIN["x"], seed=0)
first = frames[1]
frames[1] = replace(first, flow=np.zeros((H, W, 2)))
score = sens.score_axis(
frames,
"x",
px_per_mm=RENDER_PX_PER_MM,
null_flow=null,
scale_px_per_mm=RENDER_LOCAL["x"],
)
assert score.resolvable_step_mm == pytest.approx(0.02)
def test_leak_penalty_reduces_score() -> None:
clean = _score(seed=1, perpendicular_px_per_mm=0.0)
leaky = _score(seed=1, perpendicular_px_per_mm=RENDER_LOCAL["x"])
assert leaky.cross_axis_leak > clean.cross_axis_leak
assert leaky.gain_px_per_mm == pytest.approx(clean.gain_px_per_mm, abs=0.05)
assert leaky.score < clean.score
expected = leaky.snr * leaky.spearman * (1.0 - min(leaky.cross_axis_leak, 0.5))
assert leaky.score == pytest.approx(expected, rel=1e-9)
def test_leak_penalty_saturates_at_half() -> None:
frames, _ = _axis_frames("x", 0.0, seed=0)
perpendicular = (-DIRECTION_UNIT[1], DIRECTION_UNIT[0])
for index, step in enumerate(STEPS, start=1):
flow = np.zeros((H, W, 2))
flow[..., 0] = perpendicular[0] * step * 10.0
flow[..., 1] = perpendicular[1] * step * 10.0
frames[index] = replace(frames[index], flow=flow)
null = np.full((H, W, 2), 0.001)
score = sens.score_axis(
frames,
"x",
px_per_mm=RENDER_PX_PER_MM,
null_flow=null,
scale_px_per_mm=6.2,
direction_px=DIRECTION_UNIT,
)
assert score.cross_axis_leak == pytest.approx(1.0)
assert score.score == 0.0
def test_saturation_detected_at_first_decrease_after_a_rise() -> None:
frames, _ = _axis_frames("x", 0.0, seed=0)
peaks = (1, 2, 3, 4, 5, 4, 3, 2, 1, 1)
for index, (step, peak) in enumerate(zip(STEPS, peaks), start=1):
flow = np.zeros((H, W, 2))
flow[..., 0] = 0.20 * peak * DIRECTION_UNIT[0]
flow[..., 1] = 0.20 * peak * DIRECTION_UNIT[1]
frames[index] = replace(frames[index], flow=flow)
null = np.full((H, W, 2), 0.01)
score = sens.score_axis(
frames, "x", px_per_mm=RENDER_PX_PER_MM, null_flow=null, scale_px_per_mm=6.2
)
assert score.saturation_mm == pytest.approx(0.06)
def test_no_saturation_when_response_only_rises() -> None:
assert _score().saturation_mm is None
def test_spearman_is_clamped_to_zero_for_decreasing_response() -> None:
frames, _ = _axis_frames("x", 0.0, seed=0)
for index, step in enumerate(STEPS, start=1):
signal = 0.30 - (index - 1) * 0.02
flow = np.zeros((H, W, 2))
flow[..., 0] = signal * DIRECTION_UNIT[0]
flow[..., 1] = signal * DIRECTION_UNIT[1]
frames[index] = replace(frames[index], flow=flow)
null = np.full((H, W, 2), 0.01)
score = sens.score_axis(
frames, "x", px_per_mm=RENDER_PX_PER_MM, null_flow=null, scale_px_per_mm=6.2
)
assert score.spearman == 0.0
assert score.score == 0.0
def test_null_floor_is_the_p95_directional_projection() -> None:
frames, null = _axis_frames("x", AXIS_GAIN["x"], seed=5)
score = sens.score_axis(
frames,
"x",
px_per_mm=RENDER_PX_PER_MM,
null_flow=null,
scale_px_per_mm=RENDER_LOCAL["x"],
)
mask = ROI.mask(H, W)
direction = score.direction_px
projection = np.abs(null[..., 0] * direction[0] + null[..., 1] * direction[1])
assert score.floor_px == pytest.approx(float(np.percentile(projection[mask], 95.0)))
def test_cell_px_uses_the_physical_cell_not_the_camera() -> None:
frames, null = _axis_frames("x", AXIS_GAIN["x"], seed=0)
scale = sens.ScaleInput(
name="render",
width=W,
height=H,
px_per_mm=RENDER_PX_PER_MM,
frames=tuple(frames),
null_flow=null,
step_mm=0.01,
local_scale_px_per_mm={"x": RENDER_LOCAL["x"]},
local_scale_source="per_axis",
mesh_px_per_mm=13.4302,
)
scored = sens.score_scale(scale, cell_mm=[18.5 / 31, 16.4 / 31])
assert scored.cell_px == pytest.approx((18.5 / 31) * RENDER_PX_PER_MM)
assert scored.cell_over_patch_px == pytest.approx((18.5 / 31) * RENDER_PX_PER_MM / 8.0)
assert scored.displacement_px_per_step == pytest.approx(0.01 * RENDER_PX_PER_MM)
# --------------------------------------------------------------------------- #
# End-to-end report: schema, precedence, determinism, exit codes
# --------------------------------------------------------------------------- #
def _write_flow_scale(
flow_dir: Path, name: str, frames: list[sens.FrameFlow], null: np.ndarray
) -> None:
directory = flow_dir / name
directory.mkdir(parents=True, exist_ok=True)
for frame in frames:
if frame.step_mm == 0.0:
continue
np.save(directory / f"{frame.name}.npy", frame.flow.astype(np.float32))
np.save(directory / "reference_null.npy", null.astype(np.float32))
def _run_end_to_end(
tmp_path: Path,
*,
pattern: dict | None,
known_answer_passed: bool = True,
write_texture_job: bool = False,
settings_pattern: dict | None = None,
output_name: str = "out",
) -> dict:
output_dir = tmp_path / output_name
output_dir.mkdir(parents=True, exist_ok=True)
if pattern is not None and "cell_mm" not in pattern:
pattern = dict(pattern)
pattern["cell_mm"] = [18.5 / pattern["nx"], 16.4 / pattern["ny"]]
render_frames: dict[str, list[sens.FrameFlow]] = {}
process_frames: dict[str, list[sens.FrameFlow]] = {}
nulls: dict[str, tuple[np.ndarray, np.ndarray]] = {}
name_map: dict[tuple[str, float], str] = {}
manifest_frames: list[dict] = [
{
"index": 0,
"name": "000_reference",
"axis": None,
"step_mm": 0.0,
"centroid_px": [ROI.u, ROI.v],
}
]
index = 0
for axis in ("x", "y", "z"):
frames, null = _axis_frames(axis, AXIS_GAIN[axis], seed=AXIS_SEED[axis])
half_gain = AXIS_GAIN[axis] * (PROCESS_PX_PER_MM / RENDER_PX_PER_MM)
proc, proc_null = _axis_frames(
axis,
half_gain,
seed=AXIS_SEED[axis],
shape=(PROCESS_H, PROCESS_W),
roi=PROCESS_ROI,
)
render_frames[axis] = frames
process_frames[axis] = proc
nulls[axis] = (null, proc_null)
for frame in frames:
if frame.step_mm == 0.0:
continue
index += 1
name = f"{index:03d}_{axis}_+{frame.step_mm:.2f}mm"
name_map[(axis, frame.step_mm)] = name
manifest_frames.append(
{
"index": index,
"name": name,
"axis": axis,
"step_mm": frame.step_mm,
"centroid_px": list(frame.centroid_px),
}
)
def rename(frames: list[sens.FrameFlow], axis: str) -> list[sens.FrameFlow]:
renamed: list[sens.FrameFlow] = []
for frame in frames:
if frame.step_mm == 0.0:
continue
renamed.append(
replace(frame, name=name_map[(axis, frame.step_mm)], axis=axis)
)
return renamed
scale_specs = (
("render", W, H, RENDER_PX_PER_MM, ROI, render_frames, 0),
("process", PROCESS_W, PROCESS_H, PROCESS_PX_PER_MM, PROCESS_ROI, process_frames, 1),
)
report_frames_by_scale: dict[str, list[dict]] = {}
for name, width, height, px_per_mm, roi, frame_map, null_index in scale_specs:
report_frames: list[dict] = []
combined: list[sens.FrameFlow] = []
# Build the reference first so the flow directory has a consistent shape.
reference = sens.FrameFlow(
index=0,
name="000_reference",
axis=None,
step_mm=0.0,
roi=roi,
flow=np.zeros((height, width, 2)),
centroid_px=(ROI.u, ROI.v),
)
_write_flow_scale(output_dir / "flow", name, [reference], nulls["x"][null_index])
for axis in ("x", "y", "z"):
for frame in rename(frame_map[axis], axis):
combined.append(frame)
report_frames.append(
{
"index": frame.index,
"name": frame.name,
"axis": axis,
"step_mm": frame.step_mm,
"roi": {"u": roi.u, "v": roi.v, "radius": roi.radius},
"flow": {"mean_mag": 0.0},
"response_proj_px": 0.0,
"response_mm": 0.0,
}
)
# Per-axis nulls must share the scale's shape; write each axis's own null
# under a per-axis name and point every frame at the module's canonical
# reference_null by combining the three axes' nulls is not meaningful here.
# Instead: the null is scale-wide and identical for every axis, so use the x
# null (all axes share floor_sigma and seed-independent noise is fine).
_write_flow_scale(
output_dir / "flow",
name,
combined,
nulls["x"][null_index],
)
report_frames_by_scale[name] = report_frames
known_answer = {
"passed": known_answer_passed,
"gating": False,
"cases": [
{
"dx": 2,
"dy": 0,
"expected_u": 2.0,
"expected_v": 0.0,
"measured_u": 2.0 if known_answer_passed else 8.17,
"measured_v": 0.0,
}
],
}
flow_report = {
"schema": sens.FLOW_REPORT_SCHEMA,
"tracker": {"algorithm": "dis", "dis_preset": "medium", "dis_mode": 1},
"known_answer": {"passed": True},
"reference_texture_known_answer": known_answer,
"scales": [
{
"name": name,
"width": width,
"height": height,
"px_per_mm": px_per_mm,
"mm_per_px": 1.0 / px_per_mm,
"frames": report_frames_by_scale[name],
"null": {"floor_px": 0.05, "floor_source": "reference_null"},
}
for name, width, height, px_per_mm, _roi, _fm, _ni in scale_specs
],
}
(output_dir / "flow_report.json").write_text(json.dumps(flow_report), encoding="utf-8")
manifest = {
"schema": sens.RENDER_MANIFEST_SCHEMA,
"texture": {"path": str(output_dir / "marker.png"), "sha256": "abc123"},
"width": W,
"height": H,
"mesh_px_per_mm": {"min": 10.4994, "p05": 10.9457, "p50": 13.4302,
"p95": 22.8857, "max": 38.5943},
"probe_px_per_mm": {"p50": 13.2480, "x": 13.2480, "y": 17.6704, "z": 1.5193},
"probe_roi_px": {"u": ROI.u, "v": ROI.v, "radius": ROI.radius},
"reference_index": 0,
"reference_deterministic": True,
"frames": manifest_frames,
"null": {"name": "reference_null", "seed": 1},
}
(output_dir / "render_manifest.json").write_text(json.dumps(manifest), encoding="utf-8")
if write_texture_job:
(output_dir / "_texture_job.json").write_text(
json.dumps(
{
"schema": "orisys-sensor-design/texture-job-v3",
"settings": {
"nx": 20,
"ny": 20,
"seed": 7,
"vector": {"width_mm": 18.5, "height_mm": 16.4},
},
}
),
encoding="utf-8",
)
settings = SimpleNamespace(
step_mm=0.01,
max_mm=0.10,
axes=["x", "y", "z"],
resolution=[500, 500],
process_resolution=[250, 250],
radius_mm=0.5,
falloff="smooth",
group="probe",
mesh="sensing_surface",
tracker={"dis_preset": "medium", "dis_mode": 1, "patch_px": 8},
pattern=settings_pattern,
)
report, code = sens.run_sensitivity(
settings, output_dir, pattern=pattern, progress=lambda message: None
)
assert code == 0, report.get("warnings")
assert (output_dir / sens.SENSITIVITY_REPORT_FILENAME).is_file()
return report
def test_report_schema_keys(tmp_path: Path) -> None:
report = _run_end_to_end(tmp_path, pattern={"nx": 31, "ny": 31, "seed": 7})
assert report["schema"] == sens.SENSITIVITY_REPORT_SCHEMA
for key in ("generated_at", "revision", "texture", "probe", "tracker", "scales",
"overall", "warnings", "notes"):
assert key in report, key
assert set(report["overall"]) >= {"score", "verdict", "best_scale", "best_axis"}
assert report["texture"]["grid"] == [31, 31]
assert report["texture"]["cell_mm"] == pytest.approx([18.5 / 31, 16.4 / 31])
assert [scale["name"] for scale in report["scales"]] == ["render", "process"]
for scale in report["scales"]:
for key in ("name", "px_per_mm", "displacement_px_per_step", "cell_px",
"cell_over_patch_px", "null_floor_px", "axes", "score", "verdict"):
assert key in scale, key
axis = _find_axis(report, "x")
assert set(axis) >= {"axis", "direction_px", "direction_source", "points", "summary"}
assert axis["direction_source"] in {
"centroid_regression",
"centroid_finite_difference",
"roi_mean_flow_at_max_step",
"provided",
}
assert set(axis["points"][0]) == {"step_mm", "applied_px", "response_px", "above_floor"}
for key in ("gain_px_per_mm", "gain_mm_per_mm", "scale_px_per_mm", "snr",
"spearman", "resolvable_step_mm", "saturation_mm",
"cross_axis_leak", "floor_px"):
assert key in axis["summary"], key
def test_report_is_deterministic_given_same_inputs(tmp_path: Path) -> None:
first = _run_end_to_end(tmp_path, pattern={"nx": 31, "ny": 31, "seed": 7})
second = _run_end_to_end(tmp_path, pattern={"nx": 31, "ny": 31, "seed": 7})
first.pop("generated_at")
second.pop("generated_at")
assert json.dumps(first, sort_keys=True) == json.dumps(second, sort_keys=True)
def test_pattern_precedence_explicit_beats_texture_job(tmp_path: Path) -> None:
report = _run_end_to_end(
tmp_path, pattern={"nx": 31, "ny": 31, "seed": 7}, write_texture_job=True
)
assert report["texture"]["grid"] == [31, 31]
assert "pattern argument" in " ".join(report["notes"])
def test_pattern_falls_back_to_texture_job(tmp_path: Path) -> None:
report = _run_end_to_end(tmp_path, pattern=None, write_texture_job=True)
assert report["texture"]["grid"] == [20, 20]
assert report["texture"]["cell_mm"] == pytest.approx([18.5 / 20, 16.4 / 20])
assert "_texture_job.json" in " ".join(report["notes"])
def test_pattern_unknown_is_null_not_a_guessed_80x80(tmp_path: Path) -> None:
report = _run_end_to_end(tmp_path, pattern=None, write_texture_job=False)
assert report["texture"]["grid"] is None
assert report["texture"]["cell_mm"] is None
assert report["scales"][0]["cell_px"] is None
assert any("unknown" in note for note in report["notes"])
def test_settings_pattern_is_used_when_no_argument(tmp_path: Path) -> None:
report = _run_end_to_end(
tmp_path,
pattern=None,
settings_pattern={"nx": 40, "ny": 40, "seed": 7, "cell_mm": [0.4625, 0.41]},
)
assert report["texture"]["grid"] == [40, 40]
assert report["texture"]["cell_mm"] == pytest.approx([0.4625, 0.41])
assert "settings.pattern" in " ".join(report["notes"])
def test_missing_artifact_returns_3_and_still_writes_a_report(tmp_path: Path) -> None:
output_dir = tmp_path / "empty"
output_dir.mkdir()
settings = SimpleNamespace(step_mm=0.01, radius_mm=0.5, falloff="smooth")
report, code = sens.run_sensitivity(settings, output_dir, progress=lambda m: None)
assert code == 3
path = output_dir / sens.SENSITIVITY_REPORT_FILENAME
assert path.is_file()
written = json.loads(path.read_text(encoding="utf-8"))
assert written["schema"] == sens.SENSITIVITY_REPORT_SCHEMA
assert written["warnings"]
assert written["overall"]["verdict"] == "unresolvable"
def test_degraded_report_on_bad_flow_report(tmp_path: Path) -> None:
output_dir = tmp_path / "bad"
output_dir.mkdir()
(output_dir / "flow_report.json").write_text("{not json", encoding="utf-8")
(output_dir / "render_manifest.json").write_text("{}", encoding="utf-8")
report, code = sens.run_sensitivity(SimpleNamespace(step_mm=0.01), output_dir, progress=lambda m: None)
assert code == 1
assert (output_dir / sens.SENSITIVITY_REPORT_FILENAME).is_file()
assert report["error"]["kind"] == "data"
def test_subpixel_and_cell_warnings_are_emitted(tmp_path: Path) -> None:
report = _run_end_to_end(tmp_path, pattern={"nx": 80, "ny": 80, "seed": 7})
joined = " ".join(report["warnings"])
assert "below one pixel" in joined
# The cell warning is now about representability, not a "DIS patch Nyquist limit"
# statement derived from the old millimetre-based cell.
assert "cannot be represented at all" in joined
# nx=80 at 500 px is a 6.25 px cell against a 16 px effective window (ratio 0.39),
# so the effective-window guidance must fire and name the ratio.
assert "effective DIS window" in joined
def test_cell_px_is_the_camera_space_cell_and_effective_window_is_reported(
tmp_path: Path,
) -> None:
"""The governing numbers must be camera-space, and consistent with each other."""
report = _run_end_to_end(tmp_path, pattern={"nx": 80, "ny": 80, "seed": 7})
scale = next(item for item in report["scales"] if item["name"] == "render")
assert scale["reference_width_px"] == scale["width"]
assert scale["cell_px"] == pytest.approx(scale["width"] / 80.0, rel=1e-9)
assert scale["cell_px_source"].startswith("render_width/nx")
# patch 8 px x 2**finest_scale; the fixture tracker config uses dis_mode=1.
assert scale["effective_patch_px"] == pytest.approx(16.0, rel=1e-9)
assert scale["cell_over_effective_patch_px"] == pytest.approx(
scale["cell_px"] / scale["effective_patch_px"], rel=1e-9
)
assert scale["cell_over_patch_px"] == pytest.approx(scale["cell_px"] / 8.0, rel=1e-9)
def test_nx31_is_near_the_effective_window_on_a_500px_render() -> None:
"""The regression that pins the corrected interpretation.
nx=31 on a 500 px render is a 16.13 px cell, which is 1.008x the 16 px effective
window of an 8 px patch at finest_scale 1 -- the measured render-scale optimum.
"""
win = sens.cell_metrics(width=500, nx=31, cell_mm=(0.5968, 0.529), px_per_mm=13.248,
finest_scale=1)
assert win["cell_px"] == pytest.approx(16.129, rel=1e-3)
assert win["cell_px_source"].startswith("render_width/nx")
assert win["effective_patch_px"] == pytest.approx(16.0, rel=1e-9)
assert win["cell_over_effective_patch_px"] == pytest.approx(1.008, rel=1e-2)
# The old, wrong definition is still recorded for comparison, and differs ~2x.
assert win["nominal_cell_px"] == pytest.approx(7.906, rel=1e-3)
finer = sens.cell_metrics(width=500, nx=80, cell_mm=(0.23125, 0.23125), px_per_mm=13.248,
finest_scale=1)
assert finer["cell_px"] == pytest.approx(6.25, rel=1e-9)
assert finer["cell_over_effective_patch_px"] == pytest.approx(0.3906, rel=1e-3)
# dis_mode=0 halves the effective window, so the same pattern sits at 2x the ratio.
coarse = sens.cell_metrics(width=500, nx=31, cell_mm=(0.5968, 0.529), px_per_mm=13.248,
finest_scale=0)
assert coarse["effective_patch_px"] == pytest.approx(8.0, rel=1e-9)
assert coarse["cell_over_effective_patch_px"] == pytest.approx(2.016, rel=1e-2)
# Without a grid the metric falls back and says so, rather than guessing.
fallback = sens.cell_metrics(width=500, nx=None, cell_mm=(0.5968, 0.529), px_per_mm=13.248)
assert fallback["cell_px_source"].startswith("cell_mm x probe-local")
assert fallback["cell_px"] == pytest.approx(7.906, rel=1e-3)
def test_step_size_and_estimator_floor_are_reported_without_conflation(
tmp_path: Path,
) -> None:
report = _run_end_to_end(tmp_path, pattern={"nx": 80, "ny": 80, "seed": 7})
axis = _find_axis(report, "x", "render")
# The step is sub-pixel but still clears the measured floor: 0.1325 px > ~0.046 px.
assert axis["summary"]["resolvable_step_mm"] == pytest.approx(0.01)
assert axis["summary"]["floor_px"] < 0.01 * RENDER_PX_PER_MM
joined = " ".join(report["warnings"])
assert "below one pixel" in joined
# "Below one pixel" must never be asserted as "below the estimator floor".
assert "below the estimator floor" not in joined
assert "estimator floor" in joined, "the floor must still be stated"
assert "per-pixel precision limit" in joined
# The old false claim must be gone: the ROI mean does resolve single steps here.
notes = " ".join(report["notes"])
assert "Only the ROI-aggregated response over the whole sweep is resolvable" not in notes
assert "smallest step clearing the measured floor" in notes
def test_roi_confound_and_attribution_notes_are_present(tmp_path: Path) -> None:
report = _run_end_to_end(tmp_path, pattern={"nx": 31, "ny": 31, "seed": 7})
joined = " ".join(report["notes"])
assert "confounded with the ROI size" in joined
assert "never the radius" in joined
assert "17%" in joined and "95%" in joined
assert "not noise, aliasing" in joined
assert "2.0000 px" in joined
def test_null_floor_is_the_projection_p95_on_that_scale_roi(tmp_path: Path) -> None:
"""Regression guard for the 15x floor error.
A projected p95 (``|dot(flow, d_hat)|`` over the ROI) can never exceed the p95
of ``|flow|`` over the *same* field and region, so that invariant bounds the
reported ``null_floor_px``. The exact-equality check pins the field *and* the
region: using the render-scale manifest ROI on the process-scale arrays (the
bug that inflated the process floor to 0.262662 px against a true 0.017082 px)
fails here on both scales.
"""
report = _run_end_to_end(tmp_path, pattern={"nx": 80, "ny": 80, "seed": 7})
output_dir = tmp_path / "out"
flow_report = json.loads((output_dir / "flow_report.json").read_text(encoding="utf-8"))
for scale in report["scales"]:
name = scale["name"]
entry = next(item for item in flow_report["scales"] if item["name"] == name)
raw = entry["frames"][0]["roi"]
roi = sens.Roi(u=float(raw["u"]), v=float(raw["v"]), radius=float(raw["radius"]))
null = np.load(output_dir / "flow" / name / "reference_null.npy").astype(np.float64)
mask = roi.mask(null.shape[0], null.shape[1])
assert mask.any(), name
if name != "render":
# The render-scale manifest ROI is a *different* region on this scale's
# grid; confirm the guard can tell them apart.
manifest_roi = sens.Roi(u=ROI.u, v=ROI.v, radius=ROI.radius)
assert not np.array_equal(
mask, manifest_roi.mask(null.shape[0], null.shape[1])
), name
magnitude = np.sqrt(null[..., 0] ** 2 + null[..., 1] ** 2)
magnitude_p95 = float(np.percentile(magnitude[mask], 95.0))
assert scale["null_floor_px"] <= magnitude_p95 + 1e-9, name
for axis in scale["axes"]:
direction = axis["direction_px"]
projection = np.abs(
null[..., 0] * direction[0] + null[..., 1] * direction[1]
)
expected = float(np.percentile(projection[mask], 95.0))
assert axis["summary"]["floor_px"] == pytest.approx(
expected, rel=1e-9, abs=1e-6
), (name, axis["axis"])
assert axis["summary"]["floor_px"] <= magnitude_p95 + 1e-9, (name, axis["axis"])
def test_address_scale_gate_and_rescale_end_to_end(tmp_path: Path) -> None:
report = _run_end_to_end(tmp_path, pattern={"nx": 31, "ny": 31, "seed": 7})
render_scale = _find_scale(report, "render")
process_scale = _find_scale(report, "process")
assert render_scale["mesh_px_per_mm"] == pytest.approx(13.4302)
assert render_scale["px_per_mm"] == pytest.approx(RENDER_PX_PER_MM)
assert process_scale["px_per_mm"] == pytest.approx(PROCESS_PX_PER_MM)
for axis in ("x", "y", "z"):
render_axis = _find_axis(report, axis, "render")
process_axis = _find_axis(report, axis, "process")
# applied_px is per-axis and per-scale, not the isotropic scalar.
assert render_axis["points"][-1]["applied_px"] == pytest.approx(
0.10 * render_axis["summary"]["px_per_mm_axis"], rel=1e-3
)
assert process_axis["points"][-1]["applied_px"] == pytest.approx(
0.10 * process_axis["summary"]["px_per_mm_axis"], rel=1e-3
)
# px_per_mm_axis is at the scale it appears in.
assert render_axis["summary"]["px_per_mm_axis"] == pytest.approx(
RENDER_LOCAL[axis], rel=0.05
)
assert process_axis["summary"]["px_per_mm_axis"] == pytest.approx(
PROCESS_LOCAL[axis], rel=0.05
)
# z is 0.152 px over the whole sweep, not 1.325 px.
assert _find_axis(report, "z", "render")["points"][-1]["applied_px"] == pytest.approx(
0.152, abs=0.01
)
# Render z (1.519) keeps its derived ratio; process z (0.760) is gated to null.
assert _find_axis(report, "z", "render")["summary"]["gain_gate"] == "derived"
assert _find_axis(report, "z", "render")["summary"]["gain_mm_per_mm"] is not None
assert _find_axis(report, "z", "process")["summary"]["gain_gate"] == "below_min_local_scale"
assert _find_axis(report, "z", "process")["summary"]["gain_mm_per_mm"] is None
assert any(
"min_local_scale_px_per_mm" in warning for warning in report["warnings"]
)