Files
2026-09-27 19:21:03 +08:00

626 lines
22 KiB
Python

"""Tests for headless probe-deformation settings, frame plan, and job I/O."""
from __future__ import annotations
import argparse
import dataclasses
import json
from collections.abc import Mapping
from pathlib import Path
import pytest
from orisys_sensor_design.deform import (
DEFAULT_PROBE,
DEFORM_JOB_FILENAME,
DEFORM_JOB_SCHEMA,
DEFORM_REPORT_FILENAME,
FRAMES_NPZ_FILENAME,
PatternInfo,
ProbeConfigError,
ProbeFrame,
build_deform_job,
deform_job_path,
frame_plan,
load_probe_block,
merge_probe,
probe_artifact_paths,
probe_output_dir,
probe_overrides_from_args,
probe_revision_rel,
resolve_pattern,
resolve_probe,
resolve_probe_output_dir,
validate_probe,
write_deform_job,
)
from orisys_sensor_design.manifest import RevisionPaths, load_yaml, repo_root
XYNOVA_RENDER = {
"engine": "CYCLES",
"samples": 16,
"denoise": False,
"view_transform": "Standard",
"interpolate": "Closest",
"seed": 0,
}
XYNOVA_TRACKER = {"algorithm": "dis", "dis_preset": "medium", "dis_mode": 1}
@pytest.fixture()
def revision_dir() -> Path:
return repo_root() / "sensors" / "finger-xynova" / "r01"
@pytest.fixture()
def paths(revision_dir: Path) -> RevisionPaths:
return RevisionPaths.from_revision_dir(revision_dir)
def _write_revision(
tmp_path: Path,
probe_yaml: str = "",
*,
sensor_field: str = "finger",
sensor_dir: str = "finger",
revision_name: str = "r99",
) -> Path:
"""Create a minimal sensors/<sensor_dir>/<revision_name> revision under tmp_path."""
revision = tmp_path / "sensors" / sensor_dir / revision_name
(revision / "source").mkdir(parents=True)
(revision / "source" / "assembly.blend").write_bytes(b"")
design = (
f"sensor: {sensor_field}\n"
f"revision: {revision_name}\n"
"meters_per_source_unit: 0.001\n"
"source_blend: source/assembly.blend\n"
)
design += probe_yaml
(revision / "design.yaml").write_text(design, encoding="utf-8")
return revision
def _probe_namespace(**overrides: object) -> argparse.Namespace:
namespace = argparse.Namespace(
mesh=None,
group=None,
radius_mm=None,
falloff=None,
step_mm=None,
max_mm=None,
axes=None,
)
for key, value in overrides.items():
setattr(namespace, key, value)
return namespace
# ------------------------------------------------------------- settings
def _design_probe(paths: RevisionPaths) -> dict:
"""The ``probe`` block exactly as written in design.yaml.
Read rather than hard-coded. Resolution and ``dis_mode`` are deliberate choices that
change as the bench is re-targeted (250x250 needs ``dis_mode=0``; 500x500 does not),
and duplicating them here would fail on a legitimate edit while asserting nothing
extra about the pipeline. What the tests below must prove is that the resolved
settings *equal* what the design declares, at whatever value it declares.
"""
return load_yaml(paths.design_yaml)["probe"]
def _as_pair(value) -> tuple[int, int]:
return (int(value[0]), int(value[1]))
def test_defaults_come_from_design_yaml(paths: RevisionPaths) -> None:
settings = resolve_probe(paths)
design = _design_probe(paths)
assert settings.mesh == "sensing_surface"
assert settings.camera == "cam_sensor"
assert settings.group == "probe"
assert settings.uv_map == "compute"
assert settings.texture == "patch_texture_compute.png"
assert settings.radius_mm == pytest.approx(0.5)
assert settings.falloff == "smooth"
assert settings.step_mm == pytest.approx(0.01)
assert settings.max_mm == pytest.approx(0.10)
assert settings.axes == ("x", "y", "z")
assert settings.resolution == _as_pair(design["resolution"])
assert settings.process_resolution == _as_pair(design["process_resolution"])
assert settings.tracker == dict(design["tracker"])
assert settings.render == XYNOVA_RENDER
# The marker grid is resolved from the shared texture_patch block.
assert settings.pattern == PatternInfo(
nx=80, ny=80, seed=7, vector_width_mm=18.5, vector_height_mm=16.4
)
assert settings.pattern.cell_mm == pytest.approx((0.23125, 0.205))
def test_pattern_grid_resolves_from_texture_patch(paths: RevisionPaths) -> None:
pattern = resolve_pattern(paths)
assert pattern.to_job_entry() == {
"nx": 80,
"ny": 80,
"seed": 7,
"vector_width_mm": 18.5,
"vector_height_mm": 16.4,
"cell_mm": [0.23125, 0.205],
}
# The sensitivity stage duck-types settings.pattern as a mapping (.get).
assert isinstance(pattern, Mapping)
assert pattern.get("nx") == 80
assert pattern.get("seed") == 7
assert pattern.get("cell_mm") == [0.23125, 0.205]
assert dict(pattern) == pattern.to_job_entry()
def test_pattern_falls_back_to_texture_defaults(tmp_path: Path) -> None:
revision = _write_revision(tmp_path)
pattern = resolve_pattern(RevisionPaths.from_revision_dir(revision))
assert pattern == PatternInfo(
nx=80, ny=45, seed=None, vector_width_mm=164.0, vector_height_mm=160.0
)
@pytest.mark.parametrize(
("texture_yaml", "expected"),
[
("texture_patch:\n nx: 0\n", "texture_patch.nx"),
("texture_patch:\n ny: many\n", "texture_patch.ny"),
("texture_patch:\n seed: soon\n", "texture_patch.seed"),
("texture_patch:\n vector:\n width_mm: 0\n", "vector.width_mm"),
("texture_patch:\n vector:\n height_mm: -1\n", "vector.height_mm"),
("texture_patch:\n widht: 100\n", "widht"),
],
)
def test_pattern_rejects_bad_texture_patch(
tmp_path: Path, texture_yaml: str, expected: str
) -> None:
revision = _write_revision(tmp_path, texture_yaml)
with pytest.raises(ProbeConfigError) as excinfo:
resolve_pattern(RevisionPaths.from_revision_dir(revision))
assert expected in str(excinfo.value)
def test_builtin_defaults_when_probe_block_absent(tmp_path: Path) -> None:
revision = _write_revision(tmp_path)
paths = RevisionPaths.from_revision_dir(revision)
assert load_probe_block(paths) == {}
settings = resolve_probe(paths)
assert settings.radius_mm == DEFAULT_PROBE["radius_mm"]
assert settings.falloff == DEFAULT_PROBE["falloff"]
assert settings.step_mm == DEFAULT_PROBE["step_mm"]
assert settings.max_mm == DEFAULT_PROBE["max_mm"]
assert settings.axes == tuple(DEFAULT_PROBE["axes"])
# Object names fall back to design.yaml `objects` defaults.
assert settings.mesh == "sensing_surface"
assert settings.camera == "cam_sensor"
assert settings.group == "probe"
def test_design_yaml_overrides_builtin_defaults(tmp_path: Path) -> None:
revision = _write_revision(
tmp_path,
"probe:\n"
" radius_mm: 0.25\n"
" step_mm: 0.02\n"
" max_mm: 0.06\n"
" axes: [z, x]\n"
" tracker:\n"
" dis_preset: fast\n",
)
settings = resolve_probe(RevisionPaths.from_revision_dir(revision))
assert settings.radius_mm == pytest.approx(0.25)
assert settings.step_mm == pytest.approx(0.02)
assert settings.max_mm == pytest.approx(0.06)
assert settings.axes == ("z", "x")
assert settings.tracker["dis_preset"] == "fast"
# Untouched nested keys keep their defaults.
assert settings.tracker["dis_mode"] == 1
assert settings.render == XYNOVA_RENDER
def test_cli_overrides_beat_design_yaml(paths: RevisionPaths) -> None:
settings = resolve_probe(
paths,
{
"mesh": "other_mesh",
"group": "other_group",
"radius_mm": 0.25,
"falloff": "Sharp",
"step_mm": 0.02,
"max_mm": 0.08,
"axes": ["y", "z"],
},
)
assert settings.mesh == "other_mesh"
assert settings.group == "other_group"
assert settings.radius_mm == pytest.approx(0.25)
# Falloff is normalised to the contract's lowercase mode names.
assert settings.falloff == "sharp"
assert settings.step_mm == pytest.approx(0.02)
assert settings.max_mm == pytest.approx(0.08)
assert settings.axes == ("y", "z")
def test_unset_overrides_do_not_erase_configured_values(paths: RevisionPaths) -> None:
merged = merge_probe(
paths, {"radius_mm": None, "falloff": None, "axes": None, "tracker": None}
)
assert merged["radius_mm"] == 0.5
assert merged["falloff"] == "smooth"
assert merged["axes"] == ["x", "y", "z"]
# Compared against the design rather than a pinned literal: `dis_mode` is a
# deliberate per-resolution choice (250x250 needs 0, 500x500 uses 1), and the point
# of this test is that a None override does not erase what the design declares.
assert merged["tracker"] == dict(_design_probe(paths)["tracker"])
def test_probe_must_be_a_mapping(tmp_path: Path) -> None:
revision = _write_revision(tmp_path, "probe: fast\n")
with pytest.raises(ProbeConfigError) as excinfo:
resolve_probe(RevisionPaths.from_revision_dir(revision))
assert "must be a mapping" in str(excinfo.value)
@pytest.mark.parametrize(
("probe_yaml", "expected"),
[
("probe:\n radus_mm: 0.4\n", "radus_mm"),
("probe:\n tracker:\n preset: medium\n", "preset"),
("probe:\n render:\n view_transfrom: Standard\n", "view_transfrom"),
("probe:\n tracker: fast\n", "tracker"),
("probe:\n render: cycles\n", "render"),
],
)
def test_unknown_or_mistyped_keys_are_rejected(
tmp_path: Path, probe_yaml: str, expected: str
) -> None:
revision = _write_revision(tmp_path, probe_yaml)
with pytest.raises(ProbeConfigError) as excinfo:
resolve_probe(RevisionPaths.from_revision_dir(revision))
assert expected in str(excinfo.value)
@pytest.mark.parametrize(
("overrides", "expected"),
[
({"mesh": ""}, "mesh"),
({"group": ""}, "group"),
({"uv_map": ""}, "uv_map"),
({"texture": "tex.jpg"}, "texture"),
({"texture": "nested/tex.png"}, "texture"),
({"texture": ""}, "texture"),
({"radius_mm": 0.0}, "radius_mm"),
({"radius_mm": -1.0}, "radius_mm"),
({"radius_mm": True}, "radius_mm"),
({"falloff": "bounce"}, "falloff"),
({"step_mm": 0.0}, "step_mm"),
({"step_mm": -0.01}, "step_mm"),
({"max_mm": 0.0}, "max_mm"),
({"step_mm": 0.2, "max_mm": 0.1}, "max_mm"),
({"axes": []}, "axes"),
({"axes": ["w"]}, "axes"),
({"axes": ["x", "x"]}, "axes"),
({"axes": "x,y"}, "axes"),
({"resolution": [0, 500]}, "resolution.width"),
({"resolution": [500]}, "resolution"),
({"resolution": [500, 32769]}, "resolution.height"),
({"process_resolution": ["a", "b"]}, "process_resolution"),
({"tracker": {"algorithm": ""}}, "tracker.algorithm"),
({"tracker": {"dis_preset": ""}}, "tracker.dis_preset"),
({"tracker": {"dis_mode": -1}}, "tracker.dis_mode"),
({"tracker": {"dis_mode": "1"}}, "tracker.dis_mode"),
({"render": {"engine": ""}}, "render.engine"),
({"render": {"samples": 0}}, "render.samples"),
({"render": {"denoise": "no"}}, "render.denoise"),
({"render": {"view_transform": "AgX"}}, "view_transform"),
({"render": {"interpolate": ""}}, "render.interpolate"),
({"render": {"seed": "x"}}, "render.seed"),
],
)
def test_invalid_overrides_are_rejected(
paths: RevisionPaths, overrides: dict, expected: str
) -> None:
with pytest.raises(ProbeConfigError) as excinfo:
resolve_probe(paths, overrides)
assert expected in str(excinfo.value)
def test_validation_reports_every_problem(paths: RevisionPaths) -> None:
block = merge_probe(paths, {"radius_mm": 0.0, "falloff": "bounce", "step_mm": -1.0})
errors = validate_probe(block, {"sensing_surface": "sensing_surface", "cam_sensor": "cam_sensor"})
assert len([line for line in errors if line]) == 3
with pytest.raises(ProbeConfigError) as excinfo:
resolve_probe(paths, {"radius_mm": 0.0, "falloff": "bounce"})
message = str(excinfo.value)
assert "radius_mm" in message
assert "falloff" in message
def test_agx_view_transform_fails_loudly(paths: RevisionPaths) -> None:
"""AgX/Filmic silently destroys marker values; the contract forbids it."""
with pytest.raises(ProbeConfigError) as excinfo:
resolve_probe(paths, {"render": {"view_transform": "Filmic"}})
assert "Standard" in str(excinfo.value)
# ------------------------------------------------------------ frame plan
def test_frame_plan_default_has_31_ordered_frames(paths: RevisionPaths) -> None:
frames = frame_plan(resolve_probe(paths))
assert len(frames) == 1 + 3 * 10 == 31
assert [frame.index for frame in frames] == list(range(31))
assert [frame.name for frame in frames] == [
f"{frame.index:03d}_" + (
"reference" if frame.axis is None else f"{frame.axis}_+{frame.step_mm:.2f}mm"
)
for frame in frames
]
assert frames[0] == ProbeFrame(0, "000_reference", None, 0.0)
assert frames[1] == ProbeFrame(1, "001_x_+0.01mm", "x", 0.01)
assert frames[10] == ProbeFrame(10, "010_x_+0.10mm", "x", 0.10)
assert frames[11] == ProbeFrame(11, "011_y_+0.01mm", "y", 0.01)
assert frames[20] == ProbeFrame(20, "020_y_+0.10mm", "y", 0.10)
assert frames[21] == ProbeFrame(21, "021_z_+0.01mm", "z", 0.01)
assert frames[30] == ProbeFrame(30, "030_z_+0.10mm", "z", 0.10)
def test_frame_plan_is_axis_major_then_step_minor(paths: RevisionPaths) -> None:
frames = frame_plan(resolve_probe(paths))
moved = frames[1:]
assert [frame.axis for frame in moved] == ["x"] * 10 + ["y"] * 10 + ["z"] * 10
for axis in ("x", "y", "z"):
steps = [frame.step_mm for frame in moved if frame.axis == axis]
assert steps == pytest.approx([0.01 * n for n in range(1, 11)])
def test_frame_plan_single_axis_and_custom_steps(paths: RevisionPaths) -> None:
settings = dataclasses.replace(
resolve_probe(paths), axes=("z",), step_mm=0.05, max_mm=0.2
)
frames = frame_plan(settings)
assert len(frames) == 1 + round(0.2 / 0.05) == 5
assert [frame.name for frame in frames] == [
"000_reference",
"001_z_+0.05mm",
"002_z_+0.10mm",
"003_z_+0.15mm",
"004_z_+0.20mm",
]
def test_frame_plan_never_exceeds_max_mm(paths: RevisionPaths) -> None:
settings = dataclasses.replace(
resolve_probe(paths), axes=("x",), step_mm=0.03, max_mm=0.10
)
frames = frame_plan(settings)
assert [frame.step_mm for frame in frames] == pytest.approx([0.0, 0.03, 0.06, 0.09])
assert all(frame.step_mm <= settings.max_mm for frame in frames)
def test_frame_plan_names_keep_sub_centistep_precision(paths: RevisionPaths) -> None:
settings = dataclasses.replace(
resolve_probe(paths), axes=("x",), step_mm=0.005, max_mm=0.01
)
assert [frame.name for frame in frame_plan(settings)] == [
"000_reference",
"001_x_+0.005mm",
"002_x_+0.010mm",
]
def test_frame_plan_to_job_entry() -> None:
frame = ProbeFrame(10, "010_x_+0.10mm", "x", 0.10)
assert frame.to_job_entry() == {
"index": 10,
"name": "010_x_+0.10mm",
"axis": "x",
"step_mm": 0.10,
}
@pytest.mark.parametrize(
"changes",
[
{"step_mm": 0.0},
{"max_mm": 0.0},
{"axes": ()},
],
)
def test_frame_plan_rejects_degenerate_settings(
paths: RevisionPaths, changes: dict
) -> None:
settings = dataclasses.replace(resolve_probe(paths), **changes)
with pytest.raises(ProbeConfigError):
frame_plan(settings)
# --------------------------------------------------------------- paths
def test_output_dir_is_revision_scoped(paths: RevisionPaths) -> None:
root = repo_root()
assert probe_revision_rel(root, paths).as_posix() == "finger-xynova/r01"
assert probe_output_dir(root, paths) == (
root / "build" / "probe" / "finger-xynova" / "r01"
).resolve()
def test_output_dir_explicit_override_wins(
paths: RevisionPaths, tmp_path: Path
) -> None:
explicit = tmp_path / "elsewhere"
assert resolve_probe_output_dir(repo_root(), paths, explicit) == explicit.resolve()
assert resolve_probe_output_dir(repo_root(), paths) == probe_output_dir(
repo_root(), paths
)
def test_output_dir_uses_directory_names_not_design_sensor_field(
tmp_path: Path,
) -> None:
revision = _write_revision(
tmp_path, sensor_field="finger", sensor_dir="finger-xynova", revision_name="r01"
)
paths = RevisionPaths.from_revision_dir(revision)
assert probe_revision_rel(tmp_path, paths) == Path("finger-xynova") / "r01"
def test_artifact_paths_and_job_path(tmp_path: Path) -> None:
artifacts = probe_artifact_paths(tmp_path / "out")
assert set(artifacts) == {"frames", "report"}
assert artifacts["frames"] == (tmp_path / "out" / FRAMES_NPZ_FILENAME).resolve()
assert artifacts["report"] == (tmp_path / "out" / DEFORM_REPORT_FILENAME).resolve()
assert deform_job_path(tmp_path / "out") == tmp_path / "out" / DEFORM_JOB_FILENAME
# --------------------------------------------------------------- job I/O
def test_job_round_trips_with_absolute_paths(
paths: RevisionPaths, tmp_path: Path
) -> None:
settings = resolve_probe(paths)
output_dir = tmp_path / "probe"
job = build_deform_job(paths, settings, output_dir)
assert job["schema"] == DEFORM_JOB_SCHEMA
assert set(job) == {
"schema",
"revision_dir",
"source_blend",
"output_dir",
"objects",
"settings",
"frames",
"artifacts",
}
assert Path(job["revision_dir"]).is_absolute()
assert Path(job["source_blend"]).is_absolute()
assert Path(job["output_dir"]).is_absolute()
assert job["objects"] == {"mesh": "sensing_surface", "group": "probe"}
assert job["settings"] == {
"radius_mm": 0.5,
"falloff": "smooth",
"step_mm": 0.01,
"max_mm": 0.10,
"axes": ["x", "y", "z"],
"pattern": {
"nx": 80,
"ny": 80,
"seed": 7,
"vector_width_mm": 18.5,
"vector_height_mm": 16.4,
"cell_mm": [0.23125, 0.205],
},
}
assert job["frames"] == [frame.to_job_entry() for frame in frame_plan(settings)]
assert len(job["frames"]) == 31
assert job["artifacts"] == {
"frames": str((output_dir / FRAMES_NPZ_FILENAME).resolve()),
"report": str((output_dir / DEFORM_REPORT_FILENAME).resolve()),
}
assert all(Path(path).is_absolute() for path in job["artifacts"].values())
job_path = write_deform_job(job, output_dir)
assert job_path == output_dir / DEFORM_JOB_FILENAME
loaded = json.loads(job_path.read_text(encoding="utf-8"))
assert loaded == job
assert loaded["frames"][10] == {
"index": 10,
"name": "010_x_+0.10mm",
"axis": "x",
"step_mm": 0.10,
}
def test_write_job_creates_output_directory(paths: RevisionPaths, tmp_path: Path) -> None:
settings = resolve_probe(paths)
nested = tmp_path / "a" / "b"
write_deform_job(build_deform_job(paths, settings, nested), nested)
assert (nested / DEFORM_JOB_FILENAME).is_file()
# -------------------------------------------------------- CLI overrides
def test_probe_overrides_from_args() -> None:
overrides = probe_overrides_from_args(
_probe_namespace(
mesh="m",
group="g",
radius_mm=0.25,
falloff="sharp",
step_mm=0.02,
max_mm=0.08,
axes="x, z",
)
)
assert overrides == {
"mesh": "m",
"group": "g",
"radius_mm": 0.25,
"falloff": "sharp",
"step_mm": 0.02,
"max_mm": 0.08,
"axes": ["x", "z"],
}
def test_probe_overrides_from_args_unset_stays_none() -> None:
overrides = probe_overrides_from_args(_probe_namespace())
assert overrides["axes"] is None
merged = merge_probe(None, overrides)
assert merged["radius_mm"] == DEFAULT_PROBE["radius_mm"]
assert merged["axes"] == ["x", "y", "z"]
def test_probe_overrides_from_args_accepts_axis_sequence() -> None:
overrides = probe_overrides_from_args(_probe_namespace(axes=["z", "y"]))
assert overrides["axes"] == ["z", "y"]
# ------------------------------------------------------ stage consumers
def test_stage_settings_blocks_match_the_contract(paths: RevisionPaths) -> None:
settings = resolve_probe(paths)
design = _design_probe(paths)
width, height = _as_pair(design["resolution"])
assert settings.to_render_settings(Path("textures") / "patch_texture_compute.png") == {
"uv_map": "compute",
"texture": str((Path("textures") / "patch_texture_compute.png").resolve()),
"width": width,
"height": height,
"engine": "CYCLES",
"samples": 16,
"denoise": False,
"view_transform": "Standard",
"interpolate": "Closest",
"seed": 0,
}
assert settings.to_track_settings() == {
"axes": ["x", "y", "z"],
"step_mm": 0.01,
"max_mm": 0.10,
"radius_mm": 0.5,
"process_resolution": list(_as_pair(design["process_resolution"])),
"tracker": dict(design["tracker"]),
# Additive: the render-seed floor is a reproducibility diagnostic, and this
# block is what lets the estimator measure signal and noise under one model.
"noise": {
"idle_footage": "../data/dd02-0703/190.mp4",
"roi_fraction": 0.18,
"max_frames": 120,
"sigmas": [4.0, 16.0],
"draws": 1,
"seed": 0,
"models": ["render_seed", "injected"],
"scales": ["render"],
},
}