ruff format

Author: MichaelSchuldes

omega_prime/__init__.py

@@ -1,5 +1,5 @@
 __pdoc__ = {}
-__pdoc__['converters'] = False
+__pdoc__["converters"] = False
 """ .. include:: ./../README.md """
 from . import converters
 from .asam_odr import MapOdr
@@ -8,4 +8,16 @@
 from .recording import MovingObject, Recording
 from .converters import convert_lxd
 
-__all__ = ['Recording', 'MovingObject', 'MapOsi', 'Map', 'Lane', 'LaneBoundary', 'MapOdr', 'Locator', 'LaneRelation', 'converters', 'convert_lxd']
+__all__ = [
+    "Recording",
+    "MovingObject",
+    "MapOsi",
+    "Map",
+    "Lane",
+    "LaneBoundary",
+    "MapOdr",
+    "Locator",
+    "LaneRelation",
+    "converters",
+    "convert_lxd",
+]

omega_prime/asam_odr/__init__.py

@@ -1,3 +1,3 @@
 from .odr_map import MapOdr
 
-__all__ = ['MapOdr']
+__all__ = ["MapOdr"]

omega_prime/asam_odr/eulerspiral.py

@@ -1,11 +1,9 @@
-
 # copied from https://github.com/stefan-urban/pyeulerspiral/blob/master/eulerspiral/eulerspiral.py (MIT License)
 import numpy as np
 from scipy.special import fresnel
 
 
 class EulerSpiral:
-
     def __init__(self, gamma):
         self._gamma = gamma
 
@@ -14,7 +12,6 @@ def createFromLengthAndCurvature(length, curvStart, curvEnd):
         return EulerSpiral(1 * (curvEnd - curvStart) / length)
 
     def calc(self, s, x0=0, y0=0, kappa0=0, theta0=0):
-
         # Start
         C0 = x0 + 1j * y0
 
@@ -40,4 +37,4 @@ def calc(self, s, x0=0, y0=0, kappa0=0, theta0=0):
         # Tangent at each point
         theta = self._gamma * s**2 / 2 + kappa0 * s + theta0
 
-        return (Cs.real, Cs.imag, theta)
+        return (Cs.real, Cs.imag, theta)

omega_prime/asam_odr/logger.py

@@ -1,4 +1,4 @@
 import logging
 
-logger = logging.getLogger('omega-prime ASAM OpenDRIVE Reader')
-logger.setLevel(logging.INFO)
+logger = logging.getLogger("omega-prime ASAM OpenDRIVE Reader")
+logger.setLevel(logging.INFO)

omega_prime/asam_odr/odr_map.py

@@ -26,42 +26,49 @@ class MapOdr(Map):
     proj_string: str | None = None
     proj_offset: ProjectionOffset | None = None
     projection: pyproj.CRS | None = None
-    
+
     @classmethod
-    def from_file(cls, filename, topic='ground_truth_map', is_odr_xml: bool = False, is_mcap: bool = False, step_size=0.1, skip_parse: bool = False):
-        if Path(filename).suffix in ['.xodr', '.odr'] or is_odr_xml:
+    def from_file(
+        cls,
+        filename,
+        topic="ground_truth_map",
+        is_odr_xml: bool = False,
+        is_mcap: bool = False,
+        step_size=0.1,
+        skip_parse: bool = False,
+    ):
+        if Path(filename).suffix in [".xodr", ".odr"] or is_odr_xml:
             with open(filename) as f:
-                self = cls.create(odr_xml=f.read(), name=Path(filename).stem, step_size=step_size, skip_parse=skip_parse)
+                self = cls.create(
+                    odr_xml=f.read(), name=Path(filename).stem, step_size=step_size, skip_parse=skip_parse
+                )
             return self
-        elif Path(filename).suffix in ['.mcap'] or is_mcap:
+        elif Path(filename).suffix in [".mcap"] or is_mcap:
             map = next(iter(betterosi.read(filename, mcap_topics=[topic], osi_message_type=betterosi.MapAsamOpenDrive)))
-            return cls.create(odr_xml=map.open_drive_xml_content, name=map.map_reference, step_size=step_size, skip_parse=skip_parse)
-    
+            return cls.create(
+                odr_xml=map.open_drive_xml_content, name=map.map_reference, step_size=step_size, skip_parse=skip_parse
+            )
+
     @classmethod
