ykcheungab
c377c14474
2. added unwarp map export in current cli, don't need to generate by precompute in sensor running
870 lines
37 KiB
Python
870 lines
37 KiB
Python
"""Synthetic-image tests for the DIS optical-flow stage (no Blender required).
|
|
|
|
Every test here runs against constructed images: the estimator gate, the
|
|
reference-self check, the ROI geometry, the statistics and the job runner. The
|
|
one thing they cannot prove is how DIS behaves on the real Cycles renders -- that
|
|
is what the frozen ``known_answer`` gate in ``flow_report.json`` records for the
|
|
integration run.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
import types
|
|
from pathlib import Path
|
|
|
|
import numpy as np
|
|
import pytest
|
|
|
|
from orisys_sensor_design import flow
|
|
|
|
RENDER = {"name": "render", "width": 500, "height": 500, "resample": False}
|
|
PROCESS = {"name": "process", "width": 250, "height": 250, "resample": True}
|
|
#: The probe.tracker block of the default design.yaml.
|
|
TRACKER = {"algorithm": "dis", "dis_preset": "medium", "dis_mode": 1}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# config / presets
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_dis_preset_from_name_matches_sdk_mapping():
|
|
assert flow.dis_preset_from_name("ultrafast") == 0
|
|
assert flow.dis_preset_from_name("fast") == 1
|
|
assert flow.dis_preset_from_name("medium") == 2
|
|
# Mirrors DIS_no_integrate: unknown and None fall back to MEDIUM.
|
|
assert flow.dis_preset_from_name("FAST") == 1
|
|
assert flow.dis_preset_from_name(None) == 2
|
|
assert flow.dis_preset_from_name("bogus") == 2
|
|
|
|
|
|
def test_load_tracker_config_setdefaults_match_sdk():
|
|
config = flow.load_tracker_config({})
|
|
assert isinstance(config, flow.TrackerConfig)
|
|
assert dict(config) == {
|
|
"algorithm": "dis",
|
|
"dis_preset": "medium",
|
|
"dis_mode": 3,
|
|
"lower_threshold": 12.0,
|
|
"upper_threshold": 18.0,
|
|
"loop_closure_interval": 1,
|
|
}
|
|
# Attribute access, because the SDK reads config_tracker["dis_preset"].
|
|
assert config.dis_preset == "medium"
|
|
assert config.dis_mode == 3
|
|
assert config["algorithm"] == "dis"
|
|
|
|
|
|
def test_load_tracker_config_preserves_explicit_values():
|
|
config = flow.load_tracker_config({"dis_preset": "fast", "dis_mode": 1})
|
|
assert config["dis_preset"] == "fast"
|
|
assert config["dis_mode"] == 1
|
|
assert config["algorithm"] == "dis"
|
|
|
|
|
|
def test_load_tracker_config_unwraps_tracker_and_probe_blocks():
|
|
nested = flow.load_tracker_config({"tracker": {"dis_mode": 1, "dis_preset": "fast"}})
|
|
assert nested["dis_mode"] == 1 and nested["dis_preset"] == "fast"
|
|
design = flow.load_tracker_config(
|
|
{"probe": {"tracker": {"dis_mode": 1, "dis_preset": "fast"}}}
|
|
)
|
|
assert design["dis_mode"] == 1 and design["dis_preset"] == "fast"
|
|
|
|
|
|
def test_load_tracker_config_reads_json_and_yaml_files(tmp_path: Path):
|
|
json_path = tmp_path / "tracker.json"
|
|
json_path.write_text(json.dumps({"tracker": {"dis_mode": 1}}), encoding="utf-8")
|
|
assert flow.load_tracker_config(json_path)["dis_mode"] == 1
|
|
|
|
yaml_path = tmp_path / "design.yaml"
|
|
yaml_path.write_text("probe:\n tracker:\n dis_mode: 1\n dis_preset: fast\n", encoding="utf-8")
|
|
config = flow.load_tracker_config(str(yaml_path))
|
|
assert config["dis_mode"] == 1 and config["dis_preset"] == "fast"
|
|
|
|
|
|
def test_load_tracker_config_rejects_bad_sources(tmp_path: Path):
|
|
with pytest.raises(flow.FlowError, match="not found"):
|
|
flow.load_tracker_config(tmp_path / "nope.json")
|
|
with pytest.raises(flow.FlowError, match="mapping"):
|
|
flow.load_tracker_config(42) # type: ignore[arg-type]
|
|
with pytest.raises(flow.FlowError, match="only implements"):
|
|
flow.load_tracker_config({"algorithm": "raft"})
|
|
with pytest.raises(flow.FlowError, match="dis_mode"):
|
|
flow.load_tracker_config({"dis_mode": -1})
|
|
with pytest.raises(flow.FlowError, match="dis_mode"):
|
|
flow.load_tracker_config({"dis_mode": "1"})
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# image helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_resample_shape_and_area_filter():
|
|
image = np.zeros((4, 4), dtype=np.uint8)
|
|
image[::2, ::2] = 255 # 2x2 blocks average to ~64 with INTER_AREA
|
|
out = flow.resample(image, (2, 2))
|
|
assert out.shape == (2, 2)
|
|
assert out.dtype == np.uint8
|
|
assert np.all((out > 0) & (out < 255)), f"INTER_NEAREST would give 0/255: {out.tolist()}"
|
|
assert np.allclose(out, 64, atol=2)
|
|
|
|
big = flow.resample(image, (8, 8))
|
|
assert big.shape == (8, 8)
|
|
with pytest.raises(flow.FlowError, match="positive"):
|
|
flow.resample(image, (0, 4))
|
|
|
|
|
|
def test_load_gray_roundtrip(tmp_path: Path):
|
|
cv2 = pytest.importorskip("cv2")
|
|
rng = np.random.default_rng(0)
|
|
image = rng.integers(0, 256, (32, 48), dtype=np.uint8)
|
|
path = tmp_path / "frame.png"
|
|
assert cv2.imwrite(str(path), image)
|
|
loaded = flow.load_gray(path)
|
|
assert loaded.shape == (32, 48)
|
|
assert np.array_equal(loaded, image)
|
|
with pytest.raises(flow.FlowError, match="cannot read"):
|
|
flow.load_gray(tmp_path / "missing.png")
|
|
|
|
|
|
def test_missing_cv2_raises_install_hint(monkeypatch: pytest.MonkeyPatch):
|
|
monkeypatch.setattr(flow, "cv2", None)
|
|
with pytest.raises(ImportError, match=r"pip install -e \.\[probe\]"):
|
|
flow.load_gray("whatever.png")
|
|
with pytest.raises(ImportError, match=r"pip install -e \.\[probe\]"):
|
|
flow.FixedReferenceFlow(TRACKER)
|
|
with pytest.raises(ImportError, match=r"pip install -e \.\[probe\]"):
|
|
flow.dis_preset_from_name("medium")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# ROI geometry
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_region_masks_disc_and_control_ring():
|
|
roi, control = flow.region_masks((101, 101), (50.0, 50.0), 10.0, 2.0)
|
|
# Gauss-circle lattice counts, independent of this implementation:
|
|
# |{p in Z^2 : |p| <= 10}| = 317, |{p : 10 < |p| <= 20}| = 1257 - 317 = 940.
|
|
assert int(roi.sum()) == 317
|
|
assert int(control.sum()) == 940
|
|
assert not (roi & control).any()
|
|
assert roi[50, 50]
|
|
assert roi[50, 60] and not control[50, 60] # exactly on the ROI boundary
|
|
assert control[50, 65] and not roi[50, 65] # d = 15, inside the ring
|
|
assert not roi[50, 75] and not control[50, 75] # d = 25, outside both
|
|
assert roi.dtype == bool and control.dtype == bool
|
|
|
|
|
|
def test_region_masks_rejects_bad_geometry():
|
|
with pytest.raises(flow.FlowError, match="radius"):
|
|
flow.region_masks((101, 101), (50.0, 50.0), 0.0)
|
|
with pytest.raises(flow.FlowError, match="control_scale"):
|
|
flow.region_masks((101, 101), (50.0, 50.0), 10.0, 1.0)
|
|
with pytest.raises(flow.FlowError, match="finite"):
|
|
flow.region_masks((101, 101), (np.nan, 50.0), 10.0)
|
|
|
|
|
|
def test_region_masks_truncate_at_the_frame_border():
|
|
roi, _ = flow.region_masks((51, 51), (0.0, 0.0), 10.0)
|
|
assert roi[0, 0]
|
|
assert int(roi.sum()) < flow.expected_disc_pixels(10.0) / 2
|
|
# Quarter disc: by 4-fold symmetry 317 = 4*Q + 4*10 axis points + 1 origin
|
|
# with Q = 69 interior points, so the x >= 0, y >= 0 quarter holds 90.
|
|
assert int(roi.sum()) == 90
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# flow statistics
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_flow_stats_on_constant_flow():
|
|
height = width = 41
|
|
flow_field = np.zeros((height, width, 2), dtype=np.float32)
|
|
flow_field[:, :, 0] = 3.0
|
|
flow_field[:, :, 1] = -4.0
|
|
roi, _ = flow.region_masks((height, width), (20.0, 20.0), 8.0)
|
|
stats = flow.flow_stats(flow_field, roi)
|
|
assert stats["mean_u"] == pytest.approx(3.0)
|
|
assert stats["mean_v"] == pytest.approx(-4.0)
|
|
assert stats["mean_mag"] == pytest.approx(5.0)
|
|
assert stats["p50_mag"] == pytest.approx(5.0)
|
|
assert stats["p95_mag"] == pytest.approx(5.0)
|
|
assert stats["max_mag"] == pytest.approx(5.0)
|
|
assert stats["coverage"] == 1.0
|
|
|
|
|
|
def test_flow_stats_coverage_counts_nonzero_pixels():
|
|
flow_field = np.zeros((21, 21, 2), dtype=np.float32)
|
|
flow_field[0, 0, 0] = 1.0 # outside the ROI
|
|
roi, _ = flow.region_masks((21, 21), (10.0, 10.0), 3.0)
|
|
stats = flow.flow_stats(flow_field, roi)
|
|
assert stats["coverage"] == 0.0
|
|
assert stats["max_mag"] == 0.0
|
|
flow_field[10, 10, 1] = 2.0
|
|
assert flow.flow_stats(flow_field, roi)["coverage"] == pytest.approx(1.0 / int(roi.sum()))
|
|
|
|
|
|
def test_flow_stats_rejects_bad_input():
|
|
roi, _ = flow.region_masks((21, 21), (10.0, 10.0), 3.0)
|
|
with pytest.raises(flow.FlowError, match="mask selects no pixels"):
|
|
flow.flow_stats(np.zeros((21, 21, 2), np.float32), roi & False)
|
|
with pytest.raises(flow.FlowError, match="H, W, 2"):
|
|
flow.flow_stats(np.zeros((21, 21), np.float32), roi)
|
|
bad = np.zeros((21, 21, 2), dtype=np.float32)
|
|
bad[10, 10, 0] = np.nan
|
|
with pytest.raises(flow.FlowError, match="non-finite"):
|
|
flow.flow_stats(bad, roi)
|
|
with pytest.raises(flow.FlowError, match="mask shape"):
|
|
flow.flow_stats(np.zeros((21, 21, 2), np.float32), np.ones((5, 5), bool))
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# forward/backward residual (Tracker.__getForwardBackwardError port)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def texture_200():
|
|
return flow.synthetic_texture(200, 200, seed=11)
|
|
|
|
|
|
def test_forward_backward_error_is_zero_for_identity(texture_200):
|
|
height, width = texture_200.shape
|
|
flow_field = np.zeros((height, width, 2), dtype=np.float32)
|
|
assert flow.forward_backward_error(texture_200, texture_200, flow_field) == 0.0
|
|
|
|
|
|
def test_forward_backward_error_grows_with_a_wrong_flow(texture_200):
|
|
height, width = texture_200.shape
|
|
shifted = flow.known_answer_shift(texture_200, 3, -2)
|
|
truth = np.zeros((height, width, 2), dtype=np.float32)
|
|
truth[:, :, 0] = 3.0
|
|
truth[:, :, 1] = -2.0
|
|
wrong = truth.copy()
|
|
wrong[:, :, 0] += 10.0
|
|
mask = np.zeros((height, width), dtype=bool)
|
|
mask[10:-10, 10:-10] = True
|
|
good = flow.forward_backward_error(texture_200, shifted, truth, mask)
|
|
bad = flow.forward_backward_error(texture_200, shifted, wrong, mask)
|
|
assert good < 5.0, f"a correct flow should leave almost no residual, got {good}"
|
|
assert bad > 5.0 * max(good, 1e-6), f"wrong flow residual {bad} vs {good}"
|
|
with pytest.raises(flow.FlowError, match="does not match"):
|
|
flow.forward_backward_error(texture_200, shifted, truth[:50, :50], mask)
|
|
|
|
|
|
def test_known_answer_shift_is_exact_for_integer_shifts(texture_200):
|
|
shifted = flow.known_answer_shift(texture_200, 2, 0)
|
|
assert shifted.shape == texture_200.shape
|
|
assert np.array_equal(shifted[:, 2:], texture_200[:, :-2])
|
|
assert np.array_equal(shifted[:, :2], texture_200[:, :1].repeat(2, axis=1))
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# the estimator itself
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_reference_self_is_exactly_zero():
|
|
estimator = flow.FixedReferenceFlow(TRACKER)
|
|
image = flow.synthetic_texture(160, 160, seed=2)
|
|
estimator.set_reference(image)
|
|
self_flow = estimator.compute(image)
|
|
assert self_flow.shape == (160, 160, 2)
|
|
assert self_flow.dtype == np.float32
|
|
assert float(np.abs(self_flow).max()) == 0.0
|
|
|
|
|
|
def test_compute_without_a_reference_behaves_like_fixed_ref_dis():
|
|
estimator = flow.FixedReferenceFlow(TRACKER)
|
|
image = flow.synthetic_texture(64, 64, seed=3)
|
|
first = estimator.compute(image) # becomes the reference, returns zeros
|
|
assert float(np.abs(first).max()) == 0.0
|
|
assert np.array_equal(estimator.reference, image)
|
|
with pytest.raises(flow.FlowError, match="does not match the fixed reference"):
|
|
estimator.compute(np.zeros((32, 32), np.uint8))
|
|
|
|
|
|
def test_patch_size_is_measured_from_the_dis_instance():
|
|
estimator = flow.FixedReferenceFlow(TRACKER)
|
|
assert estimator.patch_px == 8
|
|
assert estimator.finest_scale == 1
|
|
assert estimator.preset_name == "medium"
|
|
|
|
|
|
def test_known_answer_recovers_an_integer_shift_at_both_scales():
|
|
block = flow.validate_estimator(TRACKER, scales=[RENDER, PROCESS])
|
|
assert block["passed"] is True
|
|
assert block["tol"] == flow.DEFAULT_KNOWN_ANSWER_TOL_PX
|
|
assert [(case["dx"], case["dy"]) for case in block["cases"]] == [
|
|
(2, 0),
|
|
(0, 2),
|
|
(1, 0),
|
|
(0, 1),
|
|
]
|
|
for case in block["cases"]:
|
|
assert set(case) == {"dx", "dy", "measured_u", "measured_v", "tol"}
|
|
assert abs(case["measured_u"] - case["dx"]) <= case["tol"]
|
|
assert abs(case["measured_v"] - case["dy"]) <= case["tol"]
|
|
|
|
|
|
def test_known_answer_gate_rejects_a_wrong_expectation():
|
|
estimator = flow.FixedReferenceFlow(TRACKER)
|
|
reference = flow.synthetic_texture(200, 200, seed=4)
|
|
# Applied 2 px, but the gate is told to expect 5 px: it must refuse.
|
|
with pytest.raises(flow.FlowError, match="known-answer gate FAILED"):
|
|
flow.measure_known_answer(estimator, reference, 2, 0, expect=(5.0, 0.0))
|
|
# And it must accept the truth.
|
|
case = flow.measure_known_answer(estimator, reference, 2, 0)
|
|
assert abs(case["measured_u"] - 2.0) <= case["tol"]
|
|
|
|
|
|
class _StubEstimator:
|
|
"""Injectable estimator: returns a constant flow vector for every frame."""
|
|
|
|
patch_px = 8
|
|
|
|
def __init__(self, u: float, v: float) -> None:
|
|
self.u, self.v = float(u), float(v)
|
|
self._reference = None
|
|
|
|
def set_reference(self, gray: np.ndarray) -> np.ndarray:
|
|
self._reference = np.asarray(gray).copy()
|
|
return self._reference
|
|
|
|
def compute(self, gray: np.ndarray) -> np.ndarray:
|
|
shape = (*np.asarray(gray).shape[:2], 2)
|
|
return np.stack(
|
|
[np.full(shape[:2], self.u, np.float32), np.full(shape[:2], self.v, np.float32)],
|
|
axis=-1,
|
|
)
|
|
|
|
|
|
def test_known_answer_gate_rejects_a_broken_estimator(monkeypatch: pytest.MonkeyPatch):
|
|
# A stub that always reports zero flow cannot recover a 2 px shift.
|
|
monkeypatch.setattr(flow, "FixedReferenceFlow", lambda config=None: _StubEstimator(0.0, 0.0))
|
|
with pytest.raises(flow.FlowError, match="known-answer gate FAILED") as excinfo:
|
|
flow.validate_estimator(TRACKER, scales=[RENDER])
|
|
assert excinfo.value.report["passed"] is False
|
|
assert excinfo.value.report["cases"][0]["measured_u"] == 0.0
|
|
|
|
|
|
def test_known_answer_gate_accepts_a_correct_stub():
|
|
# The gate is not vacuous: an estimator that answers exactly right passes.
|
|
stub = _StubEstimator(2.0, 0.0)
|
|
case = flow.measure_known_answer(stub, flow.synthetic_texture(200, 200, seed=4), 2, 0)
|
|
assert case["measured_u"] == 2.0 and case["measured_v"] == 0.0
|
|
|
|
|
|
def test_reference_self_helper_rejects_nonzero_flow():
|
|
with pytest.raises(flow.FlowError, match="reference_self is not zero"):
|
|
flow._verify_reference_self(_StubEstimator(0.5, 0.0), np.zeros((16, 16), np.uint8), "render")
|
|
assert flow._verify_reference_self(_StubEstimator(0.0, 0.0), np.zeros((16, 16), np.uint8), "render") == 0.0
|
|
|
|
|
|
def test_evaluate_known_answer_scales_shifts_and_keeps_zero_components():
|
|
block, problems = flow.evaluate_known_answer([RENDER, PROCESS], TRACKER)
|
|
assert problems == []
|
|
assert [(case["dx"], case["dy"]) for case in block["cases"]] == [
|
|
(2, 0),
|
|
(0, 2),
|
|
(1, 0),
|
|
(0, 1),
|
|
]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# strategy / direction reading
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_load_strategy_directions_accepts_contract_shapes(tmp_path: Path):
|
|
path = tmp_path / "strategy_probe.json"
|
|
path.write_text(
|
|
json.dumps({"axes": {"x": {"direction_px": [3.0, 4.0]}, "y": {"direction_px": [0, 2]}}}),
|
|
encoding="utf-8",
|
|
)
|
|
directions = flow.load_strategy_directions(path)
|
|
assert directions["x"] == pytest.approx((0.6, 0.8))
|
|
assert directions["y"] == pytest.approx((0.0, 1.0))
|
|
|
|
path.write_text(
|
|
json.dumps({"axes": [{"axis": "z", "unit_px": [0.0, -5.0]}]}), encoding="utf-8"
|
|
)
|
|
assert flow.load_strategy_directions(path)["z"] == pytest.approx((0.0, -1.0))
|
|
|
|
assert flow.load_strategy_directions(tmp_path / "absent.json") == {}
|
|
|
|
|
|
def test_directions_from_manifest_centroid_finite_difference():
|
|
manifest = {
|
|
"reference_index": 0,
|
|
"frames": [
|
|
{"index": 0, "axis": None, "step_mm": 0.0, "centroid_px": [250.0, 250.0]},
|
|
{"index": 1, "axis": "x", "step_mm": 0.01, "centroid_px": [250.01, 250.0]},
|
|
{"index": 2, "axis": "x", "step_mm": 0.10, "centroid_px": [250.13, 250.0]},
|
|
{"index": 3, "axis": "y", "step_mm": 0.10, "centroid_px": [250.0, 249.87]},
|
|
],
|
|
}
|
|
directions = flow.directions_from_manifest(manifest)
|
|
assert directions["x"] == pytest.approx((1.0, 0.0))
|
|
assert directions["y"] == pytest.approx((0.0, -1.0))
|
|
assert flow.directions_from_manifest({"frames": []}) == {}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# projection helpers (fallbacks; the render manifest is preferred)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_project_point_and_px_per_mm_at():
|
|
intrinsics = {"K": np.array([[500.0, 0, 500.0], [0, 500.0, 500.0], [0, 0, 1.0]]),
|
|
"width": 1000, "height": 1000}
|
|
scale = np.eye(4)
|
|
centre = flow.project_point(intrinsics, scale, (0.0, 0.0, 10.0), size=(500, 500))
|
|
assert centre == pytest.approx((249.5, 249.5), abs=1e-6)
|
|
px_per_mm = flow.px_per_mm_at(
|
|
intrinsics, scale, (0.0, 0.0, 10.0), (1.0, 0.0, 0.0), size=(500, 500)
|
|
)
|
|
assert px_per_mm == pytest.approx(25.0) # 500 px focal / 10 mm depth, halved by the resize
|
|
with pytest.raises(flow.FlowError, match="non-zero vector"):
|
|
flow.px_per_mm_at(intrinsics, scale, (0, 0, 10), (0, 0, 0), size=(500, 500))
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# job building
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_build_track_job_from_mapping_and_object(tmp_path: Path):
|
|
settings = {
|
|
"axes": ("x", "y", "z"),
|
|
"step_mm": 0.01,
|
|
"max_mm": 0.10,
|
|
"radius_mm": 0.5,
|
|
"resolution": (500, 500),
|
|
"process_resolution": (250, 250),
|
|
"tracker": dict(TRACKER),
|
|
}
|
|
job = flow.build_track_job(settings, tmp_path)
|
|
assert job["schema"] == flow.TRACK_JOB_SCHEMA
|
|
assert job["settings"]["axes"] == ["x", "y", "z"]
|
|
assert job["settings"]["tracker"]["dis_mode"] == 1
|
|
assert [scale["name"] for scale in job["scales"]] == ["render", "process"]
|
|
assert job["scales"][1]["resample"] is True
|
|
assert job["artifacts"]["manifest"] == str(tmp_path / flow.RENDER_MANIFEST_FILENAME)
|
|
assert job["artifacts"]["strategy"] == str(tmp_path / flow.STRATEGY_FILENAME)
|
|
|
|
namespace = types.SimpleNamespace(**settings)
|
|
assert flow.build_track_job(namespace, tmp_path) == job
|
|
assert flow.build_track_job(settings, tmp_path, include_process_scale=False)["scales"] == [
|
|
RENDER
|
|
]
|
|
same = dict(settings, process_resolution=(500, 500))
|
|
assert len(flow.build_track_job(same, tmp_path)["scales"]) == 1
|
|
with pytest.raises(flow.FlowError, match="resolution"):
|
|
flow.build_track_job({"resolution": (500, 500, 1)}, tmp_path)
|
|
|
|
|
|
def test_write_track_job_roundtrip(tmp_path: Path):
|
|
job = flow.build_track_job(
|
|
{"resolution": (500, 500), "process_resolution": (250, 250), "tracker": dict(TRACKER)},
|
|
tmp_path,
|
|
)
|
|
path = flow.write_track_job(job, tmp_path)
|
|
assert path.name == flow.TRACK_JOB_FILENAME
|
|
assert json.loads(path.read_text(encoding="utf-8")) == job
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# run_track_job end to end on synthetic frames
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _sha256(path: Path) -> str:
|
|
return hashlib.sha256(path.read_bytes()).hexdigest()
|
|
|
|
|
|
def _write_scene(root: Path, base: np.ndarray | None = None) -> Path:
|
|
"""Write synthetic frames + a contract-shaped render_manifest.json."""
|
|
cv2 = pytest.importorskip("cv2")
|
|
frames_dir = root / "frames"
|
|
frames_dir.mkdir(parents=True, exist_ok=True)
|
|
if base is None:
|
|
base = flow.synthetic_texture(500, 500, seed=7)
|
|
# 13.43 px/mm: 0.01 mm = 0.1343 px, 0.10 mm = 1.343 px.
|
|
shifts = {
|
|
"000_reference": (0.0, 0.0),
|
|
"001_x_+0.01mm": (0.1343, 0.0),
|
|
"002_x_+0.10mm": (1.343, 0.0),
|
|
"003_y_+0.10mm": (0.0, 1.343),
|
|
}
|
|
entries = []
|
|
for index, (name, (dx, dy)) in enumerate(shifts.items()):
|
|
image = base if (dx, dy) == (0.0, 0.0) else flow.known_answer_shift(base, dx, dy)
|
|
path = frames_dir / f"{name}.png"
|
|
assert cv2.imwrite(str(path), image)
|
|
axis = None if index == 0 else ("x" if "x_" in name else "y")
|
|
step = 0.0 if index == 0 else (0.01 if "0.01" in name else 0.10)
|
|
entries.append(
|
|
{
|
|
"index": index,
|
|
"name": name,
|
|
"axis": axis,
|
|
"step_mm": step,
|
|
"png": f"frames/{name}.png",
|
|
"sha256": _sha256(path),
|
|
"width": 500,
|
|
"height": 500,
|
|
"centroid_px": [250.0 + dx, 250.0 + dy],
|
|
}
|
|
)
|
|
rng = np.random.default_rng(21)
|
|
null = np.clip(
|
|
base.astype(np.int16) + rng.integers(-2, 3, base.shape, dtype=np.int16), 0, 255
|
|
).astype(np.uint8)
|
|
null_path = frames_dir / "reference_null.png"
|
|
assert cv2.imwrite(str(null_path), null)
|
|
|
|
manifest = {
|
|
"schema": flow.RENDER_MANIFEST_SCHEMA,
|
|
"width": 500,
|
|
"height": 500,
|
|
"reference_index": 0,
|
|
"probe_px_per_mm": {"p50": 13.43, "min": 10.499, "max": 38.594},
|
|
"probe_roi_px": {"u": 250.0, "v": 250.0, "radius": 20.0},
|
|
"frames": entries,
|
|
"null": {
|
|
"name": "reference_null",
|
|
"seed": 1,
|
|
"png": "frames/reference_null.png",
|
|
"sha256": _sha256(null_path),
|
|
},
|
|
}
|
|
manifest_path = root / flow.RENDER_MANIFEST_FILENAME
|
|
manifest_path.write_text(json.dumps(manifest, indent=2), encoding="utf-8")
|
|
return manifest_path
|
|
|
|
|
|
def _synthetic_job(root: Path) -> dict:
|
|
settings = {
|
|
"axes": ["x", "y"],
|
|
"step_mm": 0.10,
|
|
"max_mm": 0.10,
|
|
"radius_mm": 0.5,
|
|
"resolution": (500, 500),
|
|
"process_resolution": (250, 250),
|
|
"tracker": dict(TRACKER),
|
|
}
|
|
return flow.build_track_job(settings, root)
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def scene(tmp_path_factory: pytest.TempPathFactory) -> tuple[Path, dict]:
|
|
root = tmp_path_factory.mktemp("scene")
|
|
manifest_path = _write_scene(root)
|
|
job = _synthetic_job(root)
|
|
strategy = root / flow.STRATEGY_FILENAME
|
|
strategy.write_text(
|
|
json.dumps({"axes": {"x": {"direction_px": [1.0, 0.0]}, "y": {"direction_px": [0.0, 1.0]}}}),
|
|
encoding="utf-8",
|
|
)
|
|
report, code = flow.run_track_job(job, root, progress=None)
|
|
assert code == flow.EXIT_OK, report
|
|
return root, report
|
|
|
|
|
|
def test_run_track_job_report_matches_the_contract(scene):
|
|
root, report = scene
|
|
assert report["schema"] == flow.FLOW_REPORT_SCHEMA
|
|
assert (root / flow.FLOW_REPORT_FILENAME).is_file()
|
|
tracker = report["tracker"]
|
|
assert tracker["algorithm"] == "dis"
|
|
assert tracker["dis_preset"] == "medium"
|
|
assert tracker["dis_mode"] == 1
|
|
assert tracker["finest_scale"] == 1
|
|
assert tracker["patch_px"] == 8
|
|
assert report["known_answer"]["passed"] is True
|
|
assert report["known_answer"]["cases"]
|
|
assert report["reference_self"] == {"max_abs_flow_px": 0.0}
|
|
diagnostic = report["reference_texture_known_answer"]
|
|
assert diagnostic["available"] is True
|
|
assert diagnostic["gating"] is False
|
|
assert diagnostic["scale"] == "render"
|
|
assert [(case["dx"], case["dy"]) for case in diagnostic["cases"]] == [(2, 0), (0, 2)]
|
|
for case in diagnostic["cases"]:
|
|
assert set(case) >= {"dx", "dy", "measured_u", "measured_v", "tol", "passed"}
|
|
assert [scale["name"] for scale in report["scales"]] == ["render", "process"]
|
|
|
|
|
|
def test_run_track_job_measures_response_and_floor(scene):
|
|
_, report = scene
|
|
by_name = {scale["name"]: scale for scale in report["scales"]}
|
|
|
|
render = by_name["render"]
|
|
assert render["width"] == 500 and render["resample"] is False
|
|
assert render["px_per_mm"] == pytest.approx(13.43)
|
|
assert render["mm_per_px"] == pytest.approx(1.0 / 13.43)
|
|
assert render["px_per_mm_source"] == "render_manifest.probe_px_per_mm.p50"
|
|
assert render["probe_px_per_mm_manifest"] == {"p50": 13.43, "min": 10.499, "max": 38.594}
|
|
assert render["null"]["floor_source"] == "reference_null"
|
|
assert 0.0 < render["null"]["floor_px"] < 0.5
|
|
|
|
frames = {frame["name"]: frame for frame in render["frames"]}
|
|
assert set(frames) == {"001_x_+0.01mm", "002_x_+0.10mm", "003_y_+0.10mm"}
|
|
max_step = frames["002_x_+0.10mm"]
|
|
assert max_step["roi"] == {"u": 250.0, "v": 250.0, "radius": 20.0, "pixels": int(
|
|
flow.region_masks((500, 500), (250.0, 250.0), 20.0)[0].sum()
|
|
)}
|
|
# 1.343 px of applied shift at 13.43 px/mm -> 0.1 mm of response.
|
|
assert max_step["response_proj_px"] == pytest.approx(1.343, abs=0.2)
|
|
assert max_step["response_mm"] == pytest.approx(0.10, abs=0.015)
|
|
assert max_step["forward_backward_error"] < 20.0
|
|
# The 0.01 mm step is 0.134 px: it must still come out positive but small.
|
|
small = frames["001_x_+0.01mm"]
|
|
assert small["response_proj_px"] == pytest.approx(0.1343, abs=0.2)
|
|
assert small["response_proj_px"] < max_step["response_proj_px"]
|
|
# y response is orthogonal to x.
|
|
assert frames["003_y_+0.10mm"]["response_proj_px"] == pytest.approx(1.343, abs=0.2)
|
|
|
|
# The ROI is fixed at the reference centroid for every frame in a scale.
|
|
for frame in render["frames"]:
|
|
assert frame["roi"]["u"] == 250.0 and frame["roi"]["v"] == 250.0
|
|
assert set(frame["flow"]) == {
|
|
"mean_u", "mean_v", "mean_mag", "p50_mag", "p95_mag", "max_mag", "coverage",
|
|
}
|
|
|
|
process = by_name["process"]
|
|
assert process["width"] == 250 and process["resample"] is True
|
|
assert process["px_per_mm"] == pytest.approx(6.715)
|
|
assert process["frames"][0]["roi"]["u"] == pytest.approx(124.75)
|
|
assert process["frames"][0]["roi"]["v"] == pytest.approx(124.75)
|
|
process_max = {frame["name"]: frame for frame in process["frames"]}["002_x_+0.10mm"]
|
|
assert process_max["response_mm"] == pytest.approx(0.10, abs=0.015)
|
|
# The synthetic manifest carries 3 displaced frames while the job settings
|
|
# imply 2; the manifest wins and the mismatch is reported honestly.
|
|
assert any("displaced frames" in warning for warning in report["warnings"])
|
|
|
|
|
|
def test_run_track_job_writes_float32_flow_arrays(scene):
|
|
root, report = scene
|
|
reference = np.load(root / flow.FLOW_DIRNAME / "render" / "000_reference.npy")
|
|
assert reference.shape == (500, 500, 2)
|
|
assert reference.dtype == np.float32
|
|
assert float(np.abs(reference).max()) == 0.0
|
|
null = np.load(root / flow.FLOW_DIRNAME / "render" / "reference_null.npy")
|
|
assert null.shape == (500, 500, 2) and float(np.abs(null).max()) > 0.0
|
|
frame = np.load(root / flow.FLOW_DIRNAME / "render" / "002_x_+0.10mm.npy")
|
|
assert frame.dtype == np.float32 and frame.shape == (500, 500, 2)
|
|
small = np.load(root / flow.FLOW_DIRNAME / "process" / "003_y_+0.10mm.npy")
|
|
assert small.shape == (250, 250, 2)
|
|
assert report["response_direction_source"] == "strategy_probe.json"
|
|
|
|
|
|
def test_run_track_job_uses_manifest_centroids_when_strategy_is_absent(scene):
|
|
root, _ = scene
|
|
(root / flow.STRATEGY_FILENAME).unlink()
|
|
job = _synthetic_job(root)
|
|
report, code = flow.run_track_job(job, root, progress=None)
|
|
assert code == flow.EXIT_OK
|
|
assert report["response_direction_source"] == "render_manifest.frames[].centroid_px"
|
|
assert report["reference_self"]["max_abs_flow_px"] == 0.0
|
|
# Regenerate the strategy file for any later test that shares the scene.
|
|
(root / flow.STRATEGY_FILENAME).write_text(
|
|
json.dumps({"axes": {"x": {"direction_px": [1.0, 0.0]}, "y": {"direction_px": [0.0, 1.0]}}}),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
|
|
def test_run_track_job_missing_manifest_is_exit_3(scene):
|
|
root, _ = scene
|
|
job = _synthetic_job(root)
|
|
job["artifacts"]["manifest"] = str(root / "nope.json")
|
|
report, code = flow.run_track_job(job, root, progress=None)
|
|
assert code == flow.EXIT_MISSING_ARTIFACT and report == {}
|
|
|
|
|
|
def test_run_track_job_missing_frame_is_exit_3(scene, tmp_path: Path):
|
|
root, _ = scene
|
|
manifest = json.loads((root / flow.RENDER_MANIFEST_FILENAME).read_text(encoding="utf-8"))
|
|
manifest["frames"][2]["png"] = "frames/does_not_exist.png"
|
|
broken = tmp_path / "broken"
|
|
broken.mkdir()
|
|
(broken / flow.RENDER_MANIFEST_FILENAME).write_text(json.dumps(manifest), encoding="utf-8")
|
|
job = _synthetic_job(broken)
|
|
report, code = flow.run_track_job(job, broken, progress=None)
|
|
assert code == flow.EXIT_MISSING_ARTIFACT and report == {}
|
|
|
|
|
|
def test_run_track_job_reference_self_failure_is_exit_1(scene, monkeypatch: pytest.MonkeyPatch):
|
|
root, _ = scene
|
|
job = _synthetic_job(root)
|
|
|
|
def _boom(estimator, reference, scale_name):
|
|
raise flow.FlowError(f"reference_self is not zero at scale {scale_name}: max |flow| = 0.5 px")
|
|
|
|
monkeypatch.setattr(flow, "_verify_reference_self", _boom)
|
|
report, code = flow.run_track_job(job, root, progress=None)
|
|
assert code == flow.EXIT_SCENE_ERROR
|
|
assert report == {}
|
|
|
|
|
|
def test_run_track_job_failed_known_answer_is_exit_1(scene, monkeypatch: pytest.MonkeyPatch):
|
|
root, _ = scene
|
|
job = _synthetic_job(root)
|
|
monkeypatch.setattr(flow, "FixedReferenceFlow", lambda config=None: _StubEstimator(0.0, 0.0))
|
|
report, code = flow.run_track_job(job, root, progress=None)
|
|
assert code == flow.EXIT_SCENE_ERROR
|
|
assert report["known_answer"]["passed"] is False
|
|
assert report["known_answer"]["cases"][0]["measured_u"] == 0.0
|
|
|
|
|
|
def test_run_track_job_without_cv2_is_exit_2(scene, monkeypatch: pytest.MonkeyPatch):
|
|
root, _ = scene
|
|
job = _synthetic_job(root)
|
|
monkeypatch.setattr(flow, "cv2", None)
|
|
report, code = flow.run_track_job(job, root, progress=None)
|
|
assert code == flow.EXIT_CONFIG_ERROR and report == {}
|
|
|
|
|
|
def test_run_track_job_rejects_a_bad_schema(scene):
|
|
root, _ = scene
|
|
job = _synthetic_job(root)
|
|
job["schema"] = "something-else"
|
|
report, code = flow.run_track_job(job, root, progress=None)
|
|
assert code == flow.EXIT_CONFIG_ERROR and report == {}
|
|
|
|
|
|
def test_roi_truncation_warns_and_truncates(scene, tmp_path: Path):
|
|
root, _ = scene
|
|
manifest = json.loads((root / flow.RENDER_MANIFEST_FILENAME).read_text(encoding="utf-8"))
|
|
manifest["probe_roi_px"] = {"u": 5.0, "v": 250.0, "radius": 20.0}
|
|
shifted = tmp_path / "truncated"
|
|
shifted.mkdir()
|
|
(shifted / flow.RENDER_MANIFEST_FILENAME).write_text(json.dumps(manifest), encoding="utf-8")
|
|
for name in ("frames",):
|
|
(shifted / name).mkdir(exist_ok=True)
|
|
for entry in manifest["frames"]:
|
|
source = root / entry["png"]
|
|
target = shifted / entry["png"]
|
|
target.write_bytes(source.read_bytes())
|
|
target = shifted / manifest["null"]["png"]
|
|
target.write_bytes((root / manifest["null"]["png"]).read_bytes())
|
|
job = _synthetic_job(shifted)
|
|
report, code = flow.run_track_job(job, shifted, progress=None)
|
|
assert code == flow.EXIT_OK
|
|
render = report["scales"][0]
|
|
assert render["frames"][0]["roi"]["pixels"] < flow.expected_disc_pixels(20.0)
|
|
assert any("truncated" in warning for warning in report["warnings"])
|
|
|
|
|
|
def test_px_per_mm_falls_back_and_is_labelled(tmp_path: Path):
|
|
root = tmp_path / "noscale"
|
|
manifest_path = _write_scene(root)
|
|
manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
|
|
del manifest["probe_px_per_mm"]
|
|
manifest["probe_roi_px"] = {"u": 250.0, "v": 250.0, "radius": 20.0}
|
|
manifest_path.write_text(json.dumps(manifest), encoding="utf-8")
|
|
report, code = flow.run_track_job(_synthetic_job(root), root, progress=None)
|
|
assert code == flow.EXIT_OK
|
|
scale = report["scales"][0]
|
|
# radius 20 px / 0.5 mm = 40 px/mm, labelled as the fallback it is.
|
|
assert scale["px_per_mm"] == pytest.approx(40.0)
|
|
assert scale["px_per_mm_source"] == (
|
|
"render_manifest.probe_roi_px.radius / settings.radius_mm"
|
|
)
|
|
assert scale["probe_px_per_mm_manifest"] is None
|
|
|
|
|
|
def _checker(size: int = 500, cell: float = 3.11) -> np.ndarray:
|
|
"""Periodic marker-like pattern: the real nx=80 cell is ~3.11 px at 500x500."""
|
|
rows, cols = np.mgrid[0:size, 0:size]
|
|
return (((cols // cell) + (rows // cell)) % 2 * 200 + 30).astype(np.uint8)
|
|
|
|
|
|
def test_reference_texture_diagnostic_reports_periodic_aliasing(tmp_path: Path):
|
|
"""The gate must still pass on a periodic scene, but say so honestly.
|
|
|
|
A 3.11 px checker has a 6.22 px period, so a 2 px shift aliases onto the
|
|
wrong period. The synthetic known-answer gate tests the estimator itself
|
|
and must still pass; the reference-texture diagnostic must report the
|
|
aliasing instead of hiding it.
|
|
"""
|
|
root = tmp_path / "periodic"
|
|
_write_scene(root, base=_checker())
|
|
report, code = flow.run_track_job(_synthetic_job(root), root, progress=None)
|
|
assert code == flow.EXIT_OK
|
|
assert report["known_answer"]["passed"] is True
|
|
diagnostic = report["reference_texture_known_answer"]
|
|
assert diagnostic["available"] is True
|
|
assert not all(case["passed"] for case in diagnostic["cases"])
|
|
aliased = [case for case in diagnostic["cases"] if not case["passed"]][0]
|
|
expected = aliased["dx"] if aliased["dx"] else aliased["dy"]
|
|
measured = aliased["measured_u"] if aliased["dx"] else aliased["measured_v"]
|
|
assert abs(measured - expected) > aliased["tol"]
|
|
warning = [w for w in report["warnings"] if "periodic" in w][0]
|
|
assert "applied" in warning and "measured" in warning and "marker period" in warning
|
|
|
|
|
|
def test_px_per_mm_prefers_probe_local_and_labels_mesh_wide_fallback(tmp_path: Path):
|
|
# Probe-local block with per-axis values and no summary key: the transverse
|
|
# (x, y) mean is used and the label says so.
|
|
root = tmp_path / "peraxis"
|
|
manifest_path = _write_scene(root)
|
|
manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
|
|
manifest["probe_px_per_mm"] = {"axes": {"x": 13.0, "y": 17.0, "z": 1.5}}
|
|
manifest["mesh_px_per_mm"] = {"p50": 13.43}
|
|
manifest_path.write_text(json.dumps(manifest), encoding="utf-8")
|
|
report, code = flow.run_track_job(_synthetic_job(root), root, progress=None)
|
|
assert code == flow.EXIT_OK
|
|
scale = report["scales"][0]
|
|
assert scale["px_per_mm"] == pytest.approx(15.0)
|
|
assert scale["px_per_mm_source"] == "render_manifest.probe_px_per_mm.axes(x,y mean)"
|
|
assert scale["probe_px_per_mm_manifest"] == {"axes": {"x": 13.0, "y": 17.0, "z": 1.5}}
|
|
|
|
# Only the mesh-wide statistic exists: use it, but label it as mesh-wide.
|
|
root2 = tmp_path / "meshonly"
|
|
manifest_path2 = _write_scene(root2)
|
|
manifest2 = json.loads(manifest_path2.read_text(encoding="utf-8"))
|
|
del manifest2["probe_px_per_mm"]
|
|
manifest2["mesh_px_per_mm"] = {"p50": 13.43}
|
|
manifest_path2.write_text(json.dumps(manifest2), encoding="utf-8")
|
|
report2, code2 = flow.run_track_job(_synthetic_job(root2), root2, progress=None)
|
|
assert code2 == flow.EXIT_OK
|
|
scale2 = report2["scales"][0]
|
|
assert scale2["px_per_mm"] == pytest.approx(13.43)
|
|
assert "MESH-WIDE" in scale2["px_per_mm_source"]
|
|
|
|
|
|
def test_missing_null_frame_warns_and_reports_no_floor(tmp_path: Path):
|
|
root = tmp_path / "nonull"
|
|
manifest_path = _write_scene(root)
|
|
manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
|
|
del manifest["null"]
|
|
manifest_path.write_text(json.dumps(manifest), encoding="utf-8")
|
|
report, code = flow.run_track_job(_synthetic_job(root), root, progress=None)
|
|
assert code == flow.EXIT_OK
|
|
assert report["scales"][0]["null"]["floor_px"] is None
|
|
assert report["scales"][0]["null"]["floor_source"] == "unavailable"
|
|
# The basis label is what keeps the render-seed floor from being read as a
|
|
# sensor-noise model after the reclassification.
|
|
assert report["scales"][0]["null"]["floor_basis"] == "render_seed_reproducibility"
|
|
assert any("declares no null frame" in warning for warning in report["warnings"])
|
|
assert not (root / flow.FLOW_DIRNAME / "render" / "reference_null.npy").exists()
|
|
|
|
|
|
def test_roi_outside_the_frame_fails_loudly(tmp_path: Path):
|
|
root = tmp_path / "offscreen"
|
|
manifest_path = _write_scene(root)
|
|
manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
|
|
manifest["probe_roi_px"] = {"u": -200.0, "v": -200.0, "radius": 5.0}
|
|
manifest_path.write_text(json.dumps(manifest), encoding="utf-8")
|
|
report, code = flow.run_track_job(_synthetic_job(root), root, progress=None)
|
|
assert code == flow.EXIT_SCENE_ERROR
|
|
assert report == {}
|
|
assert not (root / flow.FLOW_REPORT_FILENAME).exists()
|