feat(perception): qualify E31 source binding
This commit is contained in:
DCCONSTRUCTIONS
@@ -0,0 +1,238 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
import numpy as np
|
||||
import pytest
|
||||
|
||||
from k1link.compute.e31_source_qualification import (
|
||||
E31SourceQualificationError,
|
||||
E31SourceQualificationProfile,
|
||||
_E30Chain,
|
||||
_offset_sweep,
|
||||
_self_mask_report,
|
||||
)
|
||||
from k1link.compute.lidar_local_surface import POINT_OCCUPIED
|
||||
|
||||
|
||||
class _SweepSource:
|
||||
def __init__(self) -> None:
|
||||
self.arrays = {
|
||||
"session_seconds": np.asarray([0.0, 0.1, 0.2], dtype=np.float64),
|
||||
"sample_available": np.ones(3, dtype=np.bool_),
|
||||
"cloud_offsets": np.asarray([0, 2, 4, 6], dtype=np.int64),
|
||||
"cloud_points_map": np.asarray(
|
||||
[
|
||||
[1.0, 0.0, 2.0],
|
||||
[1.1, 0.0, 2.0],
|
||||
[0.0, 0.0, 2.0],
|
||||
[0.1, 0.0, 2.0],
|
||||
[-1.0, 0.0, 2.0],
|
||||
[-1.1, 0.0, 2.0],
|
||||
],
|
||||
dtype=np.float32,
|
||||
),
|
||||
"pose_positions_map": np.zeros((3, 3), dtype=np.float64),
|
||||
"pose_quaternions_map_from_lidar": np.asarray(
|
||||
[[0.0, 0.0, 0.0, 1.0]] * 3,
|
||||
dtype=np.float64,
|
||||
),
|
||||
"intrinsic_fx_fy_cx_cy": np.asarray(
|
||||
[100.0, 100.0, 50.0, 50.0],
|
||||
dtype=np.float64,
|
||||
),
|
||||
"distortion_kb4": np.zeros(4, dtype=np.float64),
|
||||
"t_camera_from_lidar": np.eye(4, dtype=np.float64),
|
||||
}
|
||||
self.identity: dict[str, Any] = {
|
||||
"source_id": "sensor.camera.right",
|
||||
"camera_slot": "camera_1",
|
||||
"projection": {"width": 100, "height": 100},
|
||||
}
|
||||
|
||||
|
||||
class _SweepSurface:
|
||||
def __init__(self) -> None:
|
||||
self.arrays = {
|
||||
"point_class": np.full(6, POINT_OCCUPIED, dtype=np.uint8),
|
||||
}
|
||||
|
||||
|
||||
def _artifact(path: Path) -> dict[str, object]:
|
||||
return {
|
||||
"path": path.name,
|
||||
"byte_length": path.stat().st_size,
|
||||
"sha256": hashlib.sha256(path.read_bytes()).hexdigest(),
|
||||
}
|
||||
|
||||
|
||||
def _projection_artifact(
|
||||
root: Path,
|
||||
name: str,
|
||||
pixels: list[list[float]],
|
||||
) -> dict[str, object]:
|
||||
path = root / f"{name}.npz"
|
||||
np.savez(
|
||||
path,
|
||||
projected_pixels_xy=np.asarray(pixels, dtype=np.float32),
|
||||
projected_selected_mask=np.ones(len(pixels), dtype=np.uint8),
|
||||
)
|
||||
return _artifact(path)
|
||||
|
||||
|
||||
def test_profile_rejects_ambiguous_offset_hypotheses() -> None:
|
||||
with pytest.raises(E31SourceQualificationError):
|
||||
E31SourceQualificationProfile(offset_hypotheses_ms=(0, -50, 50))
|
||||
with pytest.raises(E31SourceQualificationError):
|
||||
E31SourceQualificationProfile(offset_hypotheses_ms=(-50, 50))
|
||||
|
||||
|
||||
def test_offset_sweep_keeps_evidenced_zero_binding() -> None:
|
||||
source = _SweepSource()
|
||||
surface = _SweepSurface()
|
||||
profile = E31SourceQualificationProfile(
|
||||
offset_hypotheses_ms=(-100, 0, 100),
|
||||
minimum_correspondence_items=1,
|
||||
)
|
||||
item = {
|
||||
"item_id": "item-1",
|
||||
"review_key": "semantic:1:1",
|
||||
"evidence_binding": {"frame_index": 1},
|
||||
"e29_snapshot": {"bbox_xyxy": [40.0, 40.0, 60.0, 60.0]},
|
||||
}
|
||||
|
||||
sweep, rows = _offset_sweep(
|
||||
source=source, # type: ignore[arg-type]
|
||||
surface=surface, # type: ignore[arg-type]
|
||||
items=(item,),
|
||||
profile=profile,
|
||||
bbox_inset_fraction=0.03,
|
||||
)
|
||||
|
||||
assert sweep["selected_offset_ms"] == 0
|
||||
assert sweep["baseline_supported_fraction"] == 1.0
|
||||
assert sweep["best_supported_fraction"] == 1.0
|
||||
assert sweep["baseline_support_deficit_fraction"] == 0.0
|
||||
assert [row["supported_count"] for row in sweep["hypotheses"]] == [0, 1, 0]
|
||||
assert [score["candidate_frame_index"] for score in rows[0]["scores"]] == [
|
||||
0,
|
||||
1,
|
||||
2,
|
||||
]
|
||||
|
||||
|
||||
def test_self_mask_admits_only_non_colliding_semantic_rule(
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
items: list[dict[str, Any]] = []
|
||||
decisions: list[dict[str, Any]] = []
|
||||
for index in range(4):
|
||||
item = {
|
||||
"item_id": f"semantic-self-{index}",
|
||||
"e29_snapshot": {
|
||||
"label": "person",
|
||||
"bbox_xyxy": [
|
||||
20.0 + index,
|
||||
75.0,
|
||||
40.0 + index,
|
||||
99.0,
|
||||
],
|
||||
},
|
||||
}
|
||||
items.append(item)
|
||||
decisions.append(
|
||||
{
|
||||
"item_id": item["item_id"],
|
||||
"cause_code": "self_points",
|
||||
"point_ownership": "self",
|
||||
}
|
||||
)
|
||||
|
||||
geometry_self = {
|
||||
"item_id": "geometry-self",
|
||||
"e29_snapshot": {},
|
||||
"artifact": _projection_artifact(
|
||||
tmp_path,
|
||||
"geometry-self",
|
||||
[[20.0, 20.0], [30.0, 30.0]],
|
||||
),
|
||||
}
|
||||
accepted_object = {
|
||||
"item_id": "accepted-object",
|
||||
"e29_snapshot": {},
|
||||
"artifact": _projection_artifact(
|
||||
tmp_path,
|
||||
"accepted-object",
|
||||
[[25.0, 25.0], [80.0, 80.0]],
|
||||
),
|
||||
}
|
||||
accepted_person = {
|
||||
"item_id": "accepted-person",
|
||||
"e29_snapshot": {
|
||||
"label": "person",
|
||||
"bbox_xyxy": [70.0, 50.0, 90.0, 90.0],
|
||||
},
|
||||
"artifact": _projection_artifact(
|
||||
tmp_path,
|
||||
"accepted-person",
|
||||
[[80.0, 80.0]],
|
||||
),
|
||||
}
|
||||
items.extend([geometry_self, accepted_object, accepted_person])
|
||||
decisions.extend(
|
||||
[
|
||||
{
|
||||
"item_id": geometry_self["item_id"],
|
||||
"cause_code": "self_points",
|
||||
"point_ownership": "self",
|
||||
},
|
||||
{
|
||||
"item_id": accepted_object["item_id"],
|
||||
"cause_code": "none",
|
||||
"point_ownership": "object",
|
||||
},
|
||||
{
|
||||
"item_id": accepted_person["item_id"],
|
||||
"cause_code": "none",
|
||||
"point_ownership": "object",
|
||||
},
|
||||
]
|
||||
)
|
||||
chain = _E30Chain(
|
||||
materialization_manifest={
|
||||
"identity": {
|
||||
"projection": {
|
||||
"width": 100,
|
||||
"height": 100,
|
||||
}
|
||||
}
|
||||
},
|
||||
items=tuple(items),
|
||||
engineering_manifest={},
|
||||
decisions=tuple(decisions),
|
||||
exceptions=(),
|
||||
human_manifest={},
|
||||
human_decisions=(),
|
||||
)
|
||||
|
||||
report = _self_mask_report(
|
||||
materialization_root=tmp_path,
|
||||
chain=chain,
|
||||
profile=E31SourceQualificationProfile(
|
||||
minimum_semantic_self_samples=4,
|
||||
),
|
||||
)
|
||||
|
||||
assert report["semantic_mask"]["status"] == "admitted"
|
||||
assert report["semantic_mask"]["application_rule"] == "bbox-center-inside-rectangle"
|
||||
assert report["semantic_mask"]["collateral_item_count"] == 0
|
||||
assert report["geometry_point_mask"]["status"] == "rejected"
|
||||
assert report["geometry_point_mask"]["collateral"] == [
|
||||
{
|
||||
"item_id": "accepted-object",
|
||||
"masked_selected_point_count": 1,
|
||||
}
|
||||
]
|
||||
assert report["exact_correction_item_ids"] == ["geometry-self"]
|
||||
Reference in New Issue
Block a user