Add viswalk trajectory loader (#252)

Co-authored-by: Mohcine Chraibi <m.chraibi@fz-juelich.de>

Author: schroedtert

notebooks/demo-data/viswalk/example.pp

@@ -411,6 +411,58 @@
     "plt.show()"
    ]
   },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "#### Loading from Viswalk trajectory files\n",
+    "\n",
+    "It is also possible to load trajectory files from [Viswalk](https://www.ptvgroup.com/en/products/pedestrian-simulation-software-ptv-viswalk) directly into *PedPy*. \n",
+    "The expected format is a CSV file with `;` as delimiter, and it should contain at least the following columns: `NO`, `SIMSEC`, `COORDCENTX`, `COORDCENTY`.\n",
+    "Comment lines may start with a `*` and will be ignored.\n",
+    "\n",
+    ":::{important}\n",
+    "Currently only Viswalk trajectory files, which use the simulation time (`SIMSEC`) are supported.\n",
+    ":::\n",
+    "\n",
+    "\n",
+    "To make the data usable for *PedPy* use:"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "from pedpy import (\n",
+    "    TrajectoryData,\n",
+    "    load_trajectory_from_viswalk,\n",
+    ")\n",
+    "import pathlib\n",
+    "\n",
+    "viswalk_file = pathlib.Path(\"demo-data/viswalk/example.pp\")\n",
+    "\n",
+    "traj_viswalk = load_trajectory_from_viswalk(trajectory_file=viswalk_file)"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {
+    "tags": [
+     "hide-input"
+    ]
+   },
+   "outputs": [],
+   "source": [
+    "import matplotlib.pyplot as plt\n",
+    "from pedpy import plot_trajectories\n",
+    "\n",
+    "plot_trajectories(traj=traj_viswalk).set_aspect(\"equal\")\n",
+    "plt.show()"
+   ]
+  },
   {
    "cell_type": "markdown",
    "metadata": {

notebooks/user_guide.ipynb

@@ -19,6 +19,7 @@
     load_trajectory_from_jupedsim_sqlite,
     load_trajectory_from_ped_data_archive_hdf5,
     load_trajectory_from_txt,
+    load_trajectory_from_viswalk,
     load_walkable_area_from_jupedsim_sqlite,
     load_walkable_area_from_ped_data_archive_hdf5,
 )
@@ -87,6 +88,7 @@
     "load_trajectory_from_jupedsim_sqlite",
     "load_trajectory_from_ped_data_archive_hdf5",
     "load_trajectory_from_txt",
+    "load_trajectory_from_viswalk",
     "load_walkable_area_from_jupedsim_sqlite",
     "load_walkable_area_from_ped_data_archive_hdf5",
     "compute_classic_density",

pedpy/__init__.py

@@ -1,4 +1,5 @@
 """Load trajectories to the internal trajectory data format."""
+import math
 import pathlib
 import sqlite3
 from enum import Enum
@@ -449,3 +450,134 @@ def load_walkable_area_from_ped_data_archive_hdf5(
         walkable_area = WalkableArea(hdf5_file.attrs["wkt_geometry"])
 
     return walkable_area
+
+
+def load_trajectory_from_viswalk(
+    *,
+    trajectory_file: pathlib.Path,
+) -> TrajectoryData:
+    """Loads data from Viswalk-csv file as :class:`~trajectory_data.TrajectoryData`.
+
+    This function reads a CSV file containing trajectory data from Viswalk simulations and
+    converts it into a :class:`~trajectory_data.TrajectoryData` object which can be used for
+    further analysis and processing in the *PedPy* framework.
+
+    .. note::
+
+        Viswalk data have a time column, that is going to be converted to a frame column for use
+        with *PedPy*.
+
+    .. warning::
+
+        Currently only Viswalk files with a time column can be loaded.
+
+    Args:
+        trajectory_file: The full path of the CSV file containing the Viswalk
+            trajectory data. The expected format is a CSV file with :code:`;` as delimiter, and it
+            should contain at least the following columns: NO, SIMSEC, COORDCENTX, COORDCENTY.
+            Comment lines may start with a :code:`*` and will be ignored.
+
+    Returns:
+        :class:`~trajectory_data.TrajectoryData` representation of the file data
+
+    Raises:
+        LoadTrajectoryError: If the provided path does not exist or is not a file.
+    """
+    _validate_is_file(trajectory_file)
+
+    traj_dataframe = _load_trajectory_data_from_viswalk(
+        trajectory_file=trajectory_file
+    )
+    traj_dataframe["frame"], traj_frame_rate = _calculate_frames_and_fps(
+        traj_dataframe
+    )
+
+    return TrajectoryData(
+        data=traj_dataframe[[ID_COL, FRAME_COL, X_COL, Y_COL]],
+        frame_rate=traj_frame_rate,
+    )
+
+
+def _calculate_frames_and_fps(
+    traj_dataframe: pd.DataFrame,
+) -> Tuple[pd.Series, int]:
+    """Calculates fps and frames based on the time column of the dataframe."""
+    mean_diff = traj_dataframe.groupby(ID_COL)["time"].diff().dropna().mean()
+    if math.isnan(mean_diff):
+        raise LoadTrajectoryError(
+            "Can not determine the frame rate used to write the trajectory file. "
+            "This may happen, if the file only contains data for a single frame."
+        )
+
+    fps = int(round(1 / mean_diff))
+    frames = traj_dataframe["time"] * fps
+    frames = frames.round().astype("int64")
+    return frames, fps
+
+
+def _load_trajectory_data_from_viswalk(
+    *, trajectory_file: pathlib.Path
+) -> pd.DataFrame:
+    """Parse the trajectory file for trajectory data.
+
+    Args:
+        trajectory_file (pathlib.Path): The file containing the trajectory data.
+            The expected format is a CSV file with ';' as delimiter, and it should
+            contain at least the following columns: NO, SIMSEC, COORDCENTX, COORDCENTY.
+            Comment lines may start with a '*' and will be ignored.
+
+    Returns:
+        The trajectory data as :class:`DataFrame`, the coordinates are
+        in meter (m).
+    """
+    columns_to_keep = ["NO", "SIMSEC", "COORDCENTX", "COORDCENTY"]
+    rename_mapping = {
+        "NO": ID_COL,
+        "SIMSEC": "time",
+        "COORDCENTX": X_COL,
+        "COORDCENTY": Y_COL,
+    }
+    common_error_message = (
+        "The given trajectory file seems to be incorrect or empty. "
+        "It should contain at least the following columns: "
+        "NO, SIMSEC, COORDCENTX, COORDCENTY, separated by ';'. "
+        "Comment lines may start with a '*' and will be ignored. "
+        f"Please check your trajectory file: {trajectory_file}."
+    )
+    try:
+        data = pd.read_csv(
+            trajectory_file,
+            delimiter=";",
+            skiprows=1,  # skip first row containing '$VISION'
+            comment="*",
+            dtype={
+                ID_COL: "int64",
+                "time": "float64",
+                X_COL: "float64",
+                Y_COL: "float64",
+            },
+            encoding="utf-8-sig",
+        )
+        got_columns = data.columns
+        cleaned_columns = got_columns.map(
+            lambda x: x.replace("$PEDESTRIAN:", "")
+        )
+        set_columns_to_keep = set(columns_to_keep)
+        set_cleaned_columns = set(cleaned_columns)
+        missing_columns = set_columns_to_keep - set_cleaned_columns
+        if missing_columns:
+            raise LoadTrajectoryError(
+                f"{common_error_message}"
+                f"Missing columns: {', '.join(missing_columns)}."
+            )
+
+        data.columns = cleaned_columns
+        data = data[columns_to_keep]
+        data.rename(columns=rename_mapping, inplace=True)
+
+        if data.empty:
+            raise LoadTrajectoryError(common_error_message)
+
+        return data
+    except pd.errors.ParserError as exc:
+        raise LoadTrajectoryError(common_error_message) from exc