Apply Ruff formatting

Author: github-actions[bot]

omega_prime/map_odr.py

@@ -115,7 +115,7 @@ def xodr_map(self):
     def from_file(
         cls,
         filename,
-        topics: list[str] = ["/ground_truth_map","ground_truth_map"],
+        topics: list[str] = ["/ground_truth_map", "ground_truth_map"],
         parse: bool = False,
         is_odr_xml: bool = False,
         is_mcap: bool = False,

omega_prime/metrics.py

@@ -16,50 +16,58 @@ class Metric:
     requires_properties: list[str] = field(default_factory=list)
     _parameters: list = field(init=False)
 
-    def compute_lazy(self, df: pl.LazyFrame, **kwargs) -> tuple[pl.LazyFrame, dict[str,pl.LazyFrame]]:
+    def compute_lazy(self, df: pl.LazyFrame, **kwargs) -> tuple[pl.LazyFrame, dict[str, pl.LazyFrame]]:
         try:
             df, properties = self.compute_func(df, **kwargs)
             assert isinstance(df, pl.LazyFrame)
             assert all(p in properties for p in self.computes_properties)
             return df, properties
 
         except TypeError as e:
-            raise TypeError(f'Missing paramter for Metric with compute_func {self.compute_func.__name__}: {repr(e)}') from e
+            raise TypeError(
+                f"Missing paramter for Metric with compute_func {self.compute_func.__name__}: {repr(e)}"
+            ) from e
 
     def __post_init__(self):
         sig = inspect.signature(self.compute_func)
         parameters = sig.parameters
-        assert 'df' in parameters
+        assert "df" in parameters
         assert all(p in parameters for p in self.requires_properties)
 
-        self._parameters = [v for k, v in parameters.items() if k not in ['df', 'args', 'kwargs']+self.requires_properties]
+        self._parameters = [
+            v for k, v in parameters.items() if k not in ["df", "args", "kwargs"] + self.requires_properties
+        ]
 
     def __call__(self, df: pl.DataFrame, **kwargs):
         try:
             if not isinstance(df, pl.LazyFrame):
                 df = pl.LazyFrame(df)
             return self.compute_lazy(df, **kwargs)
         except TypeError as e:
-            raise TypeError(f'Missing paramter for Metric with compute_func {self.compute_func.__name__}: {repr(e)}') from e
+            raise TypeError(
+                f"Missing paramter for Metric with compute_func {self.compute_func.__name__}: {repr(e)}"
+            ) from e
 
 
-def metric(computes_columns: list[str]|None = None,
-        computes_properties: list[str]|None = None,
-        requires_columns: list[str]|None = None,
-        requires_properties: list[str]|None = None):
+def metric(
+    computes_columns: list[str] | None = None,
+    computes_properties: list[str] | None = None,
+    requires_columns: list[str] | None = None,
+    requires_properties: list[str] | None = None,
+):
     def decorator(func):
         return Metric(
-                    compute_func = func,
-                    computes_columns=computes_columns or [],
-                    computes_properties=computes_properties or [],
-                    requires_columns=requires_columns or [],
-                    requires_properties=requires_properties or []
+            compute_func=func,
+            computes_columns=computes_columns or [],
+            computes_properties=computes_properties or [],
+            requires_columns=requires_columns or [],
+            requires_properties=requires_properties or [],
         )
-    return decorator
 
+    return decorator
 
 
-@metric(computes_columns = ['distance_traveled', 'vel'])
+@metric(computes_columns=["distance_traveled", "vel"])
 def driven_distance_and_vel(df) -> tuple[pl.DataFrame, dict[str, pl.DataFrame]]:
     return df.with_columns(
         (pl.col("x").diff() ** 2 + pl.col("y").diff() ** 2)
@@ -72,10 +80,11 @@ def driven_distance_and_vel(df) -> tuple[pl.DataFrame, dict[str, pl.DataFrame]]:
     ), {}
 
 
-@metric(requires_columns = ['distance_traveled', 'vel'],
-        computes_properties = ['timegaps', 'min_timegaps', 'p_timegaps', 'min_p_timegaps'])
+@metric(
+    requires_columns=["distance_traveled", "vel"],
+    computes_properties=["timegaps", "min_timegaps", "p_timegaps", "min_p_timegaps"],
+)
 def timegaps_and_p_timegaps(df, /, ego_id, time_buffer=2e9):
-
     ego_df = df.filter(idx=ego_id)
 
     crossed = df.join(ego_df, how="cross", suffix="_ego")
@@ -150,32 +159,36 @@ def timegaps_and_p_timegaps(df, /, ego_id, time_buffer=2e9):
     }
 
 
-
-
-metrics = [
-    timegaps_and_p_timegaps,
-    driven_distance_and_vel
-]
+metrics = [timegaps_and_p_timegaps, driven_distance_and_vel]
 
 
 @dataclass
 class MetricManager:
     metrics: list[Metric] = field(default_factory=lambda: metrics)
     exclude_columns: list[str] = field(default_factory=list)
     exclude_properties: list[str] = field(default_factory=list)
-    _dependencies: dict[int|str, list[int|str]] = field(init=False)
+    _dependencies: dict[int | str, list[int | str]] = field(init=False)
     _ordered_metrics: list[Metric] = field(init=False)
     _parameters: list = field(init=False)
 
     def __post_init__(self):
-        self._dependencies = {val: [i] for i,m in enumerate(self.metrics) for val in [f'column_{n}' for n in m.computes_columns]+[f'property_{n}' for n in m.computes_properties]}|{
-            i: [f'column_{n}' for n in m.requires_columns]+[f'property_{n}' for n in m.requires_properties] for i,m in enumerate(self.metrics)
+        self._dependencies = {
+            val: [i]
+            for i, m in enumerate(self.metrics)
+            for val in [f"column_{n}" for n in m.computes_columns] + [f"property_{n}" for n in m.computes_properties]
+        } | {
+            i: [f"column_{n}" for n in m.requires_columns] + [f"property_{n}" for n in m.requires_properties]
+            for i, m in enumerate(self.metrics)
         }
 
-        unresovled_dependencies= {k: v for k,vv in self._dependencies.items() for v in vv if v not in self._dependencies}
+        unresovled_dependencies = {
+            k: v for k, vv in self._dependencies.items() for v in vv if v not in self._dependencies
+        }
         if len(unresovled_dependencies) > 0:
-            error_dict = {f'self.metrics[{k}]': v for k,v in unresovled_dependencies.items()}
-            raise RuntimeError(f"There are columns and properties required by metrics, that are never computed: {error_dict}")
+            error_dict = {f"self.metrics[{k}]": v for k, v in unresovled_dependencies.items()}
+            raise RuntimeError(
+                f"There are columns and properties required by metrics, that are never computed: {error_dict}"
+            )
 
         self._parameters = [v for m in self.metrics for v in m._parameters]
 
@@ -185,7 +198,7 @@ def __post_init__(self):
     def __repr__(self):
         return f"computes columns: {[c for m in self._ordered_metrics for c in m.computes_columns]} - computes properties {[p for m in self._ordered_metrics for p in m.computes_properties]} - args {self._parameters}"
 
-    def compute(self, r: Recording, **kwargs) -> tuple[pl.DataFrame, dict[str,pl.DataFrame]]:
+    def compute(self, r: Recording, **kwargs) -> tuple[pl.DataFrame, dict[str, pl.DataFrame]]:
         if "polygon" not in r._df.columns:
             r._df = r._add_polygons(r._df)
         if "geometry" not in r._df.columns:
@@ -197,13 +210,13 @@ def compute(self, r: Recording, **kwargs) -> tuple[pl.DataFrame, dict[str,pl.Dat
             df, new_p = m.compute_lazy(
                 df=df,
                 **{k: properties[k] for k in m.requires_properties},
-                **{k: v for k,v in kwargs.items() if k in [p.name for p in m._parameters]}
+                **{k: v for k, v in kwargs.items() if k in [p.name for p in m._parameters]},
             )
             properties |= new_p
         for k in self.exclude_properties:
             del properties[k]
         df = df.drop(self.exclude_columns)
-        res = pl.collect_all([df]+list(properties.values()))
-        df, computed_props  = res[0], res[1:]
+        res = pl.collect_all([df] + list(properties.values()))
+        df, computed_props = res[0], res[1:]
         assert all(c in df.columns or c in self.exclude_columns for m in self.metrics for c in m.computes_columns)
-        return df, {k:v for k,v in zip(properties.keys(), computed_props)}
+        return df, {k: v for k, v in zip(properties.keys(), computed_props)}

omega_prime/recording.py

@@ -33,7 +33,9 @@
 def other_column_set_on_column_value(
     polars_obj, *, column_name: str, column_value, other_column_name: str, other_column_unset_value
 ):
-    return polars_obj.lazyframe.select(~(pl.col(column_name)==column_value).and_(pl.col(other_column_name) == other_column_unset_value))
+    return polars_obj.lazyframe.select(
+        ~(pl.col(column_name) == column_value).and_(pl.col(other_column_name) == other_column_unset_value)
+    )
 
 
 @extensions.register_check_method(
@@ -42,7 +44,9 @@ def other_column_set_on_column_value(
 def other_column_unset_on_column_value(
     polars_obj, *, column_name, column_value, other_column_name: str, other_column_unset_value: int
 ):
-    return  polars_obj.lazyframe.select(~(pl.col(column_name)!=column_value).and_(pl.col(other_column_name) != other_column_unset_value))
+    return polars_obj.lazyframe.select(
+        ~(pl.col(column_name) != column_value).and_(pl.col(other_column_name) != other_column_unset_value)
+    )
 
 
 recording_moving_object_schema = pa.DataFrameSchema(