diff --git a/docs/01_IMPLEMENTATION_PLAN.md b/docs/01_IMPLEMENTATION_PLAN.md index 31f936f..901efed 100644 --- a/docs/01_IMPLEMENTATION_PLAN.md +++ b/docs/01_IMPLEMENTATION_PLAN.md @@ -17,6 +17,7 @@ Each gate produces evidence and an explicit GO, PAUSE or BLOCKED result. | E31 binding sensitivity | MEASURED — E45 closes accounting for 87/87 accepted correspondences and finds no material monotonic residual association with represented image radius, rig speed or pose age. It does not supply calibration-target truth or outer-fisheye coverage. | | Detector Truth Island | PREPARED — E46 freezes 32 references with no prelabels, predictions, scores or candidate identity in the review package. Two independent reviews and adjudication are still required. | | Detector candidate comparison | FROZEN BEFORE TRUTH — E47 freezes raw-KB4 and fixed-valid-FOV-fill predictions from the same exact Mask R-CNN checkpoint. No accuracy result or winner exists before the E46 truth seal. | +| Detector truth/evaluation executors | READY, NOT RUN — E48 fail-closed review/adjudication sealing and separate post-seal E49 scoring are implemented. Neither result exists because real independent reviews are absent. | | Product interface | DEFERRED — no new windows, page anatomy or design changes are part of this stabilization increment. | The governing decision is diff --git a/docs/20_RAVNOVES00_REFERENCE_SOURCE_PRODUCT_PLAN.md b/docs/20_RAVNOVES00_REFERENCE_SOURCE_PRODUCT_PLAN.md index 4ac4953..5d71afd 100644 --- a/docs/20_RAVNOVES00_REFERENCE_SOURCE_PRODUCT_PLAN.md +++ b/docs/20_RAVNOVES00_REFERENCE_SOURCE_PRODUCT_PLAN.md @@ -170,6 +170,14 @@ these counts are descriptive rather than accuracy evidence. No truth was joined, no winner was selected and no retraining is authorized before the E46 truth seal. +E48 and E49 executors are prepared but have not produced results. E48 validates +two distinct blind-review identities, exact 32-frame coverage, model-free +annotation fields, explicit adjudication and review hashes while reading only +the E47 manifest identity. E49 is physically downstream of an accepted E48 +seal and computes the frozen AP/AR, miss, large-box, valid-FOV-centre leakage +and temporal class-count metrics. It cannot select a candidate automatically. +No E48/E49 result exists until two real reviews and adjudication are supplied. + E43 immutable protocol `e43-future-capture-protocol-28f091b9648daffce988d44c183e21f56d77988061630de934f8003fb13701d8` preregisters the later same-K1/new-route transfer. It requires the same mount, diff --git a/experiments/perception/E48_E49_GATE_IMPLEMENTATION_2026-07-29.md b/experiments/perception/E48_E49_GATE_IMPLEMENTATION_2026-07-29.md new file mode 100644 index 0000000..159367b --- /dev/null +++ b/experiments/perception/E48_E49_GATE_IMPLEMENTATION_2026-07-29.md @@ -0,0 +1,128 @@ +# E48–E49 detector truth gate implementation + +Date: 2026-07-29 + +Status: executable gate ready; no E48 or E49 result exists + +## Purpose + +E46 prepared a references-only detector Truth Island and E47 froze two +prediction candidates before truth reveal. E48–E49 make the remaining boundary +executable before either reviewer submission exists: + +1. E48 accepts exactly two completed independent blind reviews and one explicit + adjudication. +2. E48 seals only adjudicated truth and never reads E47 prediction content. +3. E49 refuses to run without an accepted E48 seal. +4. E49 then joins the exact E47 prediction generation and computes the frozen + detector metrics without selecting a winner automatically. + +The implementation does not fabricate reviewer submissions, adjudication, +truth, metrics or a candidate decision. + +## E48 review and seal contract + +Each reviewer submission must use +`missioncore.e48-independent-detector-review/v1` and preserve all 32 E46 image +identities in their original order. Reviewer IDs must be different opaque +identifiers. Every submission explicitly attests that candidate identity, +prelabels, predictions and scores were not seen. + +Every image must end in `reviewed` with: + +- a required hard-negative decision; +- every task-relevant object represented by `object_id`, target class and + `xyxy` box in the original 800×600 coordinate system; +- required `occluded` and `truncated` flags; +- no confidence score, prediction reference or unknown field. + +The adjudication document uses +`missioncore.e48-detector-review-adjudication/v1`, binds the canonical SHA-256 +of both reviewer submissions, covers every image, and explicitly accepts that +all disagreements are resolved. Its sealed time may not predate the immutable +E47 prediction freeze. + +The E48 sealer rejects: + +- duplicate reviewer identities; +- missing, reordered or changed source-image identities; +- incomplete frame coverage; +- hidden model/prediction fields; +- unknown classes, duplicate object IDs or invalid boxes; +- a hard-negative flag that conflicts with object presence; +- incomplete review/adjudication acceptance; +- review hashes that do not match the adjudication; +- an E47 manifest bound to another Truth Island. + +E48 reads the E47 manifest identity only. It does not open +`candidate-predictions.jsonl`. A successful generation records this boundary +as `prediction_content_read_by_sealer=false`. + +Human independence and blindness remain explicit signed-process attestations +bound to distinct opaque identities; the code cannot prove a person's identity +or what they saw outside the controlled package. + +## E49 frozen evaluation + +E49 accepts only: + +- the exact E46 Truth Island; +- an accepted `sealed-adjudicated-independent-truth` E48 generation; +- the exact E47 prediction freeze referenced by E48; +- a content-addressed valid-FOV mask with the same calibration SHA, camera slot + and 800×600 resolution. + +It computes: + +- COCO-style 101-point interpolated AP averaged over IoU 0.50:0.05:0.95; +- AP50 and AP75; +- AR100; +- per-class AP and recall at IoU 0.50; +- combined person/vehicle miss rate at IoU 0.50; +- false-large-box rate for boxes occupying at least 25% of the image; +- valid-FOV leakage as prediction-box centres outside the calibrated mask; +- temporal class-count flicker over adjacent frames in each frozen E46 clip. + +The valid-FOV metric is deliberately named and bounded: it is box-centre +admission, not full box-area or mask leakage. Temporal flicker is class-count +stability because the frozen candidates do not publish track identities. + +E49 always leaves `candidate_winner_selected=false` and +`model_retraining_authorized=false`. Candidate acceptance remains a separate +explicit product decision after reviewing all metrics and limitations. + +## Execution + +After two real reviewer files and an adjudication exist: + +```text +python experiments/perception/run_e48_detector_truth_seal.py \ + --truth-island-root \ + --prediction-freeze-root \ + --reviewer-a \ + --reviewer-b \ + --adjudication \ + --output-root .runtime/compute-experiments/e48/results +``` + +Only after E48 succeeds: + +```text +python experiments/perception/run_e49_detector_truth_evaluation.py \ + --truth-island-root \ + --truth-seal-root \ + --prediction-freeze-root \ + --valid-fov-root \ + --output-root .runtime/compute-experiments/e49/results +``` + +No real command is run now because the two reviewer submissions and +adjudication do not exist. + +## Resource and authority boundary + +The tooling is pure local validation and metric code. It starts no Docker +container, model, Worker 006 job, video decoder, browser session, network +mutation or second Mission Core service. + +Navigation, safety and command authority remain false. diff --git a/experiments/perception/run_e48_detector_truth_seal.py b/experiments/perception/run_e48_detector_truth_seal.py new file mode 100644 index 0000000..a30b502 --- /dev/null +++ b/experiments/perception/run_e48_detector_truth_seal.py @@ -0,0 +1,49 @@ +#!/usr/bin/env python3 +"""Seal completed E46 independent reviews into immutable E48 truth.""" + +from __future__ import annotations + +import argparse +import json +from pathlib import Path + +from k1link.compute.e48_detector_truth_seal import ( + build_e48_detector_truth_seal, +) + + +def main() -> int: + parser = argparse.ArgumentParser() + parser.add_argument("--truth-island-root", type=Path, required=True) + parser.add_argument("--prediction-freeze-root", type=Path, required=True) + parser.add_argument("--reviewer-a", type=Path, required=True) + parser.add_argument("--reviewer-b", type=Path, required=True) + parser.add_argument("--adjudication", type=Path, required=True) + parser.add_argument("--output-root", type=Path, required=True) + args = parser.parse_args() + result = build_e48_detector_truth_seal( + truth_island_root=args.truth_island_root, + prediction_freeze_root=args.prediction_freeze_root, + reviewer_a_path=args.reviewer_a, + reviewer_b_path=args.reviewer_b, + adjudication_path=args.adjudication, + output_root=args.output_root, + ) + print( + json.dumps( + { + "result_id": result["result_id"], + "result_root": str(result["result_root"]), + "status": result["report"]["status"], + "frame_count": result["report"]["frame_count"], + "object_count": result["report"]["object_count"], + }, + ensure_ascii=False, + sort_keys=True, + ) + ) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/experiments/perception/run_e49_detector_truth_evaluation.py b/experiments/perception/run_e49_detector_truth_evaluation.py new file mode 100644 index 0000000..a847fb4 --- /dev/null +++ b/experiments/perception/run_e49_detector_truth_evaluation.py @@ -0,0 +1,46 @@ +#!/usr/bin/env python3 +"""Evaluate exact E47 predictions after the E48 truth seal.""" + +from __future__ import annotations + +import argparse +import json +from pathlib import Path + +from k1link.compute.e49_detector_truth_evaluation import ( + build_e49_detector_truth_evaluation, +) + + +def main() -> int: + parser = argparse.ArgumentParser() + parser.add_argument("--truth-island-root", type=Path, required=True) + parser.add_argument("--truth-seal-root", type=Path, required=True) + parser.add_argument("--prediction-freeze-root", type=Path, required=True) + parser.add_argument("--valid-fov-root", type=Path, required=True) + parser.add_argument("--output-root", type=Path, required=True) + args = parser.parse_args() + result = build_e49_detector_truth_evaluation( + truth_island_root=args.truth_island_root, + truth_seal_root=args.truth_seal_root, + prediction_freeze_root=args.prediction_freeze_root, + valid_fov_root=args.valid_fov_root, + output_root=args.output_root, + ) + print( + json.dumps( + { + "result_id": result["result_id"], + "result_root": str(result["result_root"]), + "status": result["report"]["status"], + "candidates": result["report"]["candidates"], + }, + ensure_ascii=False, + sort_keys=True, + ) + ) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/src/k1link/compute/e48_detector_truth_seal.py b/src/k1link/compute/e48_detector_truth_seal.py new file mode 100644 index 0000000..1318aee --- /dev/null +++ b/src/k1link/compute/e48_detector_truth_seal.py @@ -0,0 +1,840 @@ +"""Seal two independent E46 reviews without reading E47 predictions.""" + +from __future__ import annotations + +import hashlib +import json +import math +import os +import re +import shutil +import uuid +from collections import Counter +from collections.abc import Iterable +from datetime import UTC, datetime +from pathlib import Path +from typing import Any, Final + +from .e46_detector_truth_island import ( + E46_CONTRACT_NAME, + E46_REFERENCES_NAME, + E46DetectorTruthIslandError, + read_e46_detector_truth_island, +) + +E48_REVIEW_SCHEMA: Final = "missioncore.e48-independent-detector-review/v1" +E48_ADJUDICATION_SCHEMA: Final = ( + "missioncore.e48-detector-review-adjudication/v1" +) +E48_RESULT_SCHEMA: Final = "missioncore.e48-detector-truth-seal/v1" +E48_TRUTH_ROW_SCHEMA: Final = "missioncore.e48-detector-truth-row/v1" +E48_REPORT_SCHEMA: Final = "missioncore.e48-detector-truth-seal-report/v1" +E48_MANIFEST_NAME: Final = "manifest.json" +E48_REPORT_NAME: Final = "truth-seal-report.json" +E48_TRUTH_NAME: Final = "adjudicated-truth.jsonl" +E48_PROVENANCE_NAME: Final = "review-provenance.json" + +_E47_RESULT_SCHEMA: Final = "missioncore.e47-detector-candidate-freeze/v1" +_REVIEW_STATE: Final = "completed-independent-no-model-assistance" +_ADJUDICATION_STATE: Final = "completed-adjudicated" +_IDENTIFIER = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{1,63}$") +_OBJECT_ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,63}$") +_AUTHORITY: Final = { + "commands_enabled": False, + "navigation_or_safety_accepted": False, +} +_REVIEW_BLINDNESS: Final = { + "candidate_identity_seen": False, + "model_prelabels_seen": False, + "model_predictions_seen": False, + "model_scores_seen": False, +} + + +class E48DetectorTruthSealError(RuntimeError): + """An independent review or truth-seal artifact is invalid.""" + + +def build_e48_detector_truth_seal( + *, + truth_island_root: Path, + prediction_freeze_root: Path, + reviewer_a_path: Path, + reviewer_b_path: Path, + adjudication_path: Path, + output_root: Path, +) -> dict[str, Any]: + """Validate two blind reviews and seal their explicit adjudication.""" + + try: + truth_island = read_e46_detector_truth_island(truth_island_root) + except E46DetectorTruthIslandError as reason: + raise E48DetectorTruthSealError("E46 truth island is invalid") from reason + if ( + truth_island.report.get("status") + != "prepared-awaiting-independent-human-review" + or truth_island.report.get("blindness", {}).get( + "truth_labels_available" + ) + is not False + ): + raise E48DetectorTruthSealError("E46 truth state is incompatible") + + references = tuple( + _read_jsonl(truth_island.result_root / E46_REFERENCES_NAME) + ) + contract = _read_json(truth_island.result_root / E46_CONTRACT_NAME) + annotation = _object(contract.get("annotation"), "E46 annotation contract") + raw_classes = annotation.get("classes") + if not isinstance(raw_classes, list) or not all( + isinstance(value, str) for value in raw_classes + ): + raise E48DetectorTruthSealError("E46 target classes are invalid") + target_classes = frozenset(raw_classes) + + freeze = _read_prediction_freeze_identity( + prediction_freeze_root, + expected_truth_island_id=truth_island.result_id, + ) + review_a_document = _read_json(reviewer_a_path.resolve(strict=True)) + review_b_document = _read_json(reviewer_b_path.resolve(strict=True)) + review_a = _validate_review( + review_a_document, + truth_island_id=truth_island.result_id, + references=references, + target_classes=target_classes, + ) + review_b = _validate_review( + review_b_document, + truth_island_id=truth_island.result_id, + references=references, + target_classes=target_classes, + ) + if review_a["reviewer_id"] == review_b["reviewer_id"]: + raise E48DetectorTruthSealError( + "independent reviewer identities must differ" + ) + review_digests = sorted( + ( + _document_sha256(review_a_document), + _document_sha256(review_b_document), + ) + ) + adjudication_document = _read_json(adjudication_path.resolve(strict=True)) + adjudication = _validate_adjudication( + adjudication_document, + truth_island_id=truth_island.result_id, + review_digests=review_digests, + references=references, + target_classes=target_classes, + ) + if _parse_utc(adjudication["sealed_at_utc"]) < _parse_utc( + freeze["created_at_utc"] + ): + raise E48DetectorTruthSealError( + "truth adjudication predates the prediction freeze" + ) + + truth_rows = tuple( + { + "schema_version": E48_TRUTH_ROW_SCHEMA, + "truth_island_sequence": image["truth_island_sequence"], + "image_id": image["image_id"], + "frame_index": image["frame_index"], + "session_seconds": image["session_seconds"], + "role": image["role"], + "group_id": image["group_id"], + "source_path": image["source_path"], + "source_image_sha256": image["source_sha256"], + "hard_negative": image["hard_negative"], + "objects": image["objects"], + "adjudicated": True, + } + for image in adjudication["images"] + ) + truth_sha256 = hashlib.sha256( + b"".join(_canonical_json(row) + b"\n" for row in truth_rows) + ).hexdigest() + review_agreement = _review_agreement( + review_a["images"], + review_b["images"], + ) + class_counts = Counter( + str(obj["category"]) + for row in truth_rows + for obj in _list(row["objects"], "truth objects") + ) + identity = { + "schema_version": E48_RESULT_SCHEMA, + "truth_island": { + "result_id": truth_island.result_id, + "manifest_sha256": _sha256( + truth_island.result_root / E48_MANIFEST_NAME + ), + }, + "prediction_freeze": freeze, + "review_submission_sha256": review_digests, + "adjudication_sha256": _document_sha256(adjudication_document), + "truth_rows_sha256": truth_sha256, + "producer_sha256": _sha256(Path(__file__).resolve(strict=True)), + "authority": _AUTHORITY, + } + identity_sha256 = hashlib.sha256(_canonical_json(identity)).hexdigest() + result_id = f"e48-detector-truth-seal-{identity_sha256}" + destination = output_root.expanduser().absolute() / result_id + if destination.exists(): + return read_e48_detector_truth_seal(destination) + + provenance = { + "schema_version": E48_RESULT_SCHEMA, + "truth_island_id": truth_island.result_id, + "prediction_freeze_id": freeze["result_id"], + "prediction_content_read_by_sealer": False, + "reviewers": [ + { + "reviewer_id": review_a["reviewer_id"], + "submission_sha256": _document_sha256(review_a_document), + "submitted_at_utc": review_a["submitted_at_utc"], + "blindness": _REVIEW_BLINDNESS, + }, + { + "reviewer_id": review_b["reviewer_id"], + "submission_sha256": _document_sha256(review_b_document), + "submitted_at_utc": review_b["submitted_at_utc"], + "blindness": _REVIEW_BLINDNESS, + }, + ], + "adjudicator_id": adjudication["adjudicator_id"], + "adjudication_sha256": _document_sha256(adjudication_document), + "sealed_at_utc": adjudication["sealed_at_utc"], + "authority": _AUTHORITY, + } + report = { + "schema_version": E48_REPORT_SCHEMA, + "result_id": result_id, + "identity_sha256": identity_sha256, + "status": "sealed-adjudicated-independent-truth", + "frame_count": len(truth_rows), + "object_count": sum( + len(_list(row["objects"], "truth objects")) for row in truth_rows + ), + "hard_negative_frame_count": sum( + 1 for row in truth_rows if row["hard_negative"] is True + ), + "class_counts": dict(sorted(class_counts.items())), + "review_agreement_before_adjudication": review_agreement, + "blindness": { + "independent_reviewer_count": 2, + "reviewer_identities_differ": True, + "model_material_seen_during_review": False, + "prediction_content_read_by_sealer": False, + "adjudication_complete": True, + }, + "decision": { + "truth_labels_available": True, + "truth_join_authorized": True, + "candidate_comparison_authorized": True, + "candidate_winner_selected": False, + "next_gate": ( + "evaluate the already-frozen E47 predictions without " + "retraining or changing either candidate" + ), + }, + "limitations": [ + ( + "the sealed truth remains source-scoped to the known " + "RAVNOVES00 right-camera sample" + ), + ( + "review agreement is diagnostic provenance and does not " + "replace adjudicated truth" + ), + ( + "reviewer independence and model-blindness are explicit " + "human attestations bound to distinct opaque identities" + ), + ( + "truth sealing does not authorize navigation, safety or " + "device commands" + ), + ], + "authority": _AUTHORITY, + } + destination.parent.mkdir(mode=0o700, parents=True, exist_ok=True) + staging = destination.parent / f".{result_id}.{uuid.uuid4().hex}.tmp" + staging.mkdir(mode=0o700, exist_ok=False) + try: + _write_jsonl(staging / E48_TRUTH_NAME, truth_rows) + _write_json(staging / E48_PROVENANCE_NAME, provenance) + _write_json(staging / E48_REPORT_NAME, report) + manifest = { + "schema_version": E48_RESULT_SCHEMA, + "result_id": result_id, + "identity_sha256": identity_sha256, + "identity": identity, + "created_at_utc": _utc_now(), + "acceptance_state": "accepted-adjudicated-truth-only", + "artifacts": [ + _artifact(staging / E48_REPORT_NAME, "truth-seal-report"), + _artifact(staging / E48_TRUTH_NAME, "adjudicated-truth"), + _artifact(staging / E48_PROVENANCE_NAME, "review-provenance"), + ], + "authority": _AUTHORITY, + } + _write_json(staging / E48_MANIFEST_NAME, manifest) + os.replace(staging, destination) + except BaseException: + shutil.rmtree(staging, ignore_errors=True) + raise + return read_e48_detector_truth_seal(destination) + + +def read_e48_detector_truth_seal(root: Path) -> dict[str, Any]: + """Read and revalidate an immutable E48 truth generation.""" + + resolved = root.resolve(strict=True) + manifest = _read_json(resolved / E48_MANIFEST_NAME) + identity = _object(manifest.get("identity"), "E48 identity") + identity_sha256 = manifest.get("identity_sha256") + if ( + manifest.get("schema_version") != E48_RESULT_SCHEMA + or not isinstance(identity_sha256, str) + or hashlib.sha256(_canonical_json(identity)).hexdigest() + != identity_sha256 + or manifest.get("result_id") + != f"e48-detector-truth-seal-{identity_sha256}" + or resolved.name != manifest.get("result_id") + or manifest.get("acceptance_state") + != "accepted-adjudicated-truth-only" + or manifest.get("authority") != _AUTHORITY + ): + raise E48DetectorTruthSealError("E48 identity is invalid") + artifacts = manifest.get("artifacts") + if not isinstance(artifacts, list) or len(artifacts) != 3: + raise E48DetectorTruthSealError("E48 artifacts are invalid") + for item in artifacts: + artifact = _object(item, "E48 artifact") + relative = artifact.get("path") + if not isinstance(relative, str): + raise E48DetectorTruthSealError("E48 artifact path is invalid") + path = resolved / relative + if ( + not path.is_file() + or artifact.get("byte_length") != path.stat().st_size + or artifact.get("sha256") != _sha256(path) + ): + raise E48DetectorTruthSealError("E48 artifact changed") + rows = tuple(_read_jsonl(resolved / E48_TRUTH_NAME)) + report = _read_json(resolved / E48_REPORT_NAME) + provenance = _read_json(resolved / E48_PROVENANCE_NAME) + if ( + report.get("schema_version") != E48_REPORT_SCHEMA + or report.get("result_id") != resolved.name + or report.get("identity_sha256") != identity_sha256 + or report.get("status") != "sealed-adjudicated-independent-truth" + or provenance.get("prediction_content_read_by_sealer") is not False + or any(row.get("schema_version") != E48_TRUTH_ROW_SCHEMA for row in rows) + or hashlib.sha256( + b"".join(_canonical_json(row) + b"\n" for row in rows) + ).hexdigest() + != identity.get("truth_rows_sha256") + ): + raise E48DetectorTruthSealError("E48 result is invalid") + return { + "result_id": resolved.name, + "result_root": resolved, + "manifest": manifest, + "report": report, + "provenance": provenance, + "truth_rows": rows, + } + + +def _validate_review( + document: dict[str, Any], + *, + truth_island_id: str, + references: tuple[dict[str, Any], ...], + target_classes: frozenset[str], +) -> dict[str, Any]: + _exact_keys( + document, + { + "schema_version", + "truth_island_id", + "state", + "reviewer_id", + "review_round", + "blindness", + "images", + "acceptance", + }, + "review submission", + ) + reviewer_id = _identifier(document.get("reviewer_id"), "reviewer_id") + if ( + document.get("schema_version") != E48_REVIEW_SCHEMA + or document.get("truth_island_id") != truth_island_id + or document.get("state") != _REVIEW_STATE + or document.get("review_round") != 1 + or document.get("blindness") != _REVIEW_BLINDNESS + ): + raise E48DetectorTruthSealError("review identity or blindness is invalid") + acceptance = _object(document.get("acceptance"), "review acceptance") + _exact_keys( + acceptance, + {"all_images_reviewed", "independent", "submitted_at_utc"}, + "review acceptance", + ) + if ( + acceptance.get("all_images_reviewed") is not True + or acceptance.get("independent") is not True + ): + raise E48DetectorTruthSealError("review acceptance is incomplete") + submitted_at_utc = _utc_timestamp( + acceptance.get("submitted_at_utc"), + "review submitted_at_utc", + ) + images = _normalize_images( + document.get("images"), + references=references, + target_classes=target_classes, + expected_state="reviewed", + ) + return { + "reviewer_id": reviewer_id, + "submitted_at_utc": submitted_at_utc, + "images": images, + } + + +def _validate_adjudication( + document: dict[str, Any], + *, + truth_island_id: str, + review_digests: list[str], + references: tuple[dict[str, Any], ...], + target_classes: frozenset[str], +) -> dict[str, Any]: + _exact_keys( + document, + { + "schema_version", + "truth_island_id", + "state", + "adjudicator_id", + "review_submission_sha256", + "images", + "acceptance", + }, + "adjudication", + ) + adjudicator_id = _identifier( + document.get("adjudicator_id"), + "adjudicator_id", + ) + digests = document.get("review_submission_sha256") + if ( + document.get("schema_version") != E48_ADJUDICATION_SCHEMA + or document.get("truth_island_id") != truth_island_id + or document.get("state") != _ADJUDICATION_STATE + or not isinstance(digests, list) + or sorted(digests) != review_digests + ): + raise E48DetectorTruthSealError("adjudication identity is invalid") + acceptance = _object(document.get("acceptance"), "adjudication acceptance") + _exact_keys( + acceptance, + { + "all_images_adjudicated", + "all_disagreements_resolved", + "sealed_at_utc", + }, + "adjudication acceptance", + ) + if ( + acceptance.get("all_images_adjudicated") is not True + or acceptance.get("all_disagreements_resolved") is not True + ): + raise E48DetectorTruthSealError("adjudication acceptance is incomplete") + sealed_at_utc = _utc_timestamp( + acceptance.get("sealed_at_utc"), + "adjudication sealed_at_utc", + ) + images = _normalize_images( + document.get("images"), + references=references, + target_classes=target_classes, + expected_state="adjudicated", + ) + return { + "adjudicator_id": adjudicator_id, + "sealed_at_utc": sealed_at_utc, + "images": images, + } + + +def _normalize_images( + value: object, + *, + references: tuple[dict[str, Any], ...], + target_classes: frozenset[str], + expected_state: str, +) -> tuple[dict[str, Any], ...]: + images = _list(value, "review images") + if len(images) != len(references): + raise E48DetectorTruthSealError("review image coverage is incomplete") + normalized: list[dict[str, Any]] = [] + for raw_image, reference in zip(images, references, strict=True): + image = _object(raw_image, "review image") + _exact_keys( + image, + { + "truth_island_sequence", + "image_id", + "frame_index", + "session_seconds", + "role", + "group_id", + "source_path", + "source_sha256", + "review_state", + "hard_negative", + "objects", + "notes", + }, + "review image", + ) + expected_identity = { + "truth_island_sequence": reference["truth_island_sequence"], + "image_id": reference["image_id"], + "frame_index": reference["frame_index"], + "session_seconds": reference["session_seconds"], + "role": reference["role"], + "group_id": reference["group_id"], + "source_path": reference["source_path"], + "source_sha256": reference["sha256"], + } + if any(image.get(key) != expected for key, expected in expected_identity.items()): + raise E48DetectorTruthSealError("review source identity changed") + if image.get("review_state") != expected_state: + raise E48DetectorTruthSealError("review image state is incomplete") + hard_negative = image.get("hard_negative") + if not isinstance(hard_negative, bool): + raise E48DetectorTruthSealError("hard-negative state is missing") + notes = _optional_text(image.get("notes"), "image notes") + objects = _normalize_objects( + image.get("objects"), + target_classes=target_classes, + ) + if hard_negative != (len(objects) == 0): + raise E48DetectorTruthSealError( + "hard-negative state conflicts with reviewed objects" + ) + normalized.append( + { + **expected_identity, + "review_state": expected_state, + "hard_negative": hard_negative, + "objects": list(objects), + "notes": notes, + } + ) + return tuple(normalized) + + +def _normalize_objects( + value: object, + *, + target_classes: frozenset[str], +) -> tuple[dict[str, Any], ...]: + objects = _list(value, "review objects") + normalized: list[dict[str, Any]] = [] + object_ids: set[str] = set() + for raw_object in objects: + obj = _object(raw_object, "review object") + _exact_keys( + obj, + { + "object_id", + "category", + "box_xyxy", + "occluded", + "truncated", + "notes", + }, + "review object", + ) + object_id = obj.get("object_id") + if ( + not isinstance(object_id, str) + or _OBJECT_ID.fullmatch(object_id) is None + or object_id in object_ids + ): + raise E48DetectorTruthSealError("review object id is invalid") + object_ids.add(object_id) + category = obj.get("category") + if category not in target_classes: + raise E48DetectorTruthSealError("review object class is invalid") + box = _box(obj.get("box_xyxy")) + occluded = obj.get("occluded") + truncated = obj.get("truncated") + if not isinstance(occluded, bool) or not isinstance(truncated, bool): + raise E48DetectorTruthSealError("review object flags are invalid") + normalized.append( + { + "object_id": object_id, + "category": category, + "box_xyxy": box, + "occluded": occluded, + "truncated": truncated, + "notes": _optional_text(obj.get("notes"), "object notes"), + } + ) + normalized.sort(key=lambda row: str(row["object_id"])) + return tuple(normalized) + + +def _review_agreement( + review_a: tuple[dict[str, Any], ...], + review_b: tuple[dict[str, Any], ...], +) -> dict[str, Any]: + matched_count = 0 + matched_ious: list[float] = [] + unmatched_a = 0 + unmatched_b = 0 + hard_negative_agreement = 0 + for image_a, image_b in zip(review_a, review_b, strict=True): + if image_a["hard_negative"] == image_b["hard_negative"]: + hard_negative_agreement += 1 + objects_a = _list(image_a["objects"], "review A objects") + objects_b = _list(image_b["objects"], "review B objects") + available_b = set(range(len(objects_b))) + for obj_a in objects_a: + candidate = _object(obj_a, "review A object") + best: tuple[float, int] | None = None + for index in available_b: + other = _object(objects_b[index], "review B object") + if candidate["category"] != other["category"]: + continue + overlap = box_iou( + _box(candidate["box_xyxy"]), + _box(other["box_xyxy"]), + ) + if best is None or overlap > best[0]: + best = (overlap, index) + if best is not None and best[0] >= 0.5: + matched_count += 1 + matched_ious.append(best[0]) + available_b.remove(best[1]) + else: + unmatched_a += 1 + unmatched_b += len(available_b) + return { + "matched_same_class_iou_gte_0_5": matched_count, + "unmatched_reviewer_a": unmatched_a, + "unmatched_reviewer_b": unmatched_b, + "matched_iou_mean": ( + round(sum(matched_ious) / len(matched_ious), 9) + if matched_ious + else None + ), + "hard_negative_agreement_frames": hard_negative_agreement, + "frame_count": len(review_a), + } + + +def box_iou(left: list[float], right: list[float]) -> float: + """Return axis-aligned intersection over union for two validated boxes.""" + + intersection_width = max(0.0, min(left[2], right[2]) - max(left[0], right[0])) + intersection_height = max( + 0.0, + min(left[3], right[3]) - max(left[1], right[1]), + ) + intersection = intersection_width * intersection_height + left_area = (left[2] - left[0]) * (left[3] - left[1]) + right_area = (right[2] - right[0]) * (right[3] - right[1]) + union = left_area + right_area - intersection + return intersection / union if union > 0.0 else 0.0 + + +def _read_prediction_freeze_identity( + root: Path, + *, + expected_truth_island_id: str, +) -> dict[str, Any]: + resolved = root.resolve(strict=True) + manifest_path = resolved / E48_MANIFEST_NAME + manifest = _read_json(manifest_path) + identity = _object(manifest.get("identity"), "E47 identity") + identity_sha256 = manifest.get("identity_sha256") + truth_island = _object(identity.get("truth_island"), "E47 truth island") + if ( + manifest.get("schema_version") != _E47_RESULT_SCHEMA + or not isinstance(identity_sha256, str) + or hashlib.sha256(_canonical_json(identity)).hexdigest() + != identity_sha256 + or manifest.get("result_id") + != f"e47-detector-candidate-freeze-{identity_sha256}" + or resolved.name != manifest.get("result_id") + or manifest.get("acceptance_state") + != "accepted-prediction-freeze-only" + or manifest.get("authority") != _AUTHORITY + or truth_island.get("result_id") != expected_truth_island_id + or truth_island.get("truth_labels_available") is not False + ): + raise E48DetectorTruthSealError("E47 prediction freeze is invalid") + _utc_timestamp(manifest.get("created_at_utc"), "E47 created_at_utc") + return { + "result_id": resolved.name, + "manifest_sha256": _sha256(manifest_path), + "identity_sha256": identity_sha256, + "created_at_utc": manifest["created_at_utc"], + "prediction_content_read_by_sealer": False, + } + + +def _box(value: object) -> list[float]: + if ( + not isinstance(value, list) + or len(value) != 4 + or not all( + isinstance(item, (int, float)) + and not isinstance(item, bool) + and math.isfinite(float(item)) + for item in value + ) + ): + raise E48DetectorTruthSealError("review object box is invalid") + box = [float(item) for item in value] + if not ( + 0.0 <= box[0] < box[2] <= 800.0 + and 0.0 <= box[1] < box[3] <= 600.0 + ): + raise E48DetectorTruthSealError("review object box is out of bounds") + return box + + +def _identifier(value: object, field: str) -> str: + if not isinstance(value, str) or _IDENTIFIER.fullmatch(value) is None: + raise E48DetectorTruthSealError(f"{field} is invalid") + return value + + +def _optional_text(value: object, field: str) -> str | None: + if value is None: + return None + if not isinstance(value, str) or len(value) > 2000: + raise E48DetectorTruthSealError(f"{field} is invalid") + return value + + +def _utc_timestamp(value: object, field: str) -> str: + if not isinstance(value, str): + raise E48DetectorTruthSealError(f"{field} is invalid") + try: + parsed = datetime.fromisoformat(value.replace("Z", "+00:00")) + except ValueError as reason: + raise E48DetectorTruthSealError(f"{field} is invalid") from reason + if parsed.tzinfo is None or parsed.utcoffset() != UTC.utcoffset(parsed): + raise E48DetectorTruthSealError(f"{field} must be UTC") + return value + + +def _parse_utc(value: str) -> datetime: + return datetime.fromisoformat(value.replace("Z", "+00:00")) + + +def _exact_keys( + value: dict[str, Any], + expected: set[str], + field: str, +) -> None: + if set(value) != expected: + raise E48DetectorTruthSealError(f"{field} fields are invalid") + + +def _object(value: object, field: str) -> dict[str, Any]: + if not isinstance(value, dict): + raise E48DetectorTruthSealError(f"{field} must be an object") + return value + + +def _list(value: object, field: str) -> list[Any]: + if not isinstance(value, list): + raise E48DetectorTruthSealError(f"{field} must be a list") + return value + + +def _read_json(path: Path) -> dict[str, Any]: + try: + value = json.loads(path.read_text(encoding="utf-8")) + except (OSError, ValueError) as reason: + raise E48DetectorTruthSealError(f"cannot read {path.name}") from reason + return _object(value, path.name) + + +def _read_jsonl(path: Path) -> Iterable[dict[str, Any]]: + try: + with path.open("r", encoding="utf-8") as handle: + for line in handle: + yield _object(json.loads(line), path.name) + except (OSError, ValueError) as reason: + raise E48DetectorTruthSealError(f"cannot read {path.name}") from reason + + +def _document_sha256(document: dict[str, Any]) -> str: + return hashlib.sha256(_canonical_json(document)).hexdigest() + + +def _canonical_json(value: object) -> bytes: + return json.dumps( + value, + ensure_ascii=False, + sort_keys=True, + separators=(",", ":"), + allow_nan=False, + ).encode("utf-8") + + +def _sha256(path: Path) -> str: + digest = hashlib.sha256() + with path.open("rb") as handle: + for chunk in iter(lambda: handle.read(1024 * 1024), b""): + digest.update(chunk) + return digest.hexdigest() + + +def _artifact(path: Path, role: str) -> dict[str, Any]: + return { + "path": path.name, + "role": role, + "byte_length": path.stat().st_size, + "sha256": _sha256(path), + } + + +def _write_json(path: Path, value: object) -> None: + path.write_text( + json.dumps( + value, + ensure_ascii=False, + sort_keys=True, + indent=2, + allow_nan=False, + ) + + "\n", + encoding="utf-8", + ) + + +def _write_jsonl(path: Path, rows: Iterable[object]) -> None: + with path.open("w", encoding="utf-8") as handle: + for row in rows: + handle.write(_canonical_json(row).decode("utf-8")) + handle.write("\n") + + +def _utc_now() -> str: + return datetime.now(UTC).isoformat().replace("+00:00", "Z") diff --git a/src/k1link/compute/e49_detector_truth_evaluation.py b/src/k1link/compute/e49_detector_truth_evaluation.py new file mode 100644 index 0000000..66b6fb9 --- /dev/null +++ b/src/k1link/compute/e49_detector_truth_evaluation.py @@ -0,0 +1,819 @@ +"""Evaluate frozen E47 detector candidates only after an accepted E48 seal.""" + +from __future__ import annotations + +import hashlib +import json +import math +import os +import shutil +import uuid +from collections import defaultdict +from collections.abc import Iterable +from datetime import UTC, datetime +from pathlib import Path +from typing import Any, Final + +from PIL import Image + +from .e46_detector_truth_island import ( + E46DetectorTruthIslandError, + read_e46_detector_truth_island, +) +from .e47_detector_candidate_freeze import ( + E47_PREDICTIONS_NAME, + E47DetectorCandidateFreezeError, + read_e47_detector_candidate_freeze, +) +from .e48_detector_truth_seal import ( + E48DetectorTruthSealError, + box_iou, + read_e48_detector_truth_seal, +) + +E49_RESULT_SCHEMA: Final = "missioncore.e49-detector-truth-evaluation/v1" +E49_REPORT_SCHEMA: Final = "missioncore.e49-detector-evaluation-report/v1" +E49_MANIFEST_NAME: Final = "manifest.json" +E49_REPORT_NAME: Final = "detector-evaluation-report.json" + +_VALID_FOV_SCHEMA: Final = "missioncore.k1-valid-fov-mask/v1" +_IOU_THRESHOLDS: Final = tuple(round(0.5 + index * 0.05, 2) for index in range(10)) +_CRITICAL_CLASSES: Final = frozenset( + {"person", "bicycle", "motorcycle", "car", "heavy_vehicle"} +) +_LARGE_BOX_FRACTION: Final = 0.25 +_AUTHORITY: Final = { + "commands_enabled": False, + "navigation_or_safety_accepted": False, +} + + +class E49DetectorTruthEvaluationError(RuntimeError): + """A sealed truth, prediction freeze or evaluation result is invalid.""" + + +def build_e49_detector_truth_evaluation( + *, + truth_island_root: Path, + truth_seal_root: Path, + prediction_freeze_root: Path, + valid_fov_root: Path, + output_root: Path, +) -> dict[str, Any]: + """Join accepted truth with the exact frozen prediction generation.""" + + try: + truth_island = read_e46_detector_truth_island(truth_island_root) + truth_seal = read_e48_detector_truth_seal(truth_seal_root) + prediction_freeze = read_e47_detector_candidate_freeze( + prediction_freeze_root + ) + except ( + E46DetectorTruthIslandError, + E47DetectorCandidateFreezeError, + E48DetectorTruthSealError, + ) as reason: + raise E49DetectorTruthEvaluationError( + "detector evaluation input is invalid" + ) from reason + if ( + truth_seal["report"].get("status") + != "sealed-adjudicated-independent-truth" + or truth_seal["report"].get("decision", {}).get( + "candidate_comparison_authorized" + ) + is not True + ): + raise E49DetectorTruthEvaluationError("E48 truth is not accepted") + seal_identity = _object( + truth_seal["manifest"].get("identity"), + "E48 identity", + ) + sealed_source = _object( + seal_identity.get("truth_island"), + "E48 truth source", + ) + sealed_freeze = _object( + seal_identity.get("prediction_freeze"), + "E48 prediction freeze", + ) + if ( + sealed_source.get("result_id") != truth_island.result_id + or sealed_freeze.get("result_id") != prediction_freeze["result_id"] + ): + raise E49DetectorTruthEvaluationError( + "truth, seal and prediction identities differ" + ) + freeze_identity = _object( + prediction_freeze["manifest"].get("identity"), + "E47 identity", + ) + freeze_source = _object( + freeze_identity.get("truth_island"), + "E47 truth island", + ) + if freeze_source.get("result_id") != truth_island.result_id: + raise E49DetectorTruthEvaluationError( + "prediction freeze belongs to another truth island" + ) + + valid_fov = _read_valid_fov( + valid_fov_root, + calibration_sha256=str( + truth_island.manifest["identity"]["source"]["calibration_sha256"] + ), + calibration_slot=str( + truth_island.manifest["identity"]["source"]["calibration_slot"] + ), + ) + prediction_rows = tuple( + _read_jsonl( + prediction_freeze["result_root"] / E47_PREDICTIONS_NAME + ) + ) + truth_rows = tuple(truth_seal["truth_rows"]) + metrics = evaluate_frozen_detector_candidates( + truth_rows=truth_rows, + prediction_rows=prediction_rows, + valid_fov_mask=valid_fov["mask"], + ) + profile = { + "profile_id": "e49-ravnoves00-detector-evaluation/v1", + "iou_thresholds": list(_IOU_THRESHOLDS), + "ap_interpolation_recall_points": 101, + "max_detections_per_image": 100, + "per_class_recall_iou": 0.5, + "person_vehicle_miss_iou": 0.5, + "false_large_box_iou": 0.5, + "large_box_image_fraction": _LARGE_BOX_FRACTION, + "valid_fov_leakage_rule": "prediction-box-centre-outside-mask", + "temporal_flicker_rule": ( + "mean normalized per-class detection-count delta over adjacent " + "frames in each frozen temporal group" + ), + "candidate_selection_policy": "no-automatic-winner", + } + identity = { + "schema_version": E49_RESULT_SCHEMA, + "truth_island": { + "result_id": truth_island.result_id, + "manifest_sha256": _sha256( + truth_island.result_root / E49_MANIFEST_NAME + ), + }, + "truth_seal": { + "result_id": truth_seal["result_id"], + "manifest_sha256": _sha256( + truth_seal["result_root"] / E49_MANIFEST_NAME + ), + }, + "prediction_freeze": { + "result_id": prediction_freeze["result_id"], + "manifest_sha256": _sha256( + prediction_freeze["result_root"] / E49_MANIFEST_NAME + ), + }, + "valid_fov": valid_fov["identity"], + "profile": profile, + "metrics_sha256": hashlib.sha256(_canonical_json(metrics)).hexdigest(), + "producer_sha256": _sha256(Path(__file__).resolve(strict=True)), + "authority": _AUTHORITY, + } + identity_sha256 = hashlib.sha256(_canonical_json(identity)).hexdigest() + result_id = f"e49-detector-truth-evaluation-{identity_sha256}" + destination = output_root.expanduser().absolute() / result_id + if destination.exists(): + return read_e49_detector_truth_evaluation(destination) + + report = { + "schema_version": E49_REPORT_SCHEMA, + "result_id": result_id, + "identity_sha256": identity_sha256, + "status": "completed-sealed-truth-candidate-comparison", + "frame_count": len(truth_rows), + "candidate_count": len(metrics), + "profile": profile, + "candidates": metrics, + "decision": { + "truth_join_performed": True, + "accuracy_metrics_available": True, + "candidate_winner_selected": False, + "model_retraining_authorized": False, + "next_gate": ( + "review the preregistered metrics and explicitly accept, " + "reject or retain both preprocessing candidates" + ), + }, + "limitations": [ + ( + "metrics are source-scoped to the 32-frame RAVNOVES00 " + "detector Truth Island" + ), + ( + "valid-FOV leakage measures box-centre admission rather than " + "full mask or box-area leakage" + ), + ( + "temporal flicker is a class-count stability metric because " + "the frozen detector candidates contain no track identity" + ), + ], + "authority": _AUTHORITY, + } + destination.parent.mkdir(mode=0o700, parents=True, exist_ok=True) + staging = destination.parent / f".{result_id}.{uuid.uuid4().hex}.tmp" + staging.mkdir(mode=0o700, exist_ok=False) + try: + _write_json(staging / E49_REPORT_NAME, report) + manifest = { + "schema_version": E49_RESULT_SCHEMA, + "result_id": result_id, + "identity_sha256": identity_sha256, + "identity": identity, + "created_at_utc": _utc_now(), + "acceptance_state": "accepted-metrics-only-no-winner", + "artifacts": [ + _artifact(staging / E49_REPORT_NAME, "evaluation-report") + ], + "authority": _AUTHORITY, + } + _write_json(staging / E49_MANIFEST_NAME, manifest) + os.replace(staging, destination) + except BaseException: + shutil.rmtree(staging, ignore_errors=True) + raise + return read_e49_detector_truth_evaluation(destination) + + +def read_e49_detector_truth_evaluation(root: Path) -> dict[str, Any]: + """Read and revalidate an immutable E49 evaluation result.""" + + resolved = root.resolve(strict=True) + manifest = _read_json(resolved / E49_MANIFEST_NAME) + identity = _object(manifest.get("identity"), "E49 identity") + identity_sha256 = manifest.get("identity_sha256") + if ( + manifest.get("schema_version") != E49_RESULT_SCHEMA + or not isinstance(identity_sha256, str) + or hashlib.sha256(_canonical_json(identity)).hexdigest() + != identity_sha256 + or manifest.get("result_id") + != f"e49-detector-truth-evaluation-{identity_sha256}" + or resolved.name != manifest.get("result_id") + or manifest.get("acceptance_state") + != "accepted-metrics-only-no-winner" + or manifest.get("authority") != _AUTHORITY + ): + raise E49DetectorTruthEvaluationError("E49 identity is invalid") + artifacts = manifest.get("artifacts") + if not isinstance(artifacts, list) or len(artifacts) != 1: + raise E49DetectorTruthEvaluationError("E49 artifacts are invalid") + artifact = _object(artifacts[0], "E49 report artifact") + path = resolved / str(artifact.get("path")) + if ( + not path.is_file() + or artifact.get("byte_length") != path.stat().st_size + or artifact.get("sha256") != _sha256(path) + ): + raise E49DetectorTruthEvaluationError("E49 report changed") + report = _read_json(path) + if ( + report.get("schema_version") != E49_REPORT_SCHEMA + or report.get("result_id") != resolved.name + or report.get("identity_sha256") != identity_sha256 + or report.get("status") + != "completed-sealed-truth-candidate-comparison" + or hashlib.sha256( + _canonical_json(report.get("candidates")) + ).hexdigest() + != identity.get("metrics_sha256") + ): + raise E49DetectorTruthEvaluationError("E49 report is invalid") + return { + "result_id": resolved.name, + "result_root": resolved, + "manifest": manifest, + "report": report, + } + + +def evaluate_frozen_detector_candidates( + *, + truth_rows: tuple[dict[str, Any], ...], + prediction_rows: tuple[dict[str, Any], ...], + valid_fov_mask: Image.Image, +) -> dict[str, Any]: + """Compute frozen detection metrics without selecting a winner.""" + + if not truth_rows: + raise E49DetectorTruthEvaluationError("truth rows are empty") + if valid_fov_mask.mode != "L" or valid_fov_mask.size != (800, 600): + raise E49DetectorTruthEvaluationError("valid-FOV mask is invalid") + truth_by_sequence: dict[int, dict[str, Any]] = {} + for row in truth_rows: + sequence = _integer(row.get("truth_island_sequence"), "truth sequence") + if sequence in truth_by_sequence: + raise E49DetectorTruthEvaluationError("truth sequence is duplicated") + truth_by_sequence[sequence] = row + candidate_rows: dict[str, dict[int, dict[str, Any]]] = defaultdict(dict) + for row in prediction_rows: + candidate_id = _text(row.get("candidate_id"), "candidate id") + sequence = _integer( + row.get("truth_island_sequence"), + "prediction sequence", + ) + truth = truth_by_sequence.get(sequence) + if truth is None: + raise E49DetectorTruthEvaluationError( + "prediction has no sealed truth row" + ) + if ( + row.get("source_image_sha256") + != truth.get("source_image_sha256") + or row.get("frame_index") != truth.get("frame_index") + or row.get("image_id") != truth.get("image_id") + or row.get("truth_joined") is not False + or sequence in candidate_rows[candidate_id] + ): + raise E49DetectorTruthEvaluationError( + "prediction source identity is invalid" + ) + candidate_rows[candidate_id][sequence] = row + expected_sequences = set(truth_by_sequence) + if not candidate_rows or any( + set(rows) != expected_sequences for rows in candidate_rows.values() + ): + raise E49DetectorTruthEvaluationError( + "candidate prediction coverage is incomplete" + ) + return { + candidate_id: _candidate_metrics( + truth_by_sequence=truth_by_sequence, + prediction_by_sequence=rows, + valid_fov_mask=valid_fov_mask, + ) + for candidate_id, rows in sorted(candidate_rows.items()) + } + + +def _candidate_metrics( + *, + truth_by_sequence: dict[int, dict[str, Any]], + prediction_by_sequence: dict[int, dict[str, Any]], + valid_fov_mask: Image.Image, +) -> dict[str, Any]: + categories = sorted( + { + str(obj["category"]) + for truth in truth_by_sequence.values() + for obj in _objects(truth) + } + ) + if not categories: + raise E49DetectorTruthEvaluationError( + "sealed truth contains no detector objects" + ) + ap_by_threshold: dict[float, list[float]] = defaultdict(list) + ar_by_threshold: dict[float, list[float]] = defaultdict(list) + per_class: dict[str, dict[str, Any]] = {} + for category in categories: + class_thresholds: dict[float, tuple[float, float]] = {} + for threshold in _IOU_THRESHOLDS: + average_precision, recall = _class_ap_recall( + category=category, + threshold=threshold, + truth_by_sequence=truth_by_sequence, + prediction_by_sequence=prediction_by_sequence, + ) + class_thresholds[threshold] = (average_precision, recall) + ap_by_threshold[threshold].append(average_precision) + ar_by_threshold[threshold].append(recall) + per_class[category] = { + "ground_truth_count": sum( + 1 + for truth in truth_by_sequence.values() + for obj in _objects(truth) + if obj["category"] == category + ), + "ap_50_95": _mean( + [values[0] for values in class_thresholds.values()] + ), + "ap50": class_thresholds[0.5][0], + "ap75": class_thresholds[0.75][0], + "recall50": class_thresholds[0.5][1], + } + all_ap = [value for values in ap_by_threshold.values() for value in values] + all_ar = [value for values in ar_by_threshold.values() for value in values] + critical_total = 0 + critical_missed = 0 + for category in sorted(_CRITICAL_CLASSES & set(categories)): + matched, total = _class_match_count( + category=category, + threshold=0.5, + truth_by_sequence=truth_by_sequence, + prediction_by_sequence=prediction_by_sequence, + ) + critical_total += total + critical_missed += total - matched + large_predictions = 0 + false_large_predictions = 0 + boundary_leaks = 0 + prediction_count = 0 + for sequence, prediction_row in prediction_by_sequence.items(): + predictions = _predictions(prediction_row) + prediction_count += len(predictions) + matched_indices = _matched_prediction_indices( + truth=_objects(truth_by_sequence[sequence]), + predictions=predictions, + threshold=0.5, + ) + for index, prediction in enumerate(predictions): + box = _box(prediction.get("box_xyxy")) + area_fraction = ( + (box[2] - box[0]) * (box[3] - box[1]) / (800.0 * 600.0) + ) + if area_fraction >= _LARGE_BOX_FRACTION: + large_predictions += 1 + if index not in matched_indices: + false_large_predictions += 1 + centre_x = min(799, max(0, int((box[0] + box[2]) / 2.0))) + centre_y = min(599, max(0, int((box[1] + box[3]) / 2.0))) + if valid_fov_mask.getpixel((centre_x, centre_y)) == 0: + boundary_leaks += 1 + return { + "coco_ap_50_95": _mean(all_ap), + "ap50": _mean(ap_by_threshold[0.5]), + "ap75": _mean(ap_by_threshold[0.75]), + "ar100": _mean(all_ar), + "per_class": per_class, + "person_vehicle_miss_rate": ( + round(critical_missed / critical_total, 9) + if critical_total + else None + ), + "person_vehicle_ground_truth_count": critical_total, + "false_large_box_rate": ( + round(false_large_predictions / large_predictions, 9) + if large_predictions + else 0.0 + ), + "large_prediction_count": large_predictions, + "valid_fov_boundary_leakage": ( + round(boundary_leaks / prediction_count, 9) + if prediction_count + else 0.0 + ), + "prediction_count": prediction_count, + "temporal_detection_flicker": _temporal_flicker( + truth_by_sequence=truth_by_sequence, + prediction_by_sequence=prediction_by_sequence, + ), + "candidate_winner_selected": False, + } + + +def _class_ap_recall( + *, + category: str, + threshold: float, + truth_by_sequence: dict[int, dict[str, Any]], + prediction_by_sequence: dict[int, dict[str, Any]], +) -> tuple[float, float]: + ground_truth = { + sequence: [ + obj for obj in _objects(row) if obj["category"] == category + ] + for sequence, row in truth_by_sequence.items() + } + total_truth = sum(len(objects) for objects in ground_truth.values()) + predictions = sorted( + ( + ( + float(prediction["score"]), + sequence, + prediction, + ) + for sequence, row in prediction_by_sequence.items() + for prediction in _predictions(row)[:100] + if prediction.get("category") == category + ), + key=lambda item: (-item[0], item[1]), + ) + matched: dict[int, set[int]] = defaultdict(set) + true_positives: list[int] = [] + false_positives: list[int] = [] + for _, sequence, prediction in predictions: + box = _box(prediction.get("box_xyxy")) + best: tuple[float, int] | None = None + for index, truth in enumerate(ground_truth[sequence]): + if index in matched[sequence]: + continue + overlap = box_iou(box, _box(truth.get("box_xyxy"))) + if best is None or overlap > best[0]: + best = (overlap, index) + if best is not None and best[0] >= threshold: + matched[sequence].add(best[1]) + true_positives.append(1) + false_positives.append(0) + else: + true_positives.append(0) + false_positives.append(1) + if total_truth == 0: + raise E49DetectorTruthEvaluationError("class truth denominator is zero") + cumulative_tp = 0 + cumulative_fp = 0 + recalls: list[float] = [] + precisions: list[float] = [] + for true_positive, false_positive in zip( + true_positives, + false_positives, + strict=True, + ): + cumulative_tp += true_positive + cumulative_fp += false_positive + recalls.append(cumulative_tp / total_truth) + precisions.append(cumulative_tp / (cumulative_tp + cumulative_fp)) + interpolated = [ + max( + ( + precision + for recall, precision in zip(recalls, precisions, strict=True) + if recall >= recall_point / 100.0 + ), + default=0.0, + ) + for recall_point in range(101) + ] + return _mean(interpolated), round(cumulative_tp / total_truth, 9) + + +def _class_match_count( + *, + category: str, + threshold: float, + truth_by_sequence: dict[int, dict[str, Any]], + prediction_by_sequence: dict[int, dict[str, Any]], +) -> tuple[int, int]: + matched = 0 + total = 0 + for sequence, truth_row in truth_by_sequence.items(): + truth = [obj for obj in _objects(truth_row) if obj["category"] == category] + predictions = [ + prediction + for prediction in _predictions(prediction_by_sequence[sequence]) + if prediction.get("category") == category + ] + total += len(truth) + matched += len( + _matched_prediction_indices( + truth=truth, + predictions=predictions, + threshold=threshold, + ) + ) + return min(matched, total), total + + +def _matched_prediction_indices( + *, + truth: list[dict[str, Any]], + predictions: list[dict[str, Any]], + threshold: float, +) -> set[int]: + matched_truth: set[int] = set() + matched_predictions: set[int] = set() + ordered = sorted( + enumerate(predictions), + key=lambda item: -float(item[1].get("score", 0.0)), + ) + for prediction_index, prediction in ordered: + category = prediction.get("category") + box = _box(prediction.get("box_xyxy")) + best: tuple[float, int] | None = None + for truth_index, truth_object in enumerate(truth): + if ( + truth_index in matched_truth + or truth_object.get("category") != category + ): + continue + overlap = box_iou(box, _box(truth_object.get("box_xyxy"))) + if best is None or overlap > best[0]: + best = (overlap, truth_index) + if best is not None and best[0] >= threshold: + matched_truth.add(best[1]) + matched_predictions.add(prediction_index) + return matched_predictions + + +def _temporal_flicker( + *, + truth_by_sequence: dict[int, dict[str, Any]], + prediction_by_sequence: dict[int, dict[str, Any]], +) -> float | None: + groups: dict[str, list[int]] = defaultdict(list) + for sequence, truth in truth_by_sequence.items(): + if truth.get("role") == "temporal": + groups[str(truth.get("group_id"))].append(sequence) + deltas: list[float] = [] + for sequences in groups.values(): + ordered = sorted(sequences) + for left_sequence, right_sequence in zip( + ordered, + ordered[1:], + strict=False, + ): + left = _class_count(_predictions(prediction_by_sequence[left_sequence])) + right = _class_count(_predictions(prediction_by_sequence[right_sequence])) + for category in sorted(set(left) | set(right)): + denominator = max(left.get(category, 0), right.get(category, 0), 1) + deltas.append( + abs(left.get(category, 0) - right.get(category, 0)) + / denominator + ) + return _mean(deltas) if deltas else None + + +def _class_count(predictions: list[dict[str, Any]]) -> dict[str, int]: + result: dict[str, int] = defaultdict(int) + for prediction in predictions: + result[str(prediction.get("category"))] += 1 + return dict(result) + + +def _read_valid_fov( + root: Path, + *, + calibration_sha256: str, + calibration_slot: str, +) -> dict[str, Any]: + resolved = root.resolve(strict=True) + manifest_path = resolved / E49_MANIFEST_NAME + manifest = _read_json(manifest_path) + identity = _object(manifest.get("identity"), "valid-FOV identity") + artifact = _object(manifest.get("artifact"), "valid-FOV artifact") + identity_sha256 = manifest.get("identity_sha256") + path = resolved / str(artifact.get("path")) + if ( + manifest.get("schema_version") != _VALID_FOV_SCHEMA + or not isinstance(identity_sha256, str) + or hashlib.sha256(_canonical_json(identity)).hexdigest() + != identity_sha256 + or manifest.get("generation_id") + != f"valid-fov-mask-{identity_sha256}" + or resolved.name != manifest.get("generation_id") + or identity.get("calibration_sha256") != calibration_sha256 + or identity.get("calibration_slot") != calibration_slot + or identity.get("admitted_resolution") != [800, 600] + or not path.is_file() + or artifact.get("byte_length") != path.stat().st_size + or artifact.get("sha256") != _sha256(path) + ): + raise E49DetectorTruthEvaluationError("valid-FOV identity is invalid") + with Image.open(path) as source: + mask = source.copy() + if ( + mask.mode != "L" + or mask.size != (800, 600) + or set(mask.getdata()) - {0, 255} + ): + raise E49DetectorTruthEvaluationError("valid-FOV pixels are invalid") + return { + "identity": { + "generation_id": resolved.name, + "manifest_sha256": _sha256(manifest_path), + "mask_sha256": artifact["sha256"], + "calibration_sha256": calibration_sha256, + "calibration_slot": calibration_slot, + }, + "mask": mask, + } + + +def _objects(row: dict[str, Any]) -> list[dict[str, Any]]: + value = row.get("objects") + if not isinstance(value, list) or not all( + isinstance(item, dict) for item in value + ): + raise E49DetectorTruthEvaluationError("truth objects are invalid") + return value + + +def _predictions(row: dict[str, Any]) -> list[dict[str, Any]]: + value = row.get("predictions") + if not isinstance(value, list) or not all( + isinstance(item, dict) for item in value + ): + raise E49DetectorTruthEvaluationError("predictions are invalid") + return value + + +def _box(value: object) -> list[float]: + if ( + not isinstance(value, list) + or len(value) != 4 + or not all( + isinstance(item, (int, float)) + and not isinstance(item, bool) + and math.isfinite(float(item)) + for item in value + ) + ): + raise E49DetectorTruthEvaluationError("detector box is invalid") + box = [float(item) for item in value] + if not ( + 0.0 <= box[0] < box[2] <= 800.0 + and 0.0 <= box[1] < box[3] <= 600.0 + ): + raise E49DetectorTruthEvaluationError("detector box is out of bounds") + return box + + +def _mean(values: Iterable[float]) -> float: + materialized = list(values) + if not materialized: + raise E49DetectorTruthEvaluationError("metric denominator is empty") + return round(sum(materialized) / len(materialized), 9) + + +def _integer(value: object, field: str) -> int: + if not isinstance(value, int) or isinstance(value, bool): + raise E49DetectorTruthEvaluationError(f"{field} is invalid") + return value + + +def _text(value: object, field: str) -> str: + if not isinstance(value, str) or not value: + raise E49DetectorTruthEvaluationError(f"{field} is invalid") + return value + + +def _object(value: object, field: str) -> dict[str, Any]: + if not isinstance(value, dict): + raise E49DetectorTruthEvaluationError(f"{field} must be an object") + return value + + +def _read_json(path: Path) -> dict[str, Any]: + try: + value = json.loads(path.read_text(encoding="utf-8")) + except (OSError, ValueError) as reason: + raise E49DetectorTruthEvaluationError( + f"cannot read {path.name}" + ) from reason + return _object(value, path.name) + + +def _read_jsonl(path: Path) -> Iterable[dict[str, Any]]: + try: + with path.open("r", encoding="utf-8") as handle: + for line in handle: + yield _object(json.loads(line), path.name) + except (OSError, ValueError) as reason: + raise E49DetectorTruthEvaluationError( + f"cannot read {path.name}" + ) from reason + + +def _canonical_json(value: object) -> bytes: + return json.dumps( + value, + ensure_ascii=False, + sort_keys=True, + separators=(",", ":"), + allow_nan=False, + ).encode("utf-8") + + +def _sha256(path: Path) -> str: + digest = hashlib.sha256() + with path.open("rb") as handle: + for chunk in iter(lambda: handle.read(1024 * 1024), b""): + digest.update(chunk) + return digest.hexdigest() + + +def _artifact(path: Path, role: str) -> dict[str, Any]: + return { + "path": path.name, + "role": role, + "byte_length": path.stat().st_size, + "sha256": _sha256(path), + } + + +def _write_json(path: Path, value: object) -> None: + path.write_text( + json.dumps( + value, + ensure_ascii=False, + sort_keys=True, + indent=2, + allow_nan=False, + ) + + "\n", + encoding="utf-8", + ) + + +def _utc_now() -> str: + return datetime.now(UTC).isoformat().replace("+00:00", "Z") diff --git a/tests/test_e48_detector_truth_seal.py b/tests/test_e48_detector_truth_seal.py new file mode 100644 index 0000000..b836d5d --- /dev/null +++ b/tests/test_e48_detector_truth_seal.py @@ -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) diff --git a/tests/test_e49_detector_truth_evaluation.py b/tests/test_e49_detector_truth_evaluation.py new file mode 100644 index 0000000..99a8380 --- /dev/null +++ b/tests/test_e49_detector_truth_evaluation.py @@ -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), + )