Implement advanced issue reporting

pyproject.toml

@@ -61,6 +61,7 @@ test = [
 
 [tool.ruff]
 line-length = 101
+target-version = "py310"
 
 [tool.ruff.lint]
 exclude = ["tests/*", "docs/*"]

raillabel_providerkit/validation/__init__.py

@@ -2,7 +2,8 @@
 # SPDX-License-Identifier: Apache-2.0
 """Package for validating raillabel data regarding the format requirements."""
 
+from .issue import Issue, IssueIdentifiers, IssueType
 from .validate_onthology.validate_onthology import validate_onthology
 from .validate_schema import validate_schema
 
-__all__ = ["validate_onthology", "validate_schema"]
+__all__ = ["Issue", "IssueIdentifiers", "IssueType", "validate_onthology", "validate_schema"]

raillabel_providerkit/validation/issue.py

@@ -0,0 +1,33 @@
+# Copyright DB InfraGO AG and contributors
+# SPDX-License-Identifier: Apache-2.0
+
+from dataclasses import dataclass
+from enum import Enum
+from uuid import UUID
+
+
+class IssueType(Enum):
+    """General classification of the issue."""
+
+    SCHEMA = "SchemaIssue"
+    EMPTY_FRAMES = "EmptyFramesIssue"
+    RAIL_SIDE = "RailSide"
+
+
+@dataclass
+class IssueIdentifiers:
+    """Information for locating an issue."""
+
+    annotation: UUID | None = None
+    frame: int | None = None
+    object: UUID | None = None
+    sensor: str | None = None
+
+
+@dataclass
+class Issue:
+    """An error that was found inside the scene."""
+
+    type: IssueType
+    reason: str
+    identifiers: IssueIdentifiers | list[str | int]

raillabel_providerkit/validation/validate.py

@@ -3,10 +3,12 @@
 
 from __future__ import annotations
 
+from raillabel_providerkit.validation import Issue
+
 from . import validate_schema
 
 
-def validate(scene_dict: dict) -> list[str]:
+def validate(scene_dict: dict) -> list[Issue]:
     """Validate a scene based on the Deutsche Bahn Requirements.
 
     Parameters
@@ -16,7 +18,7 @@ def validate(scene_dict: dict) -> list[str]:
 
     Returns
     -------
-    list[str]
+    list[Issue]
         list of all requirement errors in the scene. If an empty list is returned, then there are
         no errors present and the scene is valid.
 

raillabel_providerkit/validation/validate_empty_frames/validate_empty_frames.py

@@ -5,27 +5,25 @@
 
 import raillabel
 
+from raillabel_providerkit.validation import Issue, IssueIdentifiers, IssueType
 
-def validate_empty_frames(scene: raillabel.Scene) -> list[str]:
-    """Validate whether all frames of a scene have at least one annotation.
-
-    Parameters
-    ----------
-    scene : raillabel.Scene
-        Scene, that should be validated.
 
-    Returns
-    -------
-    list[str]
-        list of all empty frame errors in the scene. If an empty list is returned, then there are no
-        errors present.
+def validate_empty_frames(scene: raillabel.Scene) -> list[Issue]:
+    """Validate whether all frames of a scene have at least one annotation.
 
+    If an empty list is returned, then there are no errors present.
     """
-    errors: list[str] = []
+    errors = []
 
     for frame_uid, frame in scene.frames.items():
         if _is_frame_empty(frame):
-            errors.append("Frame " + str(frame_uid) + " has no annotations!")
+            errors.append(
+                Issue(
+                    type=IssueType.EMPTY_FRAMES,
+                    reason="This frame has no annotations.",
+                    identifiers=IssueIdentifiers(frame=frame_uid),
+                )
+            )
 
     return errors
 

raillabel_providerkit/validation/validate_rail_side/validate_rail_side.py

@@ -14,8 +14,10 @@
     IncludeSensorTypeFilter,
 )
 
+from raillabel_providerkit.validation import Issue, IssueIdentifiers, IssueType
 
-def validate_rail_side(scene: raillabel.Scene) -> list[str]:
+
+def validate_rail_side(scene: raillabel.Scene) -> list[Issue]:
     """Validate whether all tracks have <= one left and right rail, and that they have correct order.
 
     Parameters
@@ -30,7 +32,7 @@ def validate_rail_side(scene: raillabel.Scene) -> list[str]:
         errors present.
 
     """
-    errors: list[str] = []
+    errors = []
 
     camera_uids = list(scene.filter([IncludeSensorTypeFilter(["camera"])]).sensors.keys())
 
@@ -47,11 +49,11 @@ def validate_rail_side(scene: raillabel.Scene) -> list[str]:
             counts_per_track = _count_rails_per_track_in_frame(frame)
 
             for object_uid, (left_count, right_count) in counts_per_track.items():
