v0.2.0

Author: MichaelSchuldes

omega_prime/__init__.py

@@ -42,5 +42,5 @@
     "MovingObjectSubtype",
     "MovingObjectRole",
     "MapSegmentation",
-    "MapOsiCenterlineSegmentation"
+    "MapOsiCenterlineSegmentation",
 ]

omega_prime/map.py

@@ -457,13 +457,13 @@ def map_to_centerline_mcap(self, output_mcap_path: Path = None) -> betterosi.Gro
         else:
             # Convert string to Path if needed
             output_mcap_path = Path(output_mcap_path)
-            
+
             if output_mcap_path.is_dir():
                 output_mcap_path = output_mcap_path / "map_to_centerline.mcap"
             elif not output_mcap_path.suffix == ".mcap":
                 logging.warning(f"Output path must be a directory or .mcap file: {output_mcap_path}")
                 return ground_truth
-            
+
             with betterosi.Writer(output_mcap_path) as writer:
                 writer.add(ground_truth, topic="ground_truth_map", log_time=0)
             logging.info(f"Successfully saved map with {len(osi_lanes)} lanes to {output_mcap_path}")

omega_prime/maposicenterlinesegmentation.py

@@ -19,10 +19,7 @@
 logger = logging.getLogger(__name__)
 
 
-
-
-
-def add_lanexy_to_graph(G : nx.Graph, lanes):
+def add_lanexy_to_graph(G: nx.Graph, lanes):
     """
     Adds lane coordinates to the graph as node attributes.
 
@@ -40,7 +37,7 @@ def add_lanexy_to_graph(G : nx.Graph, lanes):
     return G
 
 
-def plot_graph(G : nx.Graph, output : Path):
+def plot_graph(G: nx.Graph, output: Path):
     """
     Plots the graph with lane coordinates.
 
@@ -71,7 +68,7 @@ def _get_lane_geometry(self, lane) -> shapely.LineString:
     def set_trafficlight(self):
         trafficlight = []
         for lane in self.lanes:
-            if hasattr(lane, 'trafficlight') and lane.trafficlight:
+            if hasattr(lane, "trafficlight") and lane.trafficlight:
                 trafficlight.append(lane.trafficlight)
         self.trafficlights = trafficlight
 
