287 lines
9.0 KiB
Python
287 lines
9.0 KiB
Python
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
import zipfile
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
from typing import Any
|
|
|
|
import pytest
|
|
from PIL import Image
|
|
|
|
from k1link.compute import e46_review_handoff as handoff
|
|
from k1link.compute import 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.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_bytes(_canonical(value) + b"\n")
|
|
|
|
|
|
def _fixture(
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> tuple[Path, Path, SimpleNamespace]:
|
|
truth_root = tmp_path / ("e46-detector-truth-island-" + "a" * 64)
|
|
evaluation_root = tmp_path / ("evaluation-pack-" + "b" * 64)
|
|
truth_root.mkdir()
|
|
image_root = evaluation_root / "images" / "valid-fov-fill"
|
|
image_root.mkdir(parents=True)
|
|
references: list[dict[str, Any]] = []
|
|
review_images: list[dict[str, Any]] = []
|
|
for sequence, image_id in enumerate((2, 5), start=1):
|
|
frame_index = sequence * 10
|
|
name = f"image-{image_id:03d}-frame-{frame_index:06d}.png"
|
|
path = image_root / name
|
|
Image.new("RGB", (800, 600), (sequence * 30, 10, 20)).save(path)
|
|
digest = hashlib.sha256(path.read_bytes()).hexdigest()
|
|
relative = f"images/valid-fov-fill/{name}"
|
|
reference = {
|
|
"schema_version": "missioncore.e46-truth-island-image-reference/v1",
|
|
"truth_island_sequence": sequence,
|
|
"image_id": image_id,
|
|
"frame_index": frame_index,
|
|
"session_seconds": float(sequence),
|
|
"role": "anchor",
|
|
"group_id": f"anchor-{sequence}",
|
|
"source_path": relative,
|
|
"sha256": digest,
|
|
"byte_length": path.stat().st_size,
|
|
}
|
|
references.append(reference)
|
|
review_images.append(
|
|
{
|
|
"truth_island_sequence": sequence,
|
|
"image_id": image_id,
|
|
"frame_index": frame_index,
|
|
"session_seconds": float(sequence),
|
|
"role": "anchor",
|
|
"group_id": f"anchor-{sequence}",
|
|
"source_path": relative,
|
|
"source_sha256": digest,
|
|
"review_state": "pending",
|
|
"hard_negative": None,
|
|
"objects": [],
|
|
"notes": None,
|
|
}
|
|
)
|
|
_write_json(
|
|
evaluation_root / "manifest.json",
|
|
{"generation_id": evaluation_root.name},
|
|
)
|
|
(truth_root / "image-references.jsonl").write_bytes(
|
|
b"".join(_canonical(row) + b"\n" for row in references)
|
|
)
|
|
_write_json(
|
|
truth_root / "blind-contract.json",
|
|
{
|
|
"annotation": {"classes": ["person", "car"]},
|
|
"reviewer_package": {
|
|
"model_predictions_included": False,
|
|
"model_prelabels_included": False,
|
|
},
|
|
},
|
|
)
|
|
_write_json(
|
|
truth_root / "review-template.json",
|
|
{
|
|
"schema_version": "missioncore.e46-detector-review-template/v1",
|
|
"truth_island_id": truth_root.name,
|
|
"state": "prepared-unreviewed-no-prelabels",
|
|
"reviewer_id": None,
|
|
"review_round": None,
|
|
"images": review_images,
|
|
"acceptance": None,
|
|
},
|
|
)
|
|
manifest = {
|
|
"identity": {
|
|
"source": {"evaluation_pack_id": evaluation_root.name},
|
|
}
|
|
}
|
|
_write_json(truth_root / "manifest.json", manifest)
|
|
truth = SimpleNamespace(
|
|
result_id=truth_root.name,
|
|
result_root=truth_root,
|
|
manifest=manifest,
|
|
report={"status": "prepared-awaiting-independent-human-review"},
|
|
)
|
|
monkeypatch.setattr(
|
|
handoff,
|
|
"read_e46_detector_truth_island",
|
|
lambda _root: truth,
|
|
)
|
|
monkeypatch.setattr(
|
|
e48,
|
|
"read_e46_detector_truth_island",
|
|
lambda _root: truth,
|
|
)
|
|
return truth_root, evaluation_root, truth
|
|
|
|
|
|
def test_handoff_is_blind_reproducible_and_converts_cvat(
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
truth_root, evaluation_root, _truth = _fixture(tmp_path, monkeypatch)
|
|
result = handoff.build_e46_review_handoff(
|
|
truth_island_root=truth_root,
|
|
evaluation_pack_root=evaluation_root,
|
|
output_root=tmp_path / "handoffs",
|
|
)
|
|
repeated = handoff.build_e46_review_handoff(
|
|
truth_island_root=truth_root,
|
|
evaluation_pack_root=evaluation_root,
|
|
output_root=tmp_path / "handoffs",
|
|
)
|
|
|
|
assert repeated.result_id == result.result_id
|
|
assert result.manifest["identity"]["frame_count"] == 2
|
|
with zipfile.ZipFile(
|
|
result.result_root / "reviewer-a" / "cvat-empty-coco.zip"
|
|
) as archive:
|
|
coco = json.loads(archive.read("annotations/instances_default.json"))
|
|
assert coco["annotations"] == []
|
|
assert "predictions" not in json.dumps(coco).lower()
|
|
|
|
source_rows = [
|
|
json.loads(line)
|
|
for line in (
|
|
result.result_root / "reviewer-a" / "source-map.jsonl"
|
|
).read_text().splitlines()
|
|
]
|
|
cvat = {
|
|
"images": [
|
|
{
|
|
"id": index,
|
|
"file_name": row["file_name"],
|
|
"width": 800,
|
|
"height": 600,
|
|
}
|
|
for index, row in enumerate(source_rows, start=11)
|
|
],
|
|
"categories": [
|
|
{"id": 1, "name": "person"},
|
|
{"id": 2, "name": "car"},
|
|
],
|
|
"annotations": [
|
|
{
|
|
"id": 1,
|
|
"image_id": 11,
|
|
"category_id": 2,
|
|
"bbox": [10.0, 20.0, 100.0, 200.0],
|
|
"attributes": {
|
|
"occluded": False,
|
|
"truncated": True,
|
|
},
|
|
}
|
|
],
|
|
}
|
|
cvat_path = tmp_path / "review-a.json"
|
|
_write_json(cvat_path, cvat)
|
|
output_path = tmp_path / "e48-review-a.json"
|
|
document = handoff.convert_cvat_coco_to_e48_review(
|
|
handoff_root=result.result_root,
|
|
truth_island_root=truth_root,
|
|
reviewer_slot="reviewer-a",
|
|
reviewer_id="human-reviewer-a",
|
|
cvat_export_path=cvat_path,
|
|
submitted_at_utc="2026-07-29T20:00:00Z",
|
|
output_path=output_path,
|
|
)
|
|
|
|
assert document["state"] == "completed-independent-no-model-assistance"
|
|
assert document["images"][0]["objects"][0]["box_xyxy"] == [
|
|
10.0,
|
|
20.0,
|
|
110.0,
|
|
220.0,
|
|
]
|
|
assert document["images"][1]["hard_negative"] is True
|
|
assert e48.validate_e48_detector_review_submission(
|
|
truth_island_root=truth_root,
|
|
review_path=output_path,
|
|
)["reviewer_id"] == "human-reviewer-a"
|
|
|
|
|
|
def test_handoff_rejects_changed_source_image(
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
truth_root, evaluation_root, _truth = _fixture(tmp_path, monkeypatch)
|
|
source = next(
|
|
(evaluation_root / "images" / "valid-fov-fill").glob("*.png")
|
|
)
|
|
source.write_bytes(b"changed")
|
|
|
|
with pytest.raises(handoff.E46ReviewHandoffError, match="identity changed"):
|
|
handoff.build_e46_review_handoff(
|
|
truth_island_root=truth_root,
|
|
evaluation_pack_root=evaluation_root,
|
|
output_root=tmp_path / "handoffs",
|
|
)
|
|
|
|
|
|
def test_converter_requires_explicit_occlusion_and_truncation(
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
truth_root, evaluation_root, _truth = _fixture(tmp_path, monkeypatch)
|
|
result = handoff.build_e46_review_handoff(
|
|
truth_island_root=truth_root,
|
|
evaluation_pack_root=evaluation_root,
|
|
output_root=tmp_path / "handoffs",
|
|
)
|
|
source_rows = [
|
|
json.loads(line)
|
|
for line in (
|
|
result.result_root / "reviewer-a" / "source-map.jsonl"
|
|
).read_text().splitlines()
|
|
]
|
|
cvat = {
|
|
"images": [
|
|
{
|
|
"id": index,
|
|
"file_name": row["file_name"],
|
|
"width": 800,
|
|
"height": 600,
|
|
}
|
|
for index, row in enumerate(source_rows, start=1)
|
|
],
|
|
"categories": [{"id": 1, "name": "car"}],
|
|
"annotations": [
|
|
{
|
|
"id": 1,
|
|
"image_id": 1,
|
|
"category_id": 1,
|
|
"bbox": [10.0, 20.0, 100.0, 200.0],
|
|
}
|
|
],
|
|
}
|
|
cvat_path = tmp_path / "invalid.json"
|
|
_write_json(cvat_path, cvat)
|
|
|
|
with pytest.raises(
|
|
handoff.E46ReviewHandoffError,
|
|
match="occluded flag is missing",
|
|
):
|
|
handoff.convert_cvat_coco_to_e48_review(
|
|
handoff_root=result.result_root,
|
|
truth_island_root=truth_root,
|
|
reviewer_slot="reviewer-a",
|
|
reviewer_id="human-reviewer-a",
|
|
cvat_export_path=cvat_path,
|
|
submitted_at_utc="2026-07-29T20:00:00Z",
|
|
output_path=tmp_path / "invalid-review.json",
|
|
)
|