-    def create(cls, odr_xml, name, step_size=.1, skip_parse: bool = False):
-        self = cls(
-            odr_xml = odr_xml,
-            name = name,
-            lane_boundaries = {},
-            lanes = {},
-            step_size=step_size,
-            _odr_object = None
-        )
+    def create(cls, odr_xml, name, step_size=0.1, skip_parse: bool = False):
+        self = cls(odr_xml=odr_xml, name=name, lane_boundaries={}, lanes={}, step_size=step_size, _odr_object=None)
         if not skip_parse:
             self.parse()
             return self
-        
+
     def parse(self):
         if self._odr_object is not None:
-            return 
+            return
         xml = etree.fromstring(self.odr_xml.encode("utf-8"))
         self._odr_object = parse_opendrive(xml)
         self.proj_string = self._odr_object.header.geo_reference
         if self.proj_string is not None:
             try:
                 self.projection = pyproj.CRS.from_proj4(self.proj_string)
             except pyproj.exceptions.CRSError as e:
-                raise pyproj.exceptions.CRSError('Povided ASAM OpenDRIVE XML does not contain a valid georeference!') from e
+                raise pyproj.exceptions.CRSError(
+                    "Povided ASAM OpenDRIVE XML does not contain a valid georeference!"
+                ) from e
         self.proj_offsets = self._odr_object.header.offset
         self.roads = convert_opendrive(self._odr_object, step_size=self.step_size)
         self.lane_boundaries = {}
@@ -71,71 +78,83 @@ def parse(self):
                 self.lane_boundaries[(rid, bid)] = b
             for lid, l in r.lanes.items():
                 self.lanes[(rid, lid)] = l
- 
-    def to_file(self, filename: Path|None = None):
+
+    def to_file(self, filename: Path | None = None):
         if filename is None:
             p = Path(self.name)
         else:
             p = Path(filename)
             if p.is_dir():
-                p = p/f'{self.name}'
-        if p.suffix == '':
-            p = Path(str(p) +'.xodr')
-        with open(p, 'w') as f:
+                p = p / f"{self.name}"
+        if p.suffix == "":
+            p = Path(str(p) + ".xodr")
+        with open(p, "w") as f:
             f.write(self.odr_xml)
-        print(f'Extracted {p}')
- 
+        print(f"Extracted {p}")
+
     def to_osi(self):
         return betterosi.MapAsamOpenDrive(map_reference=self.name, open_drive_xml_content=self.odr_xml)
-    
+
     def to_hdf(self, filename):
         import tables
+
         with tables.open_file(filename, mode="a") as h5file:
             try:
-                gmap = h5file.get_node('/map')
+                gmap = h5file.get_node("/map")
             except tables.NoSuchNodeError:
-                gmap = h5file.create_group('/', 'map')
+                gmap = h5file.create_group("/", "map")
             atom = tables.StringAtom(itemsize=len(self.odr_xml))
             ds = h5file.create_carray(gmap, "odr_xml", atom, shape=(1,))
-            ds[0] = self.odr_xml.encode('utf-8')
-            
+            ds[0] = self.odr_xml.encode("utf-8")
+
     @classmethod
     def from_hdf(cls, filename):
         import tables
+
         with tables.open_file(filename, mode="r") as h5file:
-            odr_xml = h5file.get_node('/map').odr_xml[0].decode()
+            odr_xml = h5file.get_node("/map").odr_xml[0].decode()
         return cls.create(odr_xml, str(filename))
 
-    
     def plot(self, ax=None):
         if ax is None:
-            _, ax = plt.subplots(1,1)
-            
+            _, ax = plt.subplots(1, 1)
+
         for rid, r in self.roads.items():
-            ax.plot(*r.centerline_points[:,1:3].T, c='black')  
+            ax.plot(*r.centerline_points[:, 1:3].T, c="black")
         for lid, l in self.lanes.items():
-            c = 'blue' if l.type==betterosi.LaneClassificationType.TYPE_UNKNOWN else 'green'
+            c = "blue" if l.type == betterosi.LaneClassificationType.TYPE_UNKNOWN else "green"
             lb = self.lane_boundaries[l.left_boundary_id]
             rb = self.lane_boundaries[l.right_boundary_id]
-            ax.add_patch(PltPolygon(np.concatenate([lb.polyline[:,:2], np.flip(rb.polyline[:,:2], axis=0)]), fc=c, alpha=0.5, ec='black'))
+            ax.add_patch(
+                PltPolygon(
+                    np.concatenate([lb.polyline[:, :2], np.flip(rb.polyline[:, :2], axis=0)]),
+                    fc=c,
+                    alpha=0.5,
+                    ec="black",
+                )
+            )
         ax.autoscale()
         ax.set_aspect(1)
         return ax
