Merge pull request #34 from dbinfrago/feat/annotation-type-check

feat: add annotation type-per-sensor check

Author: crorry-dev

raillabel_providerkit/validation/issue.py

@@ -28,6 +28,7 @@ class IssueType(Enum):
     SENSOR_TYPE_WRONG = "SensorTypeWrong"
     UNEXPECTED_CLASS = "UnexpectedClassIssue"
     URI_FORMAT = "UriFormatIssue"
+    ANNOTATION_SENSOR_MISMATCH = "AnnotationSensorMismatch"
 
     @classmethod
     def names(cls) -> list[str]:

raillabel_providerkit/validation/validate.py

@@ -22,6 +22,9 @@
     validate_sensors,
     validate_uris,
 )
+from .validate_annotation_type_per_sensor.validate_annotation_type_per_sensor import (
+    validate_annotation_type_per_sensor,
+)
 
 
 def validate(  # noqa: C901, PLR0913
@@ -34,6 +37,7 @@ def validate(  # noqa: C901, PLR0913
     validate_for_uris: bool = True,
     validate_for_dimensions: bool = True,
     validate_for_horizon: bool = True,
+    validate_for_annotation_type_per_sensor: bool = True,
 ) -> list[Issue]:
     """Validate a scene based on the Deutsche Bahn Requirements.
 
@@ -56,6 +60,8 @@ def validate(  # noqa: C901, PLR0913
         validate_for_dimensions: If True, issues are returned if the dimensions of cuboids are
             outside the expected values range.
         validate_for_horizon: If True, issues are returned if annotations cross the horizon.
+        validate_for_annotation_type_per_sensor: Validate that annotation types match sensor types.
+
 
     Returns:
         List of all requirement errors in the scene. If an empty list is returned, then there are no
@@ -96,4 +102,7 @@ def validate(  # noqa: C901, PLR0913
     if validate_for_horizon:
         errors.extend(validate_horizon(scene))
 
+    if validate_for_annotation_type_per_sensor:
+        errors.extend(validate_annotation_type_per_sensor(scene))
+
     return errors

raillabel_providerkit/validation/validate_annotation_type_per_sensor/__init__.py

@@ -0,0 +1,8 @@
+# Copyright DB InfraGO AG and contributors
+# SPDX-License-Identifier: Apache-2.0
+
+"""Validation for annotation type per sensor."""
+
+from .validate_annotation_type_per_sensor import validate_annotation_type_per_sensor
+
+__all__ = ["validate_annotation_type_per_sensor"]

raillabel_providerkit/validation/validate_annotation_type_per_sensor/validate_annotation_type_per_sensor.py

@@ -0,0 +1,133 @@
+# Copyright DB InfraGO AG and contributors
+# SPDX-License-Identifier: Apache-2.0
+
+from __future__ import annotations
+
+from collections.abc import Iterable
+from typing import Any
+from uuid import UUID
+
+import raillabel
+
+from raillabel_providerkit.validation.issue import Issue, IssueIdentifiers, IssueType
+
+# Mapping:
+# - camera: Bbox
+# - lidar: Cuboid
+# - radar: und Cuboid
+_ALLOWED_BY_SENSOR_TYPE: dict[str, tuple[str, ...]] = {
+    "camera": ("Bbox",),
+    "lidar": ("Cuboid",),
+    "radar": ("Bbox", "Cuboid"),
+}
+
+
+def _normalize_anno_type(t: str) -> str:
+    t_low = (t or "").strip().lower()
+    mapping = {
+        "bbox": "Bbox",
+        "cuboid": "Cuboid",
+        "poly2d": "Poly2d",
+        "poly3d": "Poly3d",
+        "num": "Num",
+        "seg3d": "Seg3d",
+    }
+    return mapping.get(t_low, t if t else "")
+
+
+def _iter_annotations(scene: raillabel.Scene) -> Iterable[tuple[int, Any]]:
+    """Iterate over all annotations in the scene."""
+    frames = getattr(scene, "frames", {}) or {}
+    for frame_id, frame in getattr(frames, "items", lambda: frames.items())():
+        annotations = getattr(frame, "annotations", {}) or {}
+        for anno in annotations.values():
+            yield frame_id, anno
+
+
+def _get_annotation_id(anno: object) -> UUID | None:
+    return getattr(anno, "id", None)
+
+
+def _get_annotation_type(anno: object) -> str | None:
+    t = getattr(anno, "type", None)
+    if not t:
+        return None
+    return _normalize_anno_type(str(t))
+
+
+def _get_sensor_id_from_annotation(anno: object) -> str | None:
+    # 1) coordinate.sensor / coordinate.sensor_id
+    coord = getattr(anno, "coordinate", None)
+    if coord is not None:
+        sid = getattr(coord, "sensor", None) or getattr(coord, "sensor_id", None)
+        if sid:
+            return str(sid)
+
+    # 2) anno.sensor / anno.sensor_id
+    for attr in ("sensor", "sensor_id"):
+        sid = getattr(anno, attr, None)
+        if sid:
+            return str(sid)
+
+    # 3) coordinates-Collection
+    coords = getattr(anno, "coordinates", None)
+    if isinstance(coords, list | tuple):
+        for c in coords:
+            sid = getattr(c, "sensor", None) or getattr(c, "sensor_id", None)
+            if sid:
+                return str(sid)
+
+    return None
+
+
+def _sensor_type(scene: raillabel.Scene, sensor_id: str) -> str | None:
+    """Return the sensor type ('camera', 'lidar', 'radar') or None if unknown."""
+    sensors = getattr(scene, "sensors", {}) or {}
+    sensor = sensors.get(sensor_id)
+    if sensor is None and hasattr(sensors, "get"):
+        sensor = sensors.get(sensor_id)
+    if sensor is None:
+        return None
+    s_type = getattr(sensor, "type", None)
+    return str(s_type).lower() if s_type else None
+
+
+def validate_annotation_type_per_sensor(scene: raillabel.Scene) -> list[Issue]:
+    """Validate that each annotation type is compatible with its sensor type.
+
+    Returns:
+        List of Issues if mismatches are found.
+    """
+    issues: list[Issue] = []
+
+    for frame_id, anno in _iter_annotations(scene):
+        a_type = _get_annotation_type(anno)
+        if not a_type:
+            continue
+
+        sensor_id = _get_sensor_id_from_annotation(anno)
+        if not sensor_id:
+            continue
+
+        s_type = _sensor_type(scene, sensor_id)
+        if not s_type:
+            continue
+
+        allowed = _ALLOWED_BY_SENSOR_TYPE.get(s_type, ())
+        if allowed and a_type not in allowed:
+            issue = Issue(
+                type=IssueType.ANNOTATION_SENSOR_MISMATCH,
+                identifiers=IssueIdentifiers(
+                    frame=frame_id,
+                    sensor=sensor_id,
+                    annotation=_get_annotation_id(anno),
+                    annotation_type=a_type,
+                ),
+                reason=(
+                    f"Annotation type '{a_type}' not allowed for sensor type '{s_type}'. "
+                    f"Allowed types: {', '.join(allowed) if allowed else 'n/a'}."
+                ),
+            )
+            issues.append(issue)
+
+    return issues

tests/validation/validate_annotation_type_per_sensor/test_validate_annotation_type_per_sensor.py

@@ -0,0 +1,111 @@
+# Copyright DB InfraGO AG and contributors
+# SPDX-License-Identifier: Apache-2.0
+
+from __future__ import annotations
+from dataclasses import dataclass
+from typing import Dict, Optional, List
+from uuid import uuid4
+
+import pytest
+
+from raillabel_providerkit.validation.issue import IssueType
+from raillabel_providerkit.validation.validate_annotation_type_per_sensor import (
+    validate_annotation_type_per_sensor,
+)
+
+
+@dataclass
+class _Coord:
+    sensor: Optional[str] = None
+    sensor_id: Optional[str] = None
+
+
+@dataclass
+class _Annotation:
+    id: str
+    type: str
+    coordinate: Optional[_Coord] = None
+
+
+@dataclass
+class _Frame:
+    annotations: Dict[str, _Annotation]
+
+
+@dataclass
+class _Sensor:
+    id: str
+    type: str
+
+
+@dataclass
+class _Scene:
+    sensors: Dict[str, _Sensor]
+    frames: Dict[int, _Frame]
+
+
+# ----------------------------------------------------------------
+
+
+def _mk_bbox(sensor_id: str) -> _Annotation:
+    return _Annotation(id=str(uuid4()), type="Bbox", coordinate=_Coord(sensor=sensor_id))
+
+
+def _mk_cuboid(sensor_id: str) -> _Annotation:
+    return _Annotation(id=str(uuid4()), type="Cuboid", coordinate=_Coord(sensor=sensor_id))
+
+
+def test_camera_allows_only_bbox():
+    scene = _Scene(
+        sensors={"cam0": _Sensor(id="cam0", type="camera")},
+        frames={0: _Frame(annotations={"a": _mk_bbox("cam0")})},
+    )
+    issues = validate_annotation_type_per_sensor(scene)
+    assert issues == []
+
+    scene_bad = _Scene(
+        sensors={"cam0": _Sensor(id="cam0", type="camera")},
+        frames={0: _Frame(annotations={"a": _mk_cuboid("cam0")})},
+    )
+    issues = validate_annotation_type_per_sensor(scene_bad)
+    assert len(issues) == 1
+    assert issues[0].type == IssueType.ANNOTATION_SENSOR_MISMATCH
+    assert issues[0].identifiers.annotation_type == "Cuboid"
+    assert issues[0].identifiers.sensor == "cam0"
+    assert issues[0].identifiers.frame == 0
+
+
+def test_lidar_allows_only_cuboid():
+    scene = _Scene(
+        sensors={"lid0": _Sensor(id="lid0", type="lidar")},
+        frames={1: _Frame(annotations={"a": _mk_cuboid("lid0")})},
+    )
+    assert validate_annotation_type_per_sensor(scene) == []
+
+    scene_bad = _Scene(
+        sensors={"lid0": _Sensor(id="lid0", type="lidar")},
+        frames={1: _Frame(annotations={"a": _mk_bbox("lid0")})},
+    )
+    issues = validate_annotation_type_per_sensor(scene_bad)
+    assert len(issues) == 1
+    assert issues[0].type == IssueType.ANNOTATION_SENSOR_MISMATCH
+    assert issues[0].identifiers.annotation_type == "Bbox"
+    assert issues[0].identifiers.sensor == "lid0"
+    assert issues[0].identifiers.frame == 1
+
+
+def test_radar_allows_bbox_and_cuboid():
+    scene = _Scene(
+        sensors={"rad0": _Sensor(id="rad0", type="radar")},
+        frames={2: _Frame(annotations={"a": _mk_bbox("rad0"), "b": _mk_cuboid("rad0")})},
+    )
+    assert validate_annotation_type_per_sensor(scene) == []
+
+
+def test_missing_sensor_binding_is_ignored():
+    ann = _Annotation(id=str(uuid4()), type="Bbox", coordinate=None)  # kein Sensor
+    scene = _Scene(
+        sensors={"cam0": _Sensor(id="cam0", type="camera")},
+        frames={3: _Frame(annotations={"a": ann})},
+    )
+    assert validate_annotation_type_per_sensor(scene) == []