ykcheungab
230b04a79d
Product name decided the pipeline: Mini worked live but replay refused it. A bundle descriptor now declares profile_id, product_family and processing_backend, with computational and visualization assets kept apart, and one resolver serves live and recorded input. Legacy bundles load through a documented adapter that never guesses a backend from a folder name.
459 lines
17 KiB
Python
459 lines
17 KiB
Python
"""Tests for the sensor bundle contract: descriptor, resolver, registry, fingerprints.
|
|
|
|
Each test states the property it protects, because the point of this layer is that a product name
|
|
stops deciding which pipeline runs. The bundles built here are small: a resolver test does not need
|
|
a real camera configuration, only the specific files the rule under test reads.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import shutil
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from orisys.sensors import (
|
|
BACKEND_COMPOUND_EYE,
|
|
BACKEND_SINGLE_CAMERA,
|
|
BACKEND_UV_UNROLL,
|
|
BUILT_IN_PROFILES,
|
|
CalibrationSelection,
|
|
DescriptorError,
|
|
ProcessorCreationError,
|
|
backend_from_config,
|
|
create_processor,
|
|
known_backends,
|
|
parse_descriptor,
|
|
resolve_bundle,
|
|
spec_for,
|
|
validate_profile,
|
|
)
|
|
|
|
REPO_ROOT = Path(__file__).resolve().parents[1]
|
|
SDK_SENSORS = REPO_ROOT / "sensors"
|
|
|
|
MINIMAL_CONFIG = {
|
|
"name": "prototype",
|
|
"version": "1.0",
|
|
"type": "compound_eye",
|
|
"camera": {"resolution_stitch": [64, 64], "fps": 30},
|
|
"stitcher": {"index": [0]},
|
|
"tracker": {"algorithm": "dis"},
|
|
"contact": {},
|
|
}
|
|
|
|
|
|
def _write_bundle(
|
|
root: Path,
|
|
*,
|
|
descriptor: dict | None = None,
|
|
config: dict | None = None,
|
|
extra: dict[str, bytes] | None = None,
|
|
legacy_manifest: dict | None = None,
|
|
) -> Path:
|
|
"""A bundle folder: ``sensor.json`` plus whichever manifest shape is asked for."""
|
|
root.mkdir(parents=True, exist_ok=True)
|
|
(root / "sensor.json").write_text(
|
|
json.dumps(config if config is not None else MINIMAL_CONFIG), encoding="utf-8"
|
|
)
|
|
if descriptor is not None:
|
|
(root / "manifest.json").write_text(json.dumps(descriptor), encoding="utf-8")
|
|
elif legacy_manifest is not None:
|
|
(root / "manifest.json").write_text(
|
|
json.dumps(legacy_manifest), encoding="utf-8"
|
|
)
|
|
for relative, payload in (extra or {}).items():
|
|
target = root / relative
|
|
target.parent.mkdir(parents=True, exist_ok=True)
|
|
target.write_bytes(payload)
|
|
return root
|
|
|
|
|
|
def _descriptor(**overrides) -> dict:
|
|
payload = {
|
|
"schema_version": 1,
|
|
"profile_id": "test.profile",
|
|
"display_name": "Test Profile",
|
|
"product_family": "mini",
|
|
"processing_backend": BACKEND_COMPOUND_EYE,
|
|
"sensor_json": "sensor.json",
|
|
}
|
|
payload.update(overrides)
|
|
return payload
|
|
|
|
|
|
# ------------------------------------------------------------------ descriptor
|
|
|
|
|
|
def test_descriptor_is_required_to_identify_an_unknown_folder(tmp_path: Path):
|
|
"""A folder name is not evidence of a backend."""
|
|
root = _write_bundle(tmp_path / "mystery")
|
|
with pytest.raises(DescriptorError) as excinfo:
|
|
resolve_bundle(root)
|
|
assert "not a built-in bundle" in str(excinfo.value)
|
|
|
|
|
|
def test_descriptor_rejects_unknown_backend_and_schema_version(tmp_path: Path):
|
|
root = _write_bundle(
|
|
tmp_path / "b", descriptor=_descriptor(processing_backend="nonsense")
|
|
)
|
|
with pytest.raises(DescriptorError) as excinfo:
|
|
parse_descriptor(root)
|
|
assert "unsupported processing_backend" in str(excinfo.value)
|
|
|
|
root2 = _write_bundle(tmp_path / "c", descriptor=_descriptor(schema_version=99))
|
|
with pytest.raises(DescriptorError) as excinfo2:
|
|
parse_descriptor(root2)
|
|
assert "unsupported descriptor schema_version" in str(excinfo2.value)
|
|
|
|
|
|
def test_descriptor_rejects_paths_that_leave_the_bundle(tmp_path: Path):
|
|
root = _write_bundle(
|
|
tmp_path / "d",
|
|
descriptor=_descriptor(computational_assets={"map": "../outside.npz"}),
|
|
)
|
|
with pytest.raises(DescriptorError) as excinfo:
|
|
parse_descriptor(root)
|
|
assert "escapes the bundle folder" in str(excinfo.value)
|
|
|
|
root2 = _write_bundle(
|
|
tmp_path / "e",
|
|
descriptor=_descriptor(computational_assets={"map": str(tmp_path / "abs.npz")}),
|
|
)
|
|
with pytest.raises(DescriptorError) as excinfo2:
|
|
parse_descriptor(root2)
|
|
assert "absolute" in str(excinfo2.value)
|
|
|
|
|
|
def test_descriptor_round_trips_with_relative_paths(tmp_path: Path):
|
|
root = _write_bundle(
|
|
tmp_path / "rt",
|
|
descriptor=_descriptor(
|
|
computational_assets={"map": "assets/map.npz"},
|
|
assets={"scene": "scene/x.oryscene"},
|
|
),
|
|
extra={"assets/map.npz": b"payload", "scene/x.oryscene": b"scene"},
|
|
)
|
|
descriptor = parse_descriptor(root)
|
|
payload = descriptor.as_dict()
|
|
assert payload["computational_assets"] == {"map": "assets/map.npz"}
|
|
assert payload["assets"] == {"scene": "scene/x.oryscene"}
|
|
assert not Path(payload["sensor_json"]).is_absolute()
|
|
|
|
|
|
# ------------------------------------------------------------------ legacy adapter
|
|
|
|
|
|
@pytest.mark.parametrize("folder", sorted(BUILT_IN_PROFILES))
|
|
def test_every_built_in_resolves_through_compatibility(folder: str):
|
|
"""The acceptance rule: existing supported bundle formats keep loading."""
|
|
root = SDK_SENSORS / folder
|
|
if not root.is_dir():
|
|
pytest.skip(f"{folder} is not installed in this checkout")
|
|
profile = resolve_bundle(root)
|
|
assert profile.product_family == BUILT_IN_PROFILES[folder]["product_family"]
|
|
assert profile.processing_backend in known_backends()
|
|
assert profile.validation.state in ("ok", "degraded", "failed")
|
|
assert profile.processing_fingerprint
|
|
|
|
|
|
def test_backend_follows_the_configuration_not_the_folder_name():
|
|
"""finger's sensor.json says single_camera; its backend comes from the SDK's own rule."""
|
|
assert backend_from_config({"type": "single_camera"}) == BACKEND_SINGLE_CAMERA
|
|
assert backend_from_config({"stitcher": {}}) == BACKEND_COMPOUND_EYE
|
|
assert backend_from_config({}) == BACKEND_SINGLE_CAMERA
|
|
assert backend_from_config({"type": "uv_unroll"}) == BACKEND_UV_UNROLL
|
|
with pytest.raises(DescriptorError):
|
|
backend_from_config({"type": "made_up"})
|
|
|
|
|
|
def test_single_camera_is_never_read_as_a_product_family():
|
|
"""``single_camera`` describes the image arrangement, not what the product is."""
|
|
profile = resolve_bundle(SDK_SENSORS / "finger")
|
|
assert profile.product_family == "finger"
|
|
assert profile.processing_backend == BACKEND_UV_UNROLL
|
|
|
|
|
|
def test_legacy_manifest_is_read_by_the_adapter_not_the_strict_parser(tmp_path: Path):
|
|
"""finger ships an old-shape manifest; it must not be rejected for lacking schema_version."""
|
|
root = _write_bundle(
|
|
tmp_path / "finger",
|
|
legacy_manifest={
|
|
"name": "finger",
|
|
"sensor_json": "sensor.json",
|
|
"assets_dir": "assets/30fps-0703",
|
|
"viz_pose": "assets/visualization/world_to_cam.npy",
|
|
},
|
|
config={**MINIMAL_CONFIG, "type": "uv_unroll"},
|
|
extra={"assets/visualization/world_to_cam.npy": b"pose"},
|
|
)
|
|
profile = resolve_bundle(root)
|
|
assert profile.product_family == "finger"
|
|
# Declared uv_unroll without the map: an error, not a silent downgrade.
|
|
assert profile.validation.state == "failed"
|
|
assert any(
|
|
issue.code == "assets.computational-undeclared"
|
|
for issue in profile.validation.errors
|
|
)
|
|
# The viz pose is picked up as a visualization asset, never as computational.
|
|
assert "viz_pose" in profile.assets
|
|
assert "viz_pose" not in profile.computational_assets
|
|
|
|
|
|
# ------------------------------------------------------------------ backend registry
|
|
|
|
|
|
def test_registry_refuses_an_unknown_identifier():
|
|
with pytest.raises(ProcessorCreationError) as excinfo:
|
|
spec_for("nonsense")
|
|
assert "unsupported processing backend" in str(excinfo.value)
|
|
|
|
|
|
def test_mini_uses_its_supported_backend_without_pretending_to_be_box():
|
|
profile = resolve_bundle(SDK_SENSORS / "mini")
|
|
assert profile.product_family == "mini"
|
|
assert profile.processing_backend == BACKEND_COMPOUND_EYE
|
|
assert profile.profile_id != resolve_bundle(SDK_SENSORS / "box").profile_id
|
|
|
|
|
|
def test_finger_mems_is_assessed_from_its_configuration(tmp_path: Path):
|
|
"""Its name must not imply MEMS acquisition support.
|
|
|
|
The studio-side ``finger-mems`` bundle copies ``box``'s configuration, so the folder name is
|
|
the only thing that makes it look special. Its backend has to come from that configuration.
|
|
"""
|
|
config = {**MINIMAL_CONFIG, "type": "compound_eye", "stitcher": {"index": [0]}}
|
|
root = _write_bundle(tmp_path / "finger-mems", config=config)
|
|
profile = resolve_bundle(root)
|
|
assert profile.product_family == "finger-mems"
|
|
assert profile.processing_backend == BACKEND_COMPOUND_EYE
|
|
# No calibration and no metadata: raw yes, calibrated no, per axis.
|
|
assert profile.capabilities.raw_force.available
|
|
assert not profile.capabilities.calibrated_force.any_available()
|
|
assert profile.capabilities.calibrated_force.fn.reason
|
|
# A UV backend is never inferred from the name.
|
|
assert profile.processing_backend != BACKEND_UV_UNROLL
|
|
|
|
|
|
def test_create_processor_names_what_is_missing_rather_than_faking_it():
|
|
"""uv_unroll needs a UV domain assembled outside a config-driven Sensor."""
|
|
profile = resolve_bundle(SDK_SENSORS / "finger")
|
|
with pytest.raises(ProcessorCreationError) as excinfo:
|
|
create_processor(profile)
|
|
assert "uv_unroll" in str(excinfo.value)
|
|
|
|
|
|
# ------------------------------------------------------------------ capabilities
|
|
|
|
|
|
def test_missing_optional_scene_does_not_block_processing(tmp_path: Path):
|
|
root = _write_bundle(
|
|
tmp_path / "sceneless",
|
|
descriptor=_descriptor(
|
|
computational_assets={"map": "assets/map.npz"},
|
|
assets={"scene": "scene/absent.oryscene"},
|
|
),
|
|
extra={"assets/map.npz": b"map"},
|
|
)
|
|
profile = resolve_bundle(root)
|
|
assert profile.validation.usable
|
|
assert profile.capabilities.video_processing.available
|
|
assert profile.capabilities.optical_flow.available
|
|
assert not profile.capabilities.visualization_3d.available
|
|
assert "missing" in profile.capabilities.visualization_3d.reason
|
|
|
|
|
|
def test_missing_computational_asset_fails_the_profile(tmp_path: Path):
|
|
root = _write_bundle(
|
|
tmp_path / "nogeom",
|
|
descriptor=_descriptor(computational_assets={"map": "assets/absent.npz"}),
|
|
)
|
|
profile = resolve_bundle(root)
|
|
assert profile.validation.state == "failed"
|
|
assert any(
|
|
issue.code == "assets.computational-missing"
|
|
for issue in profile.validation.errors
|
|
)
|
|
|
|
|
|
def test_single_camera_reports_no_force_capability(tmp_path: Path):
|
|
config = {key: value for key, value in MINIMAL_CONFIG.items() if key != "stitcher"}
|
|
config["type"] = "single_camera"
|
|
root = _write_bundle(
|
|
tmp_path / "mono",
|
|
descriptor=_descriptor(
|
|
processing_backend=BACKEND_SINGLE_CAMERA, product_family="custom"
|
|
),
|
|
config=config,
|
|
)
|
|
profile = resolve_bundle(root)
|
|
assert profile.capabilities.video_processing.available
|
|
assert not profile.capabilities.raw_force.available
|
|
assert "single_camera" in profile.capabilities.raw_force.reason
|
|
|
|
|
|
# ------------------------------------------------------------------ calibration
|
|
|
|
|
|
def test_explicit_none_is_a_choice_not_a_failure(tmp_path: Path):
|
|
root = _write_bundle(tmp_path / "nocal", descriptor=_descriptor())
|
|
profile = resolve_bundle(root, calibration_selection=CalibrationSelection.NONE)
|
|
assert profile.calibration.selection is CalibrationSelection.NONE
|
|
assert profile.calibration_fingerprint == "none"
|
|
assert not profile.capabilities.calibrated_force.any_available()
|
|
assert all(
|
|
"explicitly uncalibrated" in axis.reason
|
|
for axis in (
|
|
profile.capabilities.calibrated_force.fn,
|
|
profile.capabilities.calibrated_force.fx,
|
|
profile.capabilities.calibrated_force.fy,
|
|
)
|
|
)
|
|
|
|
|
|
def test_an_unreadable_selected_calibration_is_an_error_not_a_substitution(tmp_path: Path):
|
|
root = _write_bundle(tmp_path / "badcal", descriptor=_descriptor())
|
|
profile = resolve_bundle(
|
|
root, calibration_selection=str(tmp_path / "does-not-exist.json")
|
|
)
|
|
assert any(
|
|
issue.code == "calibration.missing" for issue in profile.validation.errors
|
|
)
|
|
assert profile.calibration.document is None
|
|
assert profile.validation.state == "failed"
|
|
|
|
|
|
def test_legacy_calibration_without_metadata_reports_unknown_applicability():
|
|
"""box ships a calibration with no identifying metadata; it applies, but is flagged."""
|
|
root = SDK_SENSORS / "box"
|
|
if not (root / "force_calibration.json").is_file():
|
|
pytest.skip("box ships no calibration in this checkout")
|
|
profile = resolve_bundle(root)
|
|
assert profile.calibration.applied
|
|
assert profile.calibration.applicability == "unknown"
|
|
assert any(
|
|
issue.code == "calibration.applicability-unknown"
|
|
for issue in profile.validation.warnings
|
|
)
|
|
|
|
|
|
def test_calibrated_force_is_reported_per_axis(tmp_path: Path):
|
|
calibration = {
|
|
"schema_version": 2,
|
|
"unit": "N",
|
|
"channels": {
|
|
"normal": {"method": "piecewise_linear", "points": [[0.0, 0.0], [10.0, 1.0]]}
|
|
},
|
|
}
|
|
root = _write_bundle(
|
|
tmp_path / "fnonly",
|
|
descriptor=_descriptor(default_calibration="force_calibration.json"),
|
|
extra={"force_calibration.json": json.dumps(calibration).encode()},
|
|
)
|
|
profile = resolve_bundle(root)
|
|
axes = profile.capabilities.calibrated_force
|
|
assert axes.fn.available
|
|
assert not axes.fx.available
|
|
assert not axes.fy.available
|
|
assert "no fn curve" not in axes.fx.detail # the detail names the axis that is missing
|
|
assert "no fx curve" in axes.fx.detail
|
|
|
|
|
|
# ------------------------------------------------------------------ fingerprints
|
|
|
|
|
|
def test_relocating_a_bundle_preserves_the_processing_fingerprint(tmp_path: Path):
|
|
"""Moving or renaming must not change what would be computed."""
|
|
first = _write_bundle(
|
|
tmp_path / "one" / "mini",
|
|
descriptor=_descriptor(computational_assets={"map": "assets/map.npz"}),
|
|
extra={"assets/map.npz": b"shared-payload"},
|
|
)
|
|
second = tmp_path / "elsewhere" / "renamed-profile"
|
|
shutil.copytree(first, second)
|
|
|
|
a = resolve_bundle(first)
|
|
b = resolve_bundle(second)
|
|
assert a.processing_fingerprint == b.processing_fingerprint
|
|
assert a.bundle_root != b.bundle_root
|
|
|
|
|
|
def test_changing_a_computational_asset_changes_the_fingerprint(tmp_path: Path):
|
|
root = _write_bundle(
|
|
tmp_path / "asset",
|
|
descriptor=_descriptor(computational_assets={"map": "assets/map.npz"}),
|
|
extra={"assets/map.npz": b"first"},
|
|
)
|
|
before = resolve_bundle(root).processing_fingerprint
|
|
(root / "assets" / "map.npz").write_bytes(b"second")
|
|
after = resolve_bundle(root).processing_fingerprint
|
|
assert before != after
|
|
|
|
|
|
def test_display_name_and_visualization_assets_stay_out_of_the_fingerprint(tmp_path: Path):
|
|
"""A cosmetic change must not invalidate a comparable record."""
|
|
def build(name: str, **descriptor_overrides) -> Path:
|
|
overrides = {"computational_assets": {"map": "assets/map.npz"}}
|
|
overrides.update(descriptor_overrides)
|
|
return _write_bundle(
|
|
tmp_path / name,
|
|
descriptor=_descriptor(**overrides),
|
|
extra={"assets/map.npz": b"map"},
|
|
)
|
|
|
|
plain = build("plain", display_name="One")
|
|
fancy = build(
|
|
"fancy",
|
|
display_name="A Much Nicer Name",
|
|
assets={"scene": "scene/absent.oryscene"},
|
|
)
|
|
assert (
|
|
resolve_bundle(plain).processing_fingerprint
|
|
== resolve_bundle(fancy).processing_fingerprint
|
|
)
|
|
|
|
|
|
def test_changing_the_configuration_changes_the_fingerprint(tmp_path: Path):
|
|
root = _write_bundle(tmp_path / "cfg", descriptor=_descriptor())
|
|
before = resolve_bundle(root).processing_fingerprint
|
|
changed = json.loads((root / "sensor.json").read_text(encoding="utf-8"))
|
|
changed["tracker"]["algorithm"] = "raft"
|
|
(root / "sensor.json").write_text(json.dumps(changed), encoding="utf-8")
|
|
after = resolve_bundle(root).processing_fingerprint
|
|
assert before != after
|
|
|
|
|
|
def test_renaming_the_config_does_not_change_the_fingerprint(tmp_path: Path):
|
|
"""``name``/``version`` identify the document; they do not change the numbers."""
|
|
root = _write_bundle(tmp_path / "named", descriptor=_descriptor())
|
|
before = resolve_bundle(root).processing_fingerprint
|
|
renamed = json.loads((root / "sensor.json").read_text(encoding="utf-8"))
|
|
renamed["name"] = "something-else"
|
|
renamed["version"] = "9.9"
|
|
(root / "sensor.json").write_text(json.dumps(renamed), encoding="utf-8")
|
|
assert resolve_bundle(root).processing_fingerprint == before
|
|
|
|
|
|
def test_validate_profile_flags_assets_that_changed_since_resolution(tmp_path: Path):
|
|
root = _write_bundle(
|
|
tmp_path / "stale",
|
|
descriptor=_descriptor(computational_assets={"map": "assets/map.npz"}),
|
|
extra={"assets/map.npz": b"original"},
|
|
)
|
|
profile = resolve_bundle(root)
|
|
assert validate_profile(profile).state in ("ok", "degraded")
|
|
|
|
(root / "assets" / "map.npz").write_bytes(b"tampered")
|
|
report = validate_profile(profile)
|
|
assert any(issue.code == "fingerprint.stale" for issue in report.errors)
|
|
|
|
|
|
def test_inventory_is_bundle_relative(tmp_path: Path):
|
|
profile = resolve_bundle(SDK_SENSORS / "box")
|
|
inventory = profile.inventory()
|
|
assert inventory["profile_id"] == profile.profile_id
|
|
assert inventory["processing_fingerprint"] == profile.processing_fingerprint
|
|
assert not str(inventory["sensor_config"]).startswith(str(profile.bundle_root))
|
|
assert "/" in inventory["sensor_config"] or "\\" in inventory["sensor_config"] or inventory["sensor_config"]
|