229 lines
7.3 KiB
Python
229 lines
7.3 KiB
Python
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import pytest
|
|
|
|
from k1link.compute.e30_review_pack import (
|
|
E30_REASON_TAXONOMY,
|
|
E30ReviewPackError,
|
|
E30ReviewSelectionProfile,
|
|
build_e30_review_pack,
|
|
)
|
|
from k1link.compute.semantic_geometry_fusion import (
|
|
CAMERA_GEOMETRY_FRAME_SCHEMA,
|
|
CAMERA_GEOMETRY_FUSION_SCHEMA,
|
|
CAMERA_GEOMETRY_REPORT_SCHEMA,
|
|
)
|
|
|
|
|
|
def _canonical(value: object) -> bytes:
|
|
return json.dumps(
|
|
value,
|
|
sort_keys=True,
|
|
separators=(",", ":"),
|
|
allow_nan=False,
|
|
).encode()
|
|
|
|
|
|
def _sha256(path: Path) -> str:
|
|
return hashlib.sha256(path.read_bytes()).hexdigest()
|
|
|
|
|
|
def _observation(status: str, index: int) -> dict[str, object]:
|
|
return {
|
|
"track_id": index,
|
|
"label": "car" if index % 2 else "person",
|
|
"association_group": "vehicle" if index % 2 else "person",
|
|
"geometry_status": status,
|
|
"geometry_reason": f"reason-{status}",
|
|
"range_m": None if status in {"single-source-camera", "unknown"} else 2.5 + index,
|
|
"support": {"connected_occupied_points": 0},
|
|
}
|
|
|
|
|
|
def _source_result(root: Path) -> Path:
|
|
result_id = "e29-camera-geometry-" + "a" * 64
|
|
result = root / result_id
|
|
result.mkdir()
|
|
frames = [
|
|
{
|
|
"schema_version": CAMERA_GEOMETRY_FRAME_SCHEMA,
|
|
"frame_index": 0,
|
|
"source_frame_index": 10,
|
|
"session_seconds": 1.0,
|
|
"semantic_observations": [
|
|
_observation("agree", 1),
|
|
_observation("single-source-camera", 2),
|
|
_observation("conflict", 3),
|
|
_observation("unknown", 4),
|
|
],
|
|
"geometry_only_occupied": [
|
|
{
|
|
"geometry_status": "single-source-geometry",
|
|
"nearest_range_m": 8.0,
|
|
"point_count": 10,
|
|
}
|
|
],
|
|
},
|
|
{
|
|
"schema_version": CAMERA_GEOMETRY_FRAME_SCHEMA,
|
|
"frame_index": 1,
|
|
"source_frame_index": 11,
|
|
"session_seconds": 2.0,
|
|
"semantic_observations": [
|
|
_observation("agree", 5),
|
|
_observation("single-source-camera", 6),
|
|
_observation("conflict", 7),
|
|
_observation("unknown", 8),
|
|
],
|
|
"geometry_only_occupied": [
|
|
{
|
|
"geometry_status": "single-source-geometry",
|
|
"nearest_range_m": 2.0,
|
|
"point_count": 12,
|
|
}
|
|
],
|
|
},
|
|
]
|
|
frames_path = result / "camera-geometry-frames.jsonl"
|
|
frames_path.write_bytes(b"".join(_canonical(frame) + b"\n" for frame in frames))
|
|
|
|
identity: dict[str, Any] = {
|
|
"frame_count": 2,
|
|
"timeline_start_seconds": 1.0,
|
|
"timeline_end_seconds": 2.0,
|
|
"source_result_id": "e10-integrated-perception-" + "b" * 64,
|
|
"source_pack_id": "e10-lidar-pack-" + "c" * 64,
|
|
"local_surface_model_id": "k1-local-surface-" + "d" * 64,
|
|
"profile": {"profile_id": "camera-first-local-surface-validation/v1"},
|
|
"authority": {
|
|
"commands_enabled": False,
|
|
"navigation_or_safety_accepted": False,
|
|
},
|
|
}
|
|
report = {
|
|
"schema_version": CAMERA_GEOMETRY_REPORT_SCHEMA,
|
|
"result_id": result_id,
|
|
"status": "diagnostic-replay-complete",
|
|
"ground_truth": False,
|
|
"identity": identity,
|
|
"metrics": {
|
|
"semantic_observations": {
|
|
"geometry_status": {
|
|
"agree": 2,
|
|
"single-source-camera": 2,
|
|
"conflict": 2,
|
|
"unknown": 2,
|
|
}
|
|
},
|
|
"geometry_only_occupied": {"cluster_count": 2},
|
|
},
|
|
"authority": {
|
|
"commands_enabled": False,
|
|
"navigation_or_safety_accepted": False,
|
|
},
|
|
}
|
|
report_path = result / "camera-geometry-report.json"
|
|
report_path.write_bytes(_canonical(report))
|
|
identity_sha256 = hashlib.sha256(_canonical(identity)).hexdigest()
|
|
result_id = f"e29-camera-geometry-{identity_sha256}"
|
|
result_with_identity = root / result_id
|
|
result.rename(result_with_identity)
|
|
result = result_with_identity
|
|
frames_path = result / frames_path.name
|
|
report_path = result / report_path.name
|
|
report["result_id"] = result_id
|
|
report_path.write_bytes(_canonical(report))
|
|
manifest = {
|
|
"schema_version": CAMERA_GEOMETRY_FUSION_SCHEMA,
|
|
"result_id": result_id,
|
|
"identity_sha256": identity_sha256,
|
|
"identity": identity,
|
|
"ground_truth": False,
|
|
"artifacts": [
|
|
{
|
|
"role": "camera-geometry-frames",
|
|
"path": frames_path.name,
|
|
"byte_length": frames_path.stat().st_size,
|
|
"sha256": _sha256(frames_path),
|
|
},
|
|
{
|
|
"role": "camera-geometry-report",
|
|
"path": report_path.name,
|
|
"byte_length": report_path.stat().st_size,
|
|
"sha256": _sha256(report_path),
|
|
},
|
|
],
|
|
}
|
|
(result / "manifest.json").write_bytes(_canonical(manifest))
|
|
return result
|
|
|
|
|
|
def test_review_pack_binds_all_strata_and_keeps_human_decision_open(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
source = _source_result(tmp_path)
|
|
profile = E30ReviewSelectionProfile(
|
|
agree_maximum=1,
|
|
camera_only_maximum=1,
|
|
unknown_maximum=1,
|
|
geometry_only_maximum=1,
|
|
temporal_bins=2,
|
|
)
|
|
|
|
first = build_e30_review_pack(
|
|
e29_result_root=source,
|
|
output_root=tmp_path / "review-packs",
|
|
profile=profile,
|
|
)
|
|
second = build_e30_review_pack(
|
|
e29_result_root=source,
|
|
output_root=tmp_path / "review-packs",
|
|
profile=profile,
|
|
)
|
|
|
|
assert first.result_id == second.result_id
|
|
assert first.manifest["human_review_complete"] is False
|
|
assert first.manifest["lab_published"] is False
|
|
assert first.manifest["source_counts"] == {
|
|
"agree": 2,
|
|
"camera-only": 2,
|
|
"conflict": 2,
|
|
"geometry-only": 2,
|
|
"unknown": 2,
|
|
}
|
|
assert first.manifest["selected_counts"] == {
|
|
"agree": 1,
|
|
"camera-only": 1,
|
|
"conflict": 2,
|
|
"geometry-only": 1,
|
|
"unknown": 1,
|
|
}
|
|
lines = (first.result_root / "review-items.jsonl").read_text().splitlines()
|
|
items = [json.loads(line) for line in lines]
|
|
assert len(items) == 6
|
|
assert all(item["review"]["state"] == "unreviewed" for item in items)
|
|
assert all(item["materialization"]["source_reprojection_required"] for item in items)
|
|
assert len({item["item_id"] for item in items}) == 6
|
|
|
|
|
|
def test_review_pack_taxonomy_is_fixed_and_source_tampering_rejects(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
assert E30_REASON_TAXONOMY[0] == "no_lidar_observation"
|
|
assert E30_REASON_TAXONOMY[-1] == "unknown"
|
|
assert len(E30_REASON_TAXONOMY) == 19
|
|
source = _source_result(tmp_path)
|
|
frames = source / "camera-geometry-frames.jsonl"
|
|
frames.write_bytes(frames.read_bytes() + b"\n")
|
|
|
|
with pytest.raises(E30ReviewPackError, match="byte length changed"):
|
|
build_e30_review_pack(
|
|
e29_result_root=source,
|
|
output_root=tmp_path / "review-packs",
|
|
)
|