@@ -90,7 +87,7 @@ def __init__(self, recording, lane_buffer=None, intersection_overlap_buffer=None
         self.lane_buffer = lane_buffer if lane_buffer is not None else 0.3
         self.intersection_overlap_buffer = intersection_overlap_buffer if intersection_overlap_buffer is not None else 1
         self.do_combine_intersections = True
-        
+
         for tl_state in recording.traffic_light_states.values():
             for tl in tl_state:
                 if tl.id.value not in self.trafficlight_ids:
@@ -108,17 +105,15 @@ def _get_lane_centerline(self, lane) -> shapely.LineString:
 
     def _get_lane_successors(self, lane) -> list:
         """Extract successor IDs from OSI centerline lane."""
-        return [succ_id.lane_id if hasattr(succ_id, "lane_id") else succ_id 
-                for succ_id in lane.successor_ids]
+        return [succ_id.lane_id if hasattr(succ_id, "lane_id") else succ_id for succ_id in lane.successor_ids]
 
     def _get_lane_predecessors(self, lane) -> list:
         """Extract predecessor IDs from OSI centerline lane."""
-        return [pred_id.lane_id if hasattr(pred_id, "lane_id") else pred_id 
-                for pred_id in lane.predecessor_ids]
+        return [pred_id.lane_id if hasattr(pred_id, "lane_id") else pred_id for pred_id in lane.predecessor_ids]
 
     def _has_traffic_light(self, lane) -> bool:
         """Check if OSI centerline lane has traffic light."""
-        return hasattr(lane, 'trafficlight') and lane.trafficlight is not None
+        return hasattr(lane, "trafficlight") and lane.trafficlight is not None
 
     def _get_traffic_light(self, lane):
         """Get traffic light object from OSI centerline lane."""
@@ -134,7 +129,7 @@ def _set_lane_is_approaching(self, lane, value: bool):
 
     def _get_lane_on_intersection(self, lane) -> bool:
         """Get the on_intersection status of OSI centerline lane."""
-        return lane.on_intersection if hasattr(lane, 'on_intersection') else False
+        return lane.on_intersection if hasattr(lane, "on_intersection") else False
 
     def init_intersections(self):
         """
@@ -523,7 +518,9 @@ def combine_intersection_on_polygon(self, intersection1, intersection2):
         """
         if self.intersections_overlap(intersection1, intersection2):
             # Create a new intersection object with the lanes from both intersections
-            combined_intersection = Intersection(intersection1.lanes + intersection2.lanes, concave_hull_ratio=self.concave_hull_ratio)
+            combined_intersection = Intersection(
+                intersection1.lanes + intersection2.lanes, concave_hull_ratio=self.concave_hull_ratio
+            )
 
             return combined_intersection
         else:
@@ -652,7 +649,11 @@ def find_isolated_connections(self):
         new_connections = []
         for component in nx.connected_components(G):
             if len(component) > 0:
-                isolated_connections.append(ConnectionSegment([self.lane_dict[i] for i in component], concave_hull_ratio=self.concave_hull_ratio))
+                isolated_connections.append(
+                    ConnectionSegment(
+                        [self.lane_dict[i] for i in component], concave_hull_ratio=self.concave_hull_ratio
+                    )
+                )
         # Check if any of the lanes in the isolated connections are part of an intersection
         for connection in isolated_connections:
             pre = False
@@ -909,7 +910,7 @@ def plot(
                     plt.savefig(output_path)
         plt.close()
 
-    def plot_intersections(self, output_plot : Path):
+    def plot_intersections(self, output_plot: Path):
         """
         Plots all intersections and saves them to the output path.
         Args:

omega_prime/mapsegment.py

@@ -13,8 +13,8 @@ class MapSegmentType(Enum):  # Added class for map segment from the scenario-dat
     RAMP_ON = "ramp_on"
     RAMP_OFF = "ramp_off"
     UNKNOWN = "unknown"
-    
-    
+
+
 class Segment(ABC):
     """A class that represents a segment of the map"""
 
@@ -52,7 +52,7 @@ def _compute_polygon_key(self):
 
     def _compute_segment_polygon(self):
         lane_centerline = [self._get_lane_geometry(lane) for lane in self.lanes]
-        multilinestring = shapely.MultiLineString(lane_centerline).buffer(.1)
+        multilinestring = shapely.MultiLineString(lane_centerline).buffer(0.1)
         combined = shapely.unary_union(multilinestring).buffer(0.1)
         try:
             hull = shapely.concave_hull(combined, self.concave_hull_ratio)
@@ -218,4 +218,4 @@ def check_if_all_lanes_are_on_segment(self):
             lane_id = self._get_lane_id(lane)
             if lane_id not in self.lane_segment_dict or self.lane_segment_dict[lane_id].segment is None:
                 return False
-        return True
+        return True

omega_prime/recording.py

@@ -350,7 +350,7 @@ def from_file(
         """
         if filepath is None and map_path is None:
             raise ValueError("Either `filepath` or `map_path` must be provided.")
-        
+
         if filepath is not None and Path(filepath).suffix == ".parquet":
             r = cls.from_parquet(filepath, parse_map=parse_map, validate=validate, step_size=step_size)
         elif filepath is not None:
@@ -649,7 +649,6 @@ def plot_altair(
         return view.add_params(op_var)
 
     def create_mapsegments(self):
-        
         if isinstance(self.map, MapOsiCenterline):
             self.mapsegment = MapOsiCenterlineSegmentation(self)
             self.mapsegment.init_intersections()

pyproject.toml

@@ -48,7 +48,7 @@ dependencies = [
     'tqdm_joblib',
     'filelock>=3.18.0'
 ]
-version = "0.1.21"
+version = "0.2.0"
 
 [project.urls]
 Homepage = "https://github.com/ika-rwth-aachen/omega-prime"