feat(perception): gate sealed truth evaluation
This commit is contained in:
DCCONSTRUCTIONS
@@ -0,0 +1,254 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import json
|
||||
from pathlib import Path
|
||||
from types import SimpleNamespace
|
||||
from typing import Any
|
||||
|
||||
import pytest
|
||||
|
||||
import k1link.compute.e48_detector_truth_seal as e48
|
||||
|
||||
|
||||
def _canonical(value: object) -> bytes:
|
||||
return json.dumps(
|
||||
value,
|
||||
ensure_ascii=False,
|
||||
sort_keys=True,
|
||||
separators=(",", ":"),
|
||||
).encode()
|
||||
|
||||
|
||||
def _write_json(path: Path, value: object) -> None:
|
||||
path.write_text(json.dumps(value), encoding="utf-8")
|
||||
|
||||
|
||||
def _fixture(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> dict[str, Any]:
|
||||
truth_root = tmp_path / "e46-detector-truth-island-test"
|
||||
truth_root.mkdir()
|
||||
references = [
|
||||
{
|
||||
"truth_island_sequence": 1,
|
||||
"image_id": 10,
|
||||
"frame_index": 100,
|
||||
"session_seconds": 10.0,
|
||||
"role": "anchor",
|
||||
"group_id": "anchor-1",
|
||||
"source_path": "images/one.png",
|
||||
"sha256": "a" * 64,
|
||||
},
|
||||
{
|
||||
"truth_island_sequence": 2,
|
||||
"image_id": 11,
|
||||
"frame_index": 101,
|
||||
"session_seconds": 10.1,
|
||||
"role": "temporal",
|
||||
"group_id": "clip-1",
|
||||
"source_path": "images/two.png",
|
||||
"sha256": "b" * 64,
|
||||
},
|
||||
]
|
||||
(truth_root / "image-references.jsonl").write_text(
|
||||
"".join(json.dumps(row) + "\n" for row in references),
|
||||
encoding="utf-8",
|
||||
)
|
||||
_write_json(
|
||||
truth_root / "blind-contract.json",
|
||||
{"annotation": {"classes": ["person", "car"]}},
|
||||
)
|
||||
_write_json(truth_root / "manifest.json", {"result_id": truth_root.name})
|
||||
monkeypatch.setattr(
|
||||
e48,
|
||||
"read_e46_detector_truth_island",
|
||||
lambda _: SimpleNamespace(
|
||||
result_id=truth_root.name,
|
||||
result_root=truth_root,
|
||||
report={
|
||||
"status": "prepared-awaiting-independent-human-review",
|
||||
"blindness": {"truth_labels_available": False},
|
||||
},
|
||||
),
|
||||
)
|
||||
|
||||
freeze_identity = {
|
||||
"schema_version": "missioncore.e47-detector-candidate-freeze/v1",
|
||||
"truth_island": {
|
||||
"result_id": truth_root.name,
|
||||
"state": "prepared-unreviewed-no-prelabels",
|
||||
"truth_labels_available": False,
|
||||
},
|
||||
"candidates": [],
|
||||
"prediction_rows_sha256": "c" * 64,
|
||||
"producer_sha256": "d" * 64,
|
||||
"authority": {
|
||||
"commands_enabled": False,
|
||||
"navigation_or_safety_accepted": False,
|
||||
},
|
||||
}
|
||||
freeze_sha = hashlib.sha256(_canonical(freeze_identity)).hexdigest()
|
||||
freeze_root = tmp_path / f"e47-detector-candidate-freeze-{freeze_sha}"
|
||||
freeze_root.mkdir()
|
||||
_write_json(
|
||||
freeze_root / "manifest.json",
|
||||
{
|
||||
"schema_version": "missioncore.e47-detector-candidate-freeze/v1",
|
||||
"result_id": freeze_root.name,
|
||||
"identity_sha256": freeze_sha,
|
||||
"identity": freeze_identity,
|
||||
"created_at_utc": "2026-07-29T10:00:00Z",
|
||||
"acceptance_state": "accepted-prediction-freeze-only",
|
||||
"authority": {
|
||||
"commands_enabled": False,
|
||||
"navigation_or_safety_accepted": False,
|
||||
},
|
||||
},
|
||||
)
|
||||
|
||||
def images(*, car_box: list[float]) -> list[dict[str, Any]]:
|
||||
return [
|
||||
{
|
||||
**{
|
||||
key: reference[key]
|
||||
for key in (
|
||||
"truth_island_sequence",
|
||||
"image_id",
|
||||
"frame_index",
|
||||
"session_seconds",
|
||||
"role",
|
||||
"group_id",
|
||||
"source_path",
|
||||
)
|
||||
},
|
||||
"source_sha256": reference["sha256"],
|
||||
"review_state": "reviewed",
|
||||
"hard_negative": index == 1,
|
||||
"objects": (
|
||||
[
|
||||
{
|
||||
"object_id": "car-1",
|
||||
"category": "car",
|
||||
"box_xyxy": car_box,
|
||||
"occluded": False,
|
||||
"truncated": False,
|
||||
"notes": None,
|
||||
}
|
||||
]
|
||||
if index == 0
|
||||
else []
|
||||
),
|
||||
"notes": None,
|
||||
}
|
||||
for index, reference in enumerate(references)
|
||||
]
|
||||
|
||||
def review(reviewer_id: str, car_box: list[float]) -> dict[str, Any]:
|
||||
return {
|
||||
"schema_version": e48.E48_REVIEW_SCHEMA,
|
||||
"truth_island_id": truth_root.name,
|
||||
"state": "completed-independent-no-model-assistance",
|
||||
"reviewer_id": reviewer_id,
|
||||
"review_round": 1,
|
||||
"blindness": {
|
||||
"candidate_identity_seen": False,
|
||||
"model_prelabels_seen": False,
|
||||
"model_predictions_seen": False,
|
||||
"model_scores_seen": False,
|
||||
},
|
||||
"images": images(car_box=car_box),
|
||||
"acceptance": {
|
||||
"all_images_reviewed": True,
|
||||
"independent": True,
|
||||
"submitted_at_utc": "2026-07-29T11:00:00Z",
|
||||
},
|
||||
}
|
||||
|
||||
review_a = review("reviewer-a", [10.0, 20.0, 100.0, 200.0])
|
||||
review_b = review("reviewer-b", [12.0, 20.0, 102.0, 200.0])
|
||||
reviewer_a_path = tmp_path / "review-a.json"
|
||||
reviewer_b_path = tmp_path / "review-b.json"
|
||||
_write_json(reviewer_a_path, review_a)
|
||||
_write_json(reviewer_b_path, review_b)
|
||||
adjudicated_images = images(car_box=[11.0, 20.0, 101.0, 200.0])
|
||||
for image in adjudicated_images:
|
||||
image["review_state"] = "adjudicated"
|
||||
adjudication = {
|
||||
"schema_version": e48.E48_ADJUDICATION_SCHEMA,
|
||||
"truth_island_id": truth_root.name,
|
||||
"state": "completed-adjudicated",
|
||||
"adjudicator_id": "adjudicator-1",
|
||||
"review_submission_sha256": sorted(
|
||||
(
|
||||
hashlib.sha256(_canonical(review_a)).hexdigest(),
|
||||
hashlib.sha256(_canonical(review_b)).hexdigest(),
|
||||
)
|
||||
),
|
||||
"images": adjudicated_images,
|
||||
"acceptance": {
|
||||
"all_images_adjudicated": True,
|
||||
"all_disagreements_resolved": True,
|
||||
"sealed_at_utc": "2026-07-29T12:00:00Z",
|
||||
},
|
||||
}
|
||||
adjudication_path = tmp_path / "adjudication.json"
|
||||
_write_json(adjudication_path, adjudication)
|
||||
return {
|
||||
"truth_island_root": truth_root,
|
||||
"prediction_freeze_root": freeze_root,
|
||||
"reviewer_a_path": reviewer_a_path,
|
||||
"reviewer_b_path": reviewer_b_path,
|
||||
"adjudication_path": adjudication_path,
|
||||
"output_root": tmp_path / "results",
|
||||
}
|
||||
|
||||
|
||||
def test_e48_seals_two_blind_reviews_after_adjudication(
|
||||
tmp_path: Path,
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
inputs = _fixture(tmp_path, monkeypatch)
|
||||
result = e48.build_e48_detector_truth_seal(**inputs)
|
||||
|
||||
assert result["report"]["status"] == "sealed-adjudicated-independent-truth"
|
||||
assert result["report"]["frame_count"] == 2
|
||||
assert result["report"]["object_count"] == 1
|
||||
assert result["provenance"]["prediction_content_read_by_sealer"] is False
|
||||
assert result["truth_rows"][0]["objects"][0]["box_xyxy"] == [
|
||||
11.0,
|
||||
20.0,
|
||||
101.0,
|
||||
200.0,
|
||||
]
|
||||
|
||||
|
||||
def test_e48_rejects_same_reviewer_identity(
|
||||
tmp_path: Path,
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
inputs = _fixture(tmp_path, monkeypatch)
|
||||
review_b = json.loads(inputs["reviewer_b_path"].read_text(encoding="utf-8"))
|
||||
review_b["reviewer_id"] = "reviewer-a"
|
||||
_write_json(inputs["reviewer_b_path"], review_b)
|
||||
|
||||
with pytest.raises(e48.E48DetectorTruthSealError, match="must differ"):
|
||||
e48.build_e48_detector_truth_seal(**inputs)
|
||||
|
||||
|
||||
def test_e48_rejects_hidden_model_score_field(
|
||||
tmp_path: Path,
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
inputs = _fixture(tmp_path, monkeypatch)
|
||||
review_a = json.loads(inputs["reviewer_a_path"].read_text(encoding="utf-8"))
|
||||
review_a["images"][0]["objects"][0]["score"] = 0.99
|
||||
_write_json(inputs["reviewer_a_path"], review_a)
|
||||
|
||||
with pytest.raises(e48.E48DetectorTruthSealError, match="fields"):
|
||||
e48.build_e48_detector_truth_seal(**inputs)
|
||||
|
||||
|
||||
def test_e48_box_iou_is_exact_for_simple_overlap() -> None:
|
||||
assert e48.box_iou(
|
||||
[0.0, 0.0, 10.0, 10.0],
|
||||
[5.0, 0.0, 15.0, 10.0],
|
||||
) == pytest.approx(1.0 / 3.0)
|
||||
@@ -0,0 +1,151 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any
|
||||
|
||||
import pytest
|
||||
from PIL import Image
|
||||
|
||||
from k1link.compute.e49_detector_truth_evaluation import (
|
||||
E49DetectorTruthEvaluationError,
|
||||
evaluate_frozen_detector_candidates,
|
||||
)
|
||||
|
||||
|
||||
def _truth_row(
|
||||
sequence: int,
|
||||
*,
|
||||
category: str,
|
||||
box: list[float],
|
||||
) -> dict[str, Any]:
|
||||
return {
|
||||
"truth_island_sequence": sequence,
|
||||
"image_id": sequence,
|
||||
"frame_index": 100 + sequence,
|
||||
"session_seconds": float(sequence),
|
||||
"role": "temporal",
|
||||
"group_id": "clip-1",
|
||||
"source_path": f"image-{sequence}.png",
|
||||
"source_image_sha256": f"{sequence:064x}",
|
||||
"hard_negative": False,
|
||||
"objects": [
|
||||
{
|
||||
"object_id": f"object-{sequence}",
|
||||
"category": category,
|
||||
"box_xyxy": box,
|
||||
"occluded": False,
|
||||
"truncated": False,
|
||||
"notes": None,
|
||||
}
|
||||
],
|
||||
"adjudicated": True,
|
||||
}
|
||||
|
||||
|
||||
def _prediction_row(
|
||||
candidate_id: str,
|
||||
truth: dict[str, Any],
|
||||
predictions: list[dict[str, Any]],
|
||||
) -> dict[str, Any]:
|
||||
return {
|
||||
"candidate_id": candidate_id,
|
||||
"truth_island_sequence": truth["truth_island_sequence"],
|
||||
"image_id": truth["image_id"],
|
||||
"frame_index": truth["frame_index"],
|
||||
"session_seconds": truth["session_seconds"],
|
||||
"source_image_sha256": truth["source_image_sha256"],
|
||||
"predictions": predictions,
|
||||
"truth_joined": False,
|
||||
}
|
||||
|
||||
|
||||
def _prediction(
|
||||
*,
|
||||
category: str,
|
||||
box: list[float],
|
||||
score: float = 0.9,
|
||||
) -> dict[str, Any]:
|
||||
return {
|
||||
"category": category,
|
||||
"score": score,
|
||||
"box_xyxy": box,
|
||||
}
|
||||
|
||||
|
||||
def test_e49_perfect_candidate_reaches_one_and_poor_candidate_does_not() -> None:
|
||||
truth = (
|
||||
_truth_row(1, category="car", box=[10.0, 10.0, 110.0, 110.0]),
|
||||
_truth_row(2, category="car", box=[20.0, 20.0, 120.0, 120.0]),
|
||||
)
|
||||
predictions = (
|
||||
_prediction_row(
|
||||
"perfect",
|
||||
truth[0],
|
||||
[_prediction(category="car", box=[10.0, 10.0, 110.0, 110.0])],
|
||||
),
|
||||
_prediction_row(
|
||||
"perfect",
|
||||
truth[1],
|
||||
[_prediction(category="car", box=[20.0, 20.0, 120.0, 120.0])],
|
||||
),
|
||||
_prediction_row(
|
||||
"poor",
|
||||
truth[0],
|
||||
[_prediction(category="car", box=[300.0, 300.0, 400.0, 400.0])],
|
||||
),
|
||||
_prediction_row("poor", truth[1], []),
|
||||
)
|
||||
|
||||
metrics = evaluate_frozen_detector_candidates(
|
||||
truth_rows=truth,
|
||||
prediction_rows=predictions,
|
||||
valid_fov_mask=Image.new("L", (800, 600), color=255),
|
||||
)
|
||||
|
||||
assert metrics["perfect"]["coco_ap_50_95"] == 1.0
|
||||
assert metrics["perfect"]["ap50"] == 1.0
|
||||
assert metrics["perfect"]["ar100"] == 1.0
|
||||
assert metrics["perfect"]["person_vehicle_miss_rate"] == 0.0
|
||||
assert metrics["perfect"]["candidate_winner_selected"] is False
|
||||
assert metrics["poor"]["coco_ap_50_95"] == 0.0
|
||||
assert metrics["poor"]["person_vehicle_miss_rate"] == 1.0
|
||||
|
||||
|
||||
def test_e49_reports_valid_fov_centre_leakage() -> None:
|
||||
truth = (
|
||||
_truth_row(1, category="car", box=[10.0, 10.0, 110.0, 110.0]),
|
||||
)
|
||||
prediction = _prediction_row(
|
||||
"candidate",
|
||||
truth[0],
|
||||
[_prediction(category="car", box=[10.0, 10.0, 110.0, 110.0])],
|
||||
)
|
||||
|
||||
metrics = evaluate_frozen_detector_candidates(
|
||||
truth_rows=truth,
|
||||
prediction_rows=(prediction,),
|
||||
valid_fov_mask=Image.new("L", (800, 600), color=0),
|
||||
)
|
||||
|
||||
assert metrics["candidate"]["valid_fov_boundary_leakage"] == 1.0
|
||||
|
||||
|
||||
def test_e49_rejects_prediction_identity_or_coverage_drift() -> None:
|
||||
truth = (
|
||||
_truth_row(1, category="person", box=[10.0, 10.0, 20.0, 30.0]),
|
||||
_truth_row(2, category="person", box=[12.0, 10.0, 22.0, 30.0]),
|
||||
)
|
||||
prediction = _prediction_row(
|
||||
"candidate",
|
||||
truth[0],
|
||||
[_prediction(category="person", box=[10.0, 10.0, 20.0, 30.0])],
|
||||
)
|
||||
|
||||
with pytest.raises(
|
||||
E49DetectorTruthEvaluationError,
|
||||
match="coverage",
|
||||
):
|
||||
evaluate_frozen_detector_candidates(
|
||||
truth_rows=truth,
|
||||
prediction_rows=(prediction,),
|
||||
valid_fov_mask=Image.new("L", (800, 600), color=255),
|
||||
)
|
||||
Reference in New Issue
Block a user