NODEDC_MISSION_CORE/tests/test_e31_source_qualification.py

from __future__ import annotations

import hashlib
from pathlib import Path
from typing import Any

import numpy as np
import pytest

from k1link.compute.e31_source_qualification import (
    E31SourceQualificationError,
    E31SourceQualificationProfile,
    _E30Chain,
    _offset_sweep,
    _self_mask_report,
)
from k1link.compute.lidar_local_surface import POINT_OCCUPIED


class _SweepSource:
    def __init__(self) -> None:
        self.arrays = {
            "session_seconds": np.asarray([0.0, 0.1, 0.2], dtype=np.float64),
            "sample_available": np.ones(3, dtype=np.bool_),
            "cloud_offsets": np.asarray([0, 2, 4, 6], dtype=np.int64),
            "cloud_points_map": np.asarray(
                [
                    [1.0, 0.0, 2.0],
                    [1.1, 0.0, 2.0],
                    [0.0, 0.0, 2.0],
                    [0.1, 0.0, 2.0],
                    [-1.0, 0.0, 2.0],
                    [-1.1, 0.0, 2.0],
                ],
                dtype=np.float32,
            ),
            "pose_positions_map": np.zeros((3, 3), dtype=np.float64),
            "pose_quaternions_map_from_lidar": np.asarray(
                [[0.0, 0.0, 0.0, 1.0]] * 3,
                dtype=np.float64,
            ),
            "intrinsic_fx_fy_cx_cy": np.asarray(
                [100.0, 100.0, 50.0, 50.0],
                dtype=np.float64,
            ),
            "distortion_kb4": np.zeros(4, dtype=np.float64),
            "t_camera_from_lidar": np.eye(4, dtype=np.float64),
        }
        self.identity: dict[str, Any] = {
            "source_id": "sensor.camera.right",
            "camera_slot": "camera_1",
            "projection": {"width": 100, "height": 100},
        }


class _SweepSurface:
    def __init__(self) -> None:
        self.arrays = {
            "point_class": np.full(6, POINT_OCCUPIED, dtype=np.uint8),
        }


def _artifact(path: Path) -> dict[str, object]:
    return {
        "path": path.name,
        "byte_length": path.stat().st_size,
        "sha256": hashlib.sha256(path.read_bytes()).hexdigest(),
    }


def _projection_artifact(
    root: Path,
    name: str,
    pixels: list[list[float]],
) -> dict[str, object]:
    path = root / f"{name}.npz"
    np.savez(
        path,
        projected_pixels_xy=np.asarray(pixels, dtype=np.float32),
        projected_selected_mask=np.ones(len(pixels), dtype=np.uint8),
    )
    return _artifact(path)


def test_profile_rejects_ambiguous_offset_hypotheses() -> None:
    with pytest.raises(E31SourceQualificationError):
        E31SourceQualificationProfile(offset_hypotheses_ms=(0, -50, 50))
    with pytest.raises(E31SourceQualificationError):
        E31SourceQualificationProfile(offset_hypotheses_ms=(-50, 50))


def test_offset_sweep_keeps_evidenced_zero_binding() -> None:
    source = _SweepSource()
    surface = _SweepSurface()
    profile = E31SourceQualificationProfile(
        offset_hypotheses_ms=(-100, 0, 100),
        minimum_correspondence_items=1,
    )
    item = {
        "item_id": "item-1",
        "review_key": "semantic:1:1",
        "evidence_binding": {"frame_index": 1},
        "e29_snapshot": {"bbox_xyxy": [40.0, 40.0, 60.0, 60.0]},
    }

    sweep, rows = _offset_sweep(
        source=source,  # type: ignore[arg-type]
        surface=surface,  # type: ignore[arg-type]
        items=(item,),
        profile=profile,
        bbox_inset_fraction=0.03,
    )

    assert sweep["selected_offset_ms"] == 0
    assert sweep["baseline_supported_fraction"] == 1.0
    assert sweep["best_supported_fraction"] == 1.0
    assert sweep["baseline_support_deficit_fraction"] == 0.0
    assert [row["supported_count"] for row in sweep["hypotheses"]] == [0, 1, 0]
    assert [score["candidate_frame_index"] for score in rows[0]["scores"]] == [
        0,
        1,
        2,
    ]


def test_self_mask_admits_only_non_colliding_semantic_rule(
    tmp_path: Path,
) -> None:
    items: list[dict[str, Any]] = []
    decisions: list[dict[str, Any]] = []
    for index in range(4):
        item = {
            "item_id": f"semantic-self-{index}",
            "e29_snapshot": {
                "label": "person",
                "bbox_xyxy": [
                    20.0 + index,
                    75.0,
                    40.0 + index,
                    99.0,
                ],
            },
        }
        items.append(item)
        decisions.append(
            {
                "item_id": item["item_id"],
                "cause_code": "self_points",
                "point_ownership": "self",
            }
        )

    geometry_self = {
        "item_id": "geometry-self",
        "e29_snapshot": {},
        "artifact": _projection_artifact(
            tmp_path,
            "geometry-self",
            [[20.0, 20.0], [30.0, 30.0]],
        ),
    }
    accepted_object = {
        "item_id": "accepted-object",
        "e29_snapshot": {},
        "artifact": _projection_artifact(
            tmp_path,
            "accepted-object",
            [[25.0, 25.0], [80.0, 80.0]],
        ),
    }
    accepted_person = {
        "item_id": "accepted-person",
        "e29_snapshot": {
            "label": "person",
            "bbox_xyxy": [70.0, 50.0, 90.0, 90.0],
        },
        "artifact": _projection_artifact(
            tmp_path,
            "accepted-person",
            [[80.0, 80.0]],
        ),
    }
    items.extend([geometry_self, accepted_object, accepted_person])
    decisions.extend(
        [
            {
                "item_id": geometry_self["item_id"],
                "cause_code": "self_points",
                "point_ownership": "self",
            },
            {
                "item_id": accepted_object["item_id"],
                "cause_code": "none",
                "point_ownership": "object",
            },
            {
                "item_id": accepted_person["item_id"],
                "cause_code": "none",
                "point_ownership": "object",
            },
        ]
    )
    chain = _E30Chain(
        materialization_manifest={
            "identity": {
                "projection": {
                    "width": 100,
                    "height": 100,
                }
            }
        },
        items=tuple(items),
        engineering_manifest={},
        decisions=tuple(decisions),
        exceptions=(),
        human_manifest={},
        human_decisions=(),
    )

    report = _self_mask_report(
        materialization_root=tmp_path,
        chain=chain,
        profile=E31SourceQualificationProfile(
            minimum_semantic_self_samples=4,
        ),
    )

    assert report["semantic_mask"]["status"] == "admitted"
    assert report["semantic_mask"]["application_rule"] == "bbox-center-inside-rectangle"
    assert report["semantic_mask"]["collateral_item_count"] == 0
    assert report["geometry_point_mask"]["status"] == "rejected"
    assert report["geometry_point_mask"]["collateral"] == [
        {
            "item_id": "accepted-object",
            "masked_selected_point_count": 1,
        }
    ]
    assert report["exact_correction_item_ids"] == ["geometry-self"]