"""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()