allow osi lanes to only have centerlines

Author: MichaelSchuldes

README.md

@@ -1,12 +1,20 @@
-<img src="https://www.ika.rwth-aachen.de/images/ika-logo-a-blau-blau-rgb.svg" align="right" width="240"> 
+
+<img src="https://www.ika.rwth-aachen.de/images/ika-logo-a-blau-blau-rgb.svg" align="right" width="240">
 </br>
 </br>
 
+[![](https://img.shields.io/badge/license-MPL%202.0-blue.svg)](https://github.com/ika-rwth-aachen/omega-prime/blob/master/LICENSE) 
+[![](https://img.shields.io/pypi/v/omega-prime.svg)](https://pypi.python.org/pypi/omega-prime)
+[![](https://github.com/ika-rwth-aachen/omega-prime/workflows/CI/badge.svg)](https://github.com/ika-rwth-aachen/omega-prime/actions)
+[![](https://img.shields.io/pypi/pyversions/omega-prime.svg)](https://pypi.python.org/pypi/omega-prime/)
+[![](https://img.shields.io/github/issues-raw/ika-rwth-aachen/omega-prime.svg)](https://github.com/ika-rwth-aachen/omega-prime/issues)
+
 > [!IMPORTANT]  
 > The data model and specification are not finalized and are still under discussion. See this repository as a proof of concept.
 
 
 # Omega-Prime: Data Model, Data Format and Python Library for Handling Ground Truth Traffic Data 
+
 Data Model, Format and Python Library for ground truth data containing information on dynamic objects, map and environmental factors optimized for representing urban traffic. The repository contains:
 - **Sepcification Document:** to be released
     - **Data Model**: What signals exists and how these are defined.

omega_prime/asam_odr/opendriveconverter/converter.py

@@ -1,17 +1,8 @@
-from dataclasses import dataclass
-from typing import Any
-
 from ..logger import logger
 from .elements.junction import setup_connections
 from .elements.road import process_road
 
 
-@dataclass(repr=False)
-class Map:
-    roads: dict
-    georeference: Any = None
-
-
 def convert_opendrive(my_opendrive, step_size=0.1) -> tuple[dict, tuple[float, float, str]]:
     """
     Converts the opendrive XML input into a OMEGAFormat road network

omega_prime/map.py

@@ -53,14 +53,19 @@ def get_osi(self) -> betterosi.LaneBoundary:
 
 
 @dataclass(repr=False)
-class Lane:
+class LaneBase:
     _map: "Map" = field(init=False)
     idx: Any
     centerline: np.ndarray
     type: betterosi.LaneClassificationType
     subtype: betterosi.LaneClassificationSubtype
     successor_ids: list[Any]
     predecessor_ids: list[Any]
+    source_reference: Any = field(init=False)
+
+
+@dataclass(repr=False)
+class Lane(LaneBase):
     right_boundary_id: Any
     left_boundary_id: Any
     polygon: shapely.Polygon = field(init=False)
@@ -70,11 +75,8 @@ class Lane:
 
 
 @dataclass(repr=False)
-class LaneOsi(Lane):
+class LaneOsiCenterline(LaneBase):
     _osi: betterosi.Lane
-    right_boundary_ids: list[int]
-    left_boundary_ids: list[int]
-    free_boundary_ids: list[int]
 
     @staticmethod
     def _get_centerline(lane: betterosi.Lane):
@@ -83,6 +85,31 @@ def _get_centerline(lane: betterosi.Lane):
             cl = np.flip(cl, axis=0)
         return cl
 
+    @classmethod
+    def create(cls, lane: betterosi.Lane):
+        return cls(
+            _osi=lane,
+            idx=lane.id.value,
+            centerline=cls._get_centerline(lane),
+            type=betterosi.LaneClassificationType(lane.classification.type),
+            subtype=betterosi.LaneClassificationSubtype(lane.classification.subtype),
+            successor_ids=[],
+            predecessor_ids=[],
+            right_boundary_id=None,
+            left_boundary_id=None,
+        )
+
+    def plot(self, ax: plt.Axes):
+        c = "black" if not self.type == betterosi.LaneClassificationType.TYPE_INTERSECTION else "green"
+        ax.plot(*np.array(self.centerline).T, color=c, alpha=0.3, zorder=-10)
+
+
+@dataclass(repr=False)
+class LaneOsi(Lane, LaneOsiCenterline):
+    right_boundary_ids: list[int]
+    left_boundary_ids: list[int]
+    free_boundary_ids: list[int]
+
     @classmethod
     def create(cls, lane: betterosi.Lane):
         return cls(
@@ -133,7 +160,6 @@ def plot(self, ax: plt.Axes):
         ax.add_patch(PltPolygon(self.polygon.exterior.coords, fc="blue", alpha=0.2, ec="black"))
 
     # for ase_engine/omega_prime
-
     def _get_oriented_borders(self):
         center_start = shapely.LineString(self.centerline).interpolate(0, normalized=True)
         left = self.left_boundary.polyline
@@ -175,9 +201,8 @@ class MapOsi(Map):
 
     @classmethod
     def create(cls, gt: betterosi.GroundTruth):
-        if not hasattr(gt, "lane") or not hasattr(gt, "lane_boundary"):
-            return None
-
+        if len(gt.lane_boundary) == 0:
+            raise RuntimeError("Empty Map")
         return cls(
             _osi=gt,
             lane_boundaries={b.id.value: LaneBoundaryOsi.create(b) for b in gt.lane_boundary},
@@ -194,3 +219,23 @@ def setup_lanes_and_boundaries(self):
             l._map = self
             l.set_boundaries()
             l.set_polygon()
+
+
+@dataclass(repr=False)
+class MapOsiCenterline(Map):
+    _osi: betterosi.GroundTruth
+    lanes: dict[int, LaneOsiCenterline]
+
+    @classmethod
+    def create(cls, gt: betterosi.GroundTruth):
+        if len(gt.lane) == 0:
+            raise RuntimeError("No Map")
+        return cls(
+            _osi=gt,
+            lanes={l.id.value: LaneOsiCenterline.create(l) for l in gt.lane},
+            lane_boundaries={},
+        )
+
+    def setup_lanes_and_boundaries(self):
+        for l in self.lanes.values():
+            l._map = self

omega_prime/recording.py

@@ -14,7 +14,8 @@
 import pandera.extensions as extensions
 
 from .asam_odr import MapOdr
-from .map import MapOsi, ProjectionOffset
+from .map import MapOsi, ProjectionOffset, MapOsiCenterline
+import itertools
 
 pi_valued = pa.Check.between(-np.pi, np.pi)
 
@@ -292,15 +293,14 @@ def to_osi_gts(self) -> list[betterosi.GroundTruth]:
             for nanos, group_df in self._df.groupby("total_nanos")
         ]
 
-        if self.map is not None and isinstance(self.map, MapOsi):
+        if self.map is not None and isinstance(self.map, MapOsi | MapOsiCenterline):
             gts[0].lane_boundary = [b._osi for b in self.map.lane_boundaries.values()]
             gts[0].lane = [l._osi for l in self.map.lanes.values()]
         return gts
 
     @classmethod
     def from_osi_gts(cls, gts: list[betterosi.GroundTruth], validate: bool = True):
         mvs = []
-        map = None
         projs = []
         for i, gt in enumerate(gts):
             total_nanos = gt.timestamp.seconds * 1_000_000_000 + gt.timestamp.nanos
@@ -351,29 +351,37 @@ def from_osi_gts(cls, gts: list[betterosi.GroundTruth], validate: bool = True):
                 )
                 for mv in gt.moving_object
             ]
-
-            if map is None:
-                map = MapOsi.create(gt)
         df_mv = pd.DataFrame(mvs).sort_values(by=["total_nanos", "idx"]).reset_index(drop="index")
-        return cls(df_mv, map, projections=projs, validate=validate)
+        return cls(df_mv, projections=projs, validate=validate)
 
     @classmethod
     def from_file(cls, filepath, xodr_path: str | None = None, validate: bool = True, skip_odr_parse: bool = False):
         gts = betterosi.read(
             filepath,
             return_ground_truth=True,
         )
+        gts, tmp_gts = itertools.tee(gts, 2)
+        first_gt = next(tmp_gts)
         r = cls.from_osi_gts(gts, validate=validate)
         if xodr_path is not None:
             r.map = MapOdr.from_file(xodr_path, skip_parse=skip_odr_parse)
         elif Path(filepath).suffix == ".mcap":
             try:
                 r.map = MapOdr.from_file(filepath, skip_parse=skip_odr_parse)
             except StopIteration:
-                if r.map is not None:
+                pass
+        if r.map is None:
+            try:
+                r.map = MapOsi.create(first_gt)
+                warn("No map provided in mcap. OSI GroundTruth map is used!")
+            except RuntimeError:
+                try:
+                    r.map = MapOsiCenterline.create(first_gt)
                     warn("No map provided in mcap. OSI GroundTruth map is used!")
+                except RuntimeError:
+                    pass
         if r.map is None:
-            warn("No xodr map provided in mcap nor OSI map in GroundTruth!")
+            warn("No xodr map provided in MCAP nor OSI map in GroundTruth!")
         return r
 
     def to_mcap(self, filepath):

pyproject.toml

@@ -15,7 +15,6 @@ keywords = ["omega-format", "osi", "open-simulation-interface", "simulation", "a
 classifiers = [
     "License :: OSI Approved :: Mozilla Public License 2.0 (MPL 2.0)",
     "Programming Language :: Python :: 3",
-    "Programming Language :: Python :: 3.9",
     "Programming Language :: Python :: 3.10",
     "Programming Language :: Python :: 3.11",
     "Programming Language :: Python :: 3.12",