-                context = {
-                    "frame_uid": frame_uid,
-                    "object_uid": object_uid,
-                    "camera_uid": camera_uid,
-                }
+                context = IssueIdentifiers(
+                    frame=frame_uid,
+                    sensor=camera_uid,
+                    object=object_uid,
+                )
 
                 count_errors = _check_rail_counts(context, left_count, right_count)
                 exactly_one_left_and_right_rail_exist = count_errors != []
@@ -64,33 +66,37 @@ def validate_rail_side(scene: raillabel.Scene) -> list[str]:
                 if left_rail is None or right_rail is None:
                     continue
 
-                errors.extend(
-                    _check_rails_for_swap_or_intersection(left_rail, right_rail, frame_uid)
-                )
+                errors.extend(_check_rails_for_swap_or_intersection(left_rail, right_rail, context))
 
     return errors
 
 
-def _check_rail_counts(context: dict, left_count: int, right_count: int) -> list[str]:
+def _check_rail_counts(context: IssueIdentifiers, left_count: int, right_count: int) -> list[Issue]:
     errors = []
     if left_count > 1:
         errors.append(
-            f"In sensor {context['camera_uid']} frame {context['frame_uid']}, the track with"
-            f" object_uid {context['object_uid']} has more than one ({left_count}) left rail."
+            Issue(
+                type=IssueType.RAIL_SIDE,
+                reason=f"This track has {left_count} left rails.",
+                identifiers=context,
+            )
         )
     if right_count > 1:
         errors.append(
-            f"In sensor {context['camera_uid']} frame {context['frame_uid']}, the track with"
-            f" object_uid {context['object_uid']} has more than one ({right_count}) right rail."
+            Issue(
+                type=IssueType.RAIL_SIDE,
+                reason=f"This track has {right_count} right rails.",
+                identifiers=context,
+            )
         )
     return errors
 
 
 def _check_rails_for_swap_or_intersection(
     left_rail: raillabel.format.Poly2d,
     right_rail: raillabel.format.Poly2d,
-    frame_uid: str | int = "unknown",
-) -> list[str]:
+    context: IssueIdentifiers,
+) -> list[Issue]:
     if left_rail.object_id != right_rail.object_id:
         return []
 
@@ -103,23 +109,22 @@ def _check_rails_for_swap_or_intersection(
     if left_x is None or right_x is None:
         return []
 
-    object_uid = left_rail.object_id
-    sensor_uid = left_rail.sensor_id if left_rail.sensor_id is not None else "unknown"
-
     if left_x >= right_x:
         return [
-            f"In sensor {sensor_uid} frame {frame_uid}, the track with"
-            f" object_uid {object_uid} has its rails swapped."
-            f" At the maximum common y={max_common_y}, the left rail has x={left_x}"
-            f" while the right rail has x={right_x}."
+            Issue(
+                type=IssueType.RAIL_SIDE,
+                reason="The left and right rails of this track are swapped.",
+                identifiers=context,
+            )
         ]
 
-    intersect_interval = _find_intersect_interval(left_rail, right_rail)
-    if intersect_interval is not None:
+    if _polylines_are_intersecting(left_rail, right_rail):
         return [
-            f"In sensor {sensor_uid} frame {frame_uid}, the track with"
-            f" object_uid {object_uid} intersects with itself."
-            f" The left and right rail intersect in y interval {intersect_interval}."
+            Issue(
+                type=IssueType.RAIL_SIDE,
+                reason="The left and right rails of this track intersect.",
+                identifiers=context,
+            )
         ]
 
     return []
@@ -162,9 +167,9 @@ def _filter_for_poly2ds(
     ]
 
 
-def _find_intersect_interval(
+def _polylines_are_intersecting(
     line1: raillabel.format.Poly2d, line2: raillabel.format.Poly2d
-) -> tuple[float, float] | None:
+) -> bool:
     """If the two polylines intersect anywhere, return the y interval where they intersect."""
     y_values_with_points_in_either_polyline: list[float] = sorted(
         _get_y_of_all_points_of_poly2d(line1).union(_get_y_of_all_points_of_poly2d(line2))
@@ -181,18 +186,18 @@ def _find_intersect_interval(
             continue
 
         if x1 == x2:
-            return (y, y)
+            return True
 
         new_order = x1 < x2
 
         order_has_flipped = order is not None and new_order != order and last_y is not None
         if order_has_flipped:
-            return (last_y, y)  # type: ignore  # noqa: PGH003
+            return True
 
         order = new_order
         last_y = y
 
-    return None
+    return False
 
 
 def _find_max_y(poly2d: raillabel.format.Poly2d) -> float: