Add intrusion computation

Author: schroedtert

pedpy/__init__.py

@@ -55,6 +55,7 @@
     compute_passing_speed,
     compute_voronoi_speed,
 )
+from .methods.foo_calculator import compute_intrusion, IntrusionMethod
 from .plotting.plotting import (
     PEDPY_BLUE,
     PEDPY_GREEN,
@@ -116,6 +117,8 @@
     "compute_mean_speed_per_frame",
     "compute_passing_speed",
     "compute_voronoi_speed",
+    "compute_intrusion",
+    "IntrusionMethod",
     "PEDPY_BLUE",
     "PEDPY_GREEN",
     "PEDPY_GREY",

pedpy/column_identifier.py

@@ -29,3 +29,5 @@
 START_POSITION_COL: Final = "start_position"
 END_POSITION_COL: Final = "end_position"
 WINDOW_SIZE_COL: Final = "window_size"
+
+INTRUSION_COL: Final = "intrusion"

pedpy/methods/foo_calculator.py

@@ -0,0 +1,52 @@
+"""Foo."""
+
+
+from enum import Enum
+
+from pedpy.data.trajectory_data import TrajectoryData
+from .method_utils import _compute_individual_distances
+from pedpy.column_identifier import FRAME_COL, ID_COL, INTRUSION_COL
+import pandas
+
+
+class IntrusionMethod(Enum):  # pylint: disable=too-few-public-methods
+    """Identifier of method to compute the intrusion."""
+
+    MAX = "max"
+    """Use the max value in neighborhood as intrusion."""
+    SUM = "sum"
+    """Use the sum of all values in neighborhood as intrusion."""
+
+
+def compute_intrusion(
+    *,
+    traj_data: TrajectoryData,
+    r_soc: float = 0.8,
+    l_min: float = 0.2,
+    method: IntrusionMethod = IntrusionMethod.SUM,
+) -> pandas.DataFrame:
+    """_summary_.
+
+    TODO add documentation here
+    Args:
+        traj_data (TrajectoryData): _description_
+        r_soc (float): _description_
+        l_min (float): _description_
+        method (IntrusionMethod, optional): _description_. Defaults to IntrusionMethod.SUM.
+
+    Returns:
+        pandas.DataFrame: _description_
+    """
+    intrusion = _compute_individual_distances(traj_data=traj_data)
+    intrusion[INTRUSION_COL] = (
+        (r_soc - l_min) / (intrusion.distance - l_min)
+    ) ** 2
+    intrusion = (
+        intrusion.groupby(by=[ID_COL, FRAME_COL])
+        .agg(
+            intrusion=(INTRUSION_COL, method.value),
+        )
+        .reset_index()
+    )
+
+    return intrusion

pedpy/methods/method_utils.py

@@ -947,3 +947,23 @@ def _check_crossing_in_frame_range(
     crossed[column_name] = crossed[column_name] == "both"
     crossed = crossed[crossed[column_name]]
     return crossed
+
+
+def _compute_individual_distances(*, traj_data: TrajectoryData) -> pd.DataFrame:
+    matrix = pd.merge(
+        traj_data.data[["id", "frame", "point"]],
+        traj_data.data[["id", "frame", "point"]],
+        how="outer",
+        on="frame",
+        suffixes=("", "_neighbor"),
+    )
+    matrix = matrix[matrix.id != matrix.id_neighbor]
+    matrix["distance"] = np.linalg.norm(
+        shapely.get_coordinates(matrix.point)
+        - shapely.get_coordinates(matrix.point_neighbor),
+        axis=1,
+    )
+
+    matrix = matrix.drop(columns=["point", "point_neighbor"])
+    matrix = matrix.rename(columns={"id_neighbor": "neighbor"})
+    return matrix.reset_index(drop=True)