351 lines
11 KiB
Python
351 lines
11 KiB
Python
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
from typing import cast
|
|
|
|
import numpy as np
|
|
import pytest
|
|
|
|
from k1link.laboratory import m48_raw_evidence as raw_module
|
|
from k1link.laboratory.m48_object_quality import M48ObjectQualityPack
|
|
from k1link.laboratory.m48_raw_evidence import (
|
|
M48_EXPECTED_FRAME_COUNT,
|
|
M48_EXPECTED_SESSION_ID,
|
|
M48_EXPECTED_SOURCE_ID,
|
|
M48_RAW_SPATIAL_FRAME_SCHEMA,
|
|
M48RawEvidenceError,
|
|
M48RawEvidenceReader,
|
|
)
|
|
from k1link.perception.geometry import RecordedFrameTemporalBinding
|
|
from k1link.perception.threat import ReplayBodyFrame, load_replay_threat_profile
|
|
|
|
REPOSITORY_ROOT = Path(__file__).resolve().parents[1]
|
|
PROFILE_PATH = REPOSITORY_ROOT / "config/perception/m4-replay-threat-v3.json"
|
|
|
|
|
|
def _canonical_sha256(value: object) -> str:
|
|
return hashlib.sha256(
|
|
json.dumps(
|
|
value,
|
|
sort_keys=True,
|
|
separators=(",", ":"),
|
|
ensure_ascii=False,
|
|
).encode()
|
|
).hexdigest()
|
|
|
|
|
|
def _sealed_e10_pack(tmp_path: Path) -> tuple[Path, str, str]:
|
|
payload = b"sealed-e10-lidar-pack"
|
|
artifact_sha256 = hashlib.sha256(payload).hexdigest()
|
|
identity = {
|
|
"available_lidar_frames": 3928,
|
|
"calibration_sha256": "1" * 64,
|
|
"camera_slot": "camera_1",
|
|
"e6_profile_sha256": "2" * 64,
|
|
"e6_result_id": "e6-fixture",
|
|
"frame_count": M48_EXPECTED_FRAME_COUNT,
|
|
"input_sha256": "3" * 64,
|
|
"job_id": "recorded-camera-fixture",
|
|
"point_count": 5,
|
|
"producer_sha256": "4" * 64,
|
|
"projection": {
|
|
"height": 600,
|
|
"model": "kb4",
|
|
"source_coordinates": "k1-map",
|
|
"target_camera": "sensor.camera.right",
|
|
"width": 800,
|
|
},
|
|
"schema_version": "missioncore.e10-lidar-replay-pack/v1",
|
|
"semantic_timeline_result_id": "result-fixture",
|
|
"session_id": M48_EXPECTED_SESSION_ID,
|
|
"source_end_frame_index": M48_EXPECTED_FRAME_COUNT - 1,
|
|
"source_id": "sensor.camera.right",
|
|
"source_start_frame_index": 0,
|
|
"temporal_binding": "accepted-e6-nearest-host-arrival-best-effort",
|
|
"temporal_policy": {
|
|
"binding": "nearest-host-arrival-best-effort",
|
|
"clock_source": "recorded-host-monotonic-arrival",
|
|
"maximum_lidar_camera_delta_ms": 100.0,
|
|
"maximum_pose_point_delta_ms": 100.0,
|
|
},
|
|
"timeline_end_seconds": 484.0,
|
|
"timeline_start_seconds": 35.0,
|
|
}
|
|
identity_sha256 = _canonical_sha256(identity)
|
|
pack_id = f"e10-lidar-pack-{identity_sha256}"
|
|
root = tmp_path / pack_id
|
|
root.mkdir()
|
|
(root / "lidar-pack.npz").write_bytes(payload)
|
|
manifest = {
|
|
"artifact": {
|
|
"byte_length": len(payload),
|
|
"media_type": "application/x-npz",
|
|
"path": "lidar-pack.npz",
|
|
"sha256": artifact_sha256,
|
|
},
|
|
"classification": "private-recorded-sensor-replay-input",
|
|
"created_at_utc": "2026-07-22T06:05:22.515Z",
|
|
"ground_truth": False,
|
|
"identity": identity,
|
|
"identity_sha256": identity_sha256,
|
|
"pack_id": pack_id,
|
|
"schema_version": "missioncore.e10-lidar-replay-pack/v1",
|
|
}
|
|
(root / "manifest.json").write_text(
|
|
json.dumps(manifest, sort_keys=True, separators=(",", ":")),
|
|
encoding="utf-8",
|
|
)
|
|
return root, pack_id, artifact_sha256
|
|
|
|
|
|
def test_e10_pack_validation_binds_identity_path_length_and_sha256(tmp_path: Path) -> None:
|
|
root, pack_id, artifact_sha256 = _sealed_e10_pack(tmp_path)
|
|
|
|
artifact = raw_module._validate_e10_pack(
|
|
root,
|
|
expected_pack_id=pack_id,
|
|
expected_artifact_sha256=artifact_sha256,
|
|
)
|
|
|
|
assert artifact == (root / "lidar-pack.npz").resolve()
|
|
|
|
(root / "lidar-pack.npz").write_bytes(b"tampered")
|
|
with pytest.raises(M48RawEvidenceError, match="artifact content changed"):
|
|
raw_module._validate_e10_pack(
|
|
root,
|
|
expected_pack_id=pack_id,
|
|
expected_artifact_sha256=artifact_sha256,
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("mutation", "message"),
|
|
[
|
|
(lambda manifest: manifest["identity"].update(session_id="other"), "identity changed"),
|
|
(
|
|
lambda manifest: manifest["artifact"].update(path="../lidar-pack.npz"),
|
|
"identity changed",
|
|
),
|
|
(lambda manifest: manifest.update(pack_id="e10-lidar-pack-wrong"), "identity changed"),
|
|
],
|
|
)
|
|
def test_e10_pack_validation_rejects_manifest_escape(
|
|
tmp_path: Path,
|
|
mutation: object,
|
|
message: str,
|
|
) -> None:
|
|
root, pack_id, artifact_sha256 = _sealed_e10_pack(tmp_path)
|
|
manifest_path = root / "manifest.json"
|
|
manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
|
|
assert callable(mutation)
|
|
mutation(manifest)
|
|
manifest_path.write_text(json.dumps(manifest), encoding="utf-8")
|
|
|
|
with pytest.raises(M48RawEvidenceError, match=message):
|
|
raw_module._validate_e10_pack(
|
|
root,
|
|
expected_pack_id=pack_id,
|
|
expected_artifact_sha256=artifact_sha256,
|
|
)
|
|
|
|
|
|
@dataclass
|
|
class _Store:
|
|
source_time_ns: int = 2_000_000_000
|
|
source_available: bool = True
|
|
|
|
def temporal_binding_for_index(self, frame_index: int) -> RecordedFrameTemporalBinding:
|
|
return RecordedFrameTemporalBinding(
|
|
frame_index=frame_index,
|
|
source_time_ns=self.source_time_ns,
|
|
source_available=self.source_available,
|
|
lidar_camera_delta_ms=1.0 if self.source_available else None,
|
|
pose_point_delta_ms=1.0 if self.source_available else None,
|
|
)
|
|
|
|
def current_points_for_frame(self, frame_index: int) -> np.ndarray:
|
|
del frame_index
|
|
return np.asarray(
|
|
[
|
|
[1.0, 0.0, 0.0],
|
|
[2.0, 0.0, 0.0],
|
|
[3.0, 0.0, 0.0],
|
|
[4.0, 0.0, 0.0],
|
|
[5.0, 0.0, 0.0],
|
|
],
|
|
dtype=np.float64,
|
|
)
|
|
|
|
|
|
@dataclass
|
|
class _BodyFrames:
|
|
available: bool = True
|
|
|
|
def body_frame_for_frame(self, frame_id: str) -> ReplayBodyFrame | None:
|
|
if not self.available:
|
|
return None
|
|
return ReplayBodyFrame(
|
|
frame_id=frame_id,
|
|
origin_map_xyz_m=(1.0, 0.0, 0.0),
|
|
basis_map_from_body=((1.0, 0.0, 0.0), (0.0, 1.0, 0.0), (0.0, 0.0, 1.0)),
|
|
sensor_height_m=1.25,
|
|
surface_slope_deg=0.0,
|
|
forward_source="fixture",
|
|
camera_forward_alignment_deg=0.0,
|
|
)
|
|
|
|
|
|
class _PredictionTrapPack:
|
|
result_id = "m48-object-quality-pack-" + "a" * 64
|
|
result_root = REPOSITORY_ROOT
|
|
manifest = {
|
|
"identity": {
|
|
"source": {
|
|
"source_id": M48_EXPECTED_SOURCE_ID,
|
|
"source_session_id": M48_EXPECTED_SESSION_ID,
|
|
}
|
|
}
|
|
}
|
|
report: dict[str, object] = {}
|
|
clips: tuple[dict[str, object], ...] = ()
|
|
frame_references = (
|
|
{
|
|
"clip_id": "clip-01",
|
|
"sequence": 2,
|
|
"source_time_ns": 2_000_000_000,
|
|
},
|
|
)
|
|
|
|
@property
|
|
def predictions(self) -> object:
|
|
raise AssertionError("neutral raw reader opened frozen predictions")
|
|
|
|
|
|
def _reader(
|
|
tmp_path: Path,
|
|
*,
|
|
store: _Store | None = None,
|
|
body_frames: _BodyFrames | None = None,
|
|
) -> M48RawEvidenceReader:
|
|
profile = load_replay_threat_profile(PROFILE_PATH)
|
|
timeline = SimpleNamespace(
|
|
store=store or _Store(),
|
|
body_frames=body_frames or _BodyFrames(),
|
|
profile=profile,
|
|
)
|
|
threat = SimpleNamespace(
|
|
result_id="m4-threat-replay-fixture",
|
|
result_root=tmp_path,
|
|
manifest={"identity": {"frames_sha256": "f" * 64}},
|
|
)
|
|
return M48RawEvidenceReader(
|
|
repository_root=tmp_path,
|
|
threat_result=threat,
|
|
timeline=timeline,
|
|
point_limit=2,
|
|
)
|
|
|
|
|
|
def test_raw_reader_is_one_based_bounded_body_frame_and_prediction_free(
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
monkeypatch.setattr(raw_module, "_validate_m47_pack_binding", lambda **_: None)
|
|
reader = _reader(tmp_path)
|
|
pack = cast(M48ObjectQualityPack, _PredictionTrapPack())
|
|
|
|
frame = reader(pack, 2)
|
|
|
|
assert set(frame) == {
|
|
"schema_version",
|
|
"pack_id",
|
|
"clip_id",
|
|
"sequence",
|
|
"source_time_ns",
|
|
"source_available",
|
|
"body_frame_available",
|
|
"point_cloud_body_xyz_m",
|
|
"rig",
|
|
"corridor",
|
|
"occupied_voxel_size_m",
|
|
"candidate_identity_included",
|
|
"graph_boxes_ids_scores_included",
|
|
"frozen_predictions_included",
|
|
"strata_included",
|
|
"authority",
|
|
}
|
|
assert frame["schema_version"] == M48_RAW_SPATIAL_FRAME_SCHEMA
|
|
assert frame["clip_id"] == "clip-01"
|
|
assert frame["sequence"] == 2
|
|
assert frame["source_available"] is True
|
|
assert frame["body_frame_available"] is True
|
|
assert frame["point_cloud_body_xyz_m"] == [[0.0, 0.0, 0.0], [3.0, 0.0, 0.0]]
|
|
assert len(cast(list[object], frame["point_cloud_body_xyz_m"])) <= 2
|
|
assert frame["candidate_identity_included"] is False
|
|
assert frame["graph_boxes_ids_scores_included"] is False
|
|
assert frame["frozen_predictions_included"] is False
|
|
assert frame["strata_included"] is False
|
|
assert "metric_obstacles" not in frame
|
|
assert "camera_proposals" not in frame
|
|
assert "decision_counts" not in frame
|
|
assert "body_frame" not in frame
|
|
|
|
|
|
def test_raw_reader_fails_closed_on_clip_or_source_time_escape(
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
monkeypatch.setattr(raw_module, "_validate_m47_pack_binding", lambda **_: None)
|
|
pack = cast(M48ObjectQualityPack, _PredictionTrapPack())
|
|
reader = _reader(tmp_path)
|
|
|
|
with pytest.raises(M48RawEvidenceError, match="outside the selected neutral clips"):
|
|
reader(pack, 1)
|
|
|
|
mismatched = _reader(tmp_path, store=_Store(source_time_ns=2_000_000_001))
|
|
with pytest.raises(M48RawEvidenceError, match="source time escaped"):
|
|
mismatched(pack, 2)
|
|
|
|
|
|
def test_raw_reader_emits_empty_cloud_when_body_frame_is_unavailable(
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
monkeypatch.setattr(raw_module, "_validate_m47_pack_binding", lambda **_: None)
|
|
reader = _reader(
|
|
tmp_path,
|
|
store=_Store(source_available=False),
|
|
body_frames=_BodyFrames(available=False),
|
|
)
|
|
pack = cast(M48ObjectQualityPack, _PredictionTrapPack())
|
|
|
|
frame = reader.frame(pack=pack, sequence=2)
|
|
|
|
assert frame["source_available"] is False
|
|
assert frame["body_frame_available"] is False
|
|
assert frame["point_cloud_body_xyz_m"] == []
|
|
assert isinstance(frame["rig"], dict)
|
|
assert isinstance(frame["corridor"], dict)
|
|
|
|
|
|
@pytest.mark.parametrize("point_limit", [0, 4097, True])
|
|
def test_raw_reader_rejects_unbounded_point_limits(
|
|
tmp_path: Path,
|
|
point_limit: int,
|
|
) -> None:
|
|
profile = load_replay_threat_profile(PROFILE_PATH)
|
|
timeline = SimpleNamespace(store=_Store(), body_frames=_BodyFrames(), profile=profile)
|
|
threat = SimpleNamespace(result_id="fixture", result_root=tmp_path, manifest={})
|
|
|
|
with pytest.raises(M48RawEvidenceError, match="point limit"):
|
|
M48RawEvidenceReader(
|
|
repository_root=tmp_path,
|
|
threat_result=threat,
|
|
timeline=timeline,
|
|
point_limit=point_limit,
|
|
)
|