-        
 
     def setup_lanes_and_boundaries(self):
         self.parse()
         for lid, l in self.lanes.items():
             l.idx = lid
             l.left_boundary = self.lane_boundaries[l.left_boundary_id]
             l.right_boundary = self.lane_boundaries[l.right_boundary_id]
-            
-            l.centerline = l.left_boundary.polyline+(np.diff(np.stack([l.left_boundary.polyline, l.right_boundary.polyline]),axis=0)/2)[0,...]
-            #l.centerline = shapely.LineString(l.centerline)
-            
-            poly_points = np.concatenate([l.left_boundary.polyline[:,:2], np.flip(l.right_boundary.polyline[:,:2], axis=0)])
+
+            l.centerline = (
+                l.left_boundary.polyline
+                + (np.diff(np.stack([l.left_boundary.polyline, l.right_boundary.polyline]), axis=0) / 2)[0, ...]
+            )
+            # l.centerline = shapely.LineString(l.centerline)
+
+            poly_points = np.concatenate(
+                [l.left_boundary.polyline[:, :2], np.flip(l.right_boundary.polyline[:, :2], axis=0)]
+            )
             l.polygon = shapely.Polygon(poly_points)
             if not l.polygon.is_valid:
                 l.polygon = shapely.convex_hull(l.polygon)
-            l.start_points = np.stack([l.left_boundary.polyline[0,:2], l.right_boundary.polyline[0,:2]])
-            l.end_points = np.stack([l.left_boundary.polyline[-1,:2], l.right_boundary.polyline[-1,:2]])
+            l.start_points = np.stack([l.left_boundary.polyline[0, :2], l.right_boundary.polyline[0, :2]])
+            l.end_points = np.stack([l.left_boundary.polyline[-1, :2], l.right_boundary.polyline[-1, :2]])

omega_prime/asam_odr/opendriveconverter/converter.py

@@ -12,7 +12,7 @@ class Map:
     georeference: Any = None
 
 
-def convert_opendrive(my_opendrive, step_size=0.1) -> tuple[dict, tuple[float,float,str]]:
+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
     :param my_opendrive: opendrive XML
@@ -22,25 +22,27 @@ def convert_opendrive(my_opendrive, step_size=0.1) -> tuple[dict, tuple[float,fl
 
     # create roads object and set geo reference
     my_roads = {}
-    #georeference = get_georeference(my_opendrive.header)
+    # georeference = get_georeference(my_opendrive.header)
 
     # tables needed for matching their ids in open drive to VVM format since many connections can only be set later
-    lookup_table = []   # [road id | lane_section | laneID | road id VVM | lane id VVM]
-    reference_table = []    # [omega signal id | omega road id | odr signal id | odr reference id | odr connected_to id]
+    lookup_table = []  # [road id | lane_section | laneID | road id VVM | lane id VVM]
+    reference_table = []  # [omega signal id | omega road id | odr signal id | odr reference id | odr connected_to id]
 
-    logger.info(f'Starting conversion of {len(my_opendrive.roads)} opendrive roads into OMEGAFormat.')
+    logger.info(f"Starting conversion of {len(my_opendrive.roads)} opendrive roads into OMEGAFormat.")
 
     # loop over individual opendrive roads and add them to omega roads dictionary in my_roads
     for road in my_opendrive.roads:
-        my_roads, lookup_table, reference_table = process_road(my_roads, road, my_opendrive, step_size, lookup_table, reference_table)
+        my_roads, lookup_table, reference_table = process_road(
+            my_roads, road, my_opendrive, step_size, lookup_table, reference_table
+        )
 
     detected_roads = list(my_roads.keys())
-    my_roads = {rid: r for rid, r in my_roads.items() if len(r.lanes)>0}
+    my_roads = {rid: r for rid, r in my_roads.items() if len(r.lanes) > 0}
     no_lane_roads = [rid for rid in detected_roads if rid not in my_roads.keys()]
-    if len(no_lane_roads)>0:
+    if len(no_lane_roads) > 0:
         print(f"There are roads with no lanes {no_lane_roads}")
     # setting up road network information via connections and references
-    logger.info('Setting up roads connections.')
+    logger.info("Setting up roads connections.")
     my_roads = setup_connections(my_roads, lookup_table, my_opendrive)
 
     return my_roads