Repository: mie-lab/trackintel (python)

Issue:
Support multiprocessing for generate_tours
It would be great if `generate_tours` would support parallelization. This can be easily implemented by using [apply_parallel](https://github.com/mie-lab/trackintel/blob/master/trackintel/preprocessing/util.py#L42) 
instead of [apply](https://github.com/mie-lab/trackintel/blob/master/trackintel/preprocessing/trips.py#L152) as it is for example done in `generate_locations` [here](https://github.com/mie-lab/trackintel/blob/master/trackintel/preprocessing/staypoints.py#L96)

Pull request:
ENH: multiprocessing for generate_tours
closes #412

Patch:
diff --git a/trackintel/preprocessing/trips.py b/trackintel/preprocessing/trips.py
index 7df3e87..b197964 100644
--- a/trackintel/preprocessing/trips.py
+++ b/trackintel/preprocessing/trips.py
@@ -8,6 +8,7 @@ from tqdm import tqdm
 
 import trackintel as ti
 from trackintel import Tours, Trips
+from trackintel.preprocessing.util import applyParallel
 
 
 def get_trips_grouped(trips, tours):
@@ -59,6 +60,7 @@ def generate_tours(
     max_time="1d",
     max_nr_gaps=0,
     print_progress=False,
+    n_jobs=1,
 ):
     """
     Generate trackintel-tours from trips
@@ -85,6 +87,12 @@ def generate_tours(
     print_progress : bool, default False
         If print_progress is True, the progress bar is displayed
 
+    n_jobs: int, default 1
+        The maximum number of concurrently running jobs. If -1 all CPUs are used. If 1 is given, no parallel
+        computing code is used at all, which is useful for debugging. See
+        https://joblib.readthedocs.io/en/latest/parallel.html#parallel-reference-documentation
+        for a detailed description
+
     Returns
     -------
     trips_with_tours: Trips
@@ -150,19 +158,14 @@ def generate_tours(
         "geom_col": geom_col,
         "crs_is_projected": crs_is_projected,
     }
-    if print_progress:
-        tqdm.pandas(desc="User tour generation")
-        tours = (
-            trips_input.groupby(["user_id"], group_keys=False, as_index=False)
-            .progress_apply(_generate_tours_user, **kwargs)
-            .reset_index(drop=True)
-        )
-    else:
-        tours = (
-            trips_input.groupby(["user_id"], group_keys=False, as_index=False)
-            .apply(_generate_tours_user, **kwargs)
-            .reset_index(drop=True)
-        )
+
+    tours = applyParallel(
+        trips_input.groupby("user_id", group_keys=False, as_index=False),
+        _generate_tours_user,
+        print_progress=print_progress,
+        n_jobs=n_jobs,
+        **kwargs
+    )
 
     # No tours found
     if len(tours) == 0:


Test patch:
diff --git a/tests/preprocessing/test_trips.py b/tests/preprocessing/test_trips.py
index 8cd1a07..0043423 100644
--- a/tests/preprocessing/test_trips.py
+++ b/tests/preprocessing/test_trips.py
@@ -146,8 +146,8 @@ class TestGenerate_tours:
 
     def test_generate_tours(self, example_trip_data):
         """Test general functionality of generate tours function"""
-        trips, sp_locs = example_trip_data
-        trips_out, tours = ti.preprocessing.trips.generate_tours(trips)
+        trips, _ = example_trip_data
+        trips_out, _ = ti.preprocessing.trips.generate_tours(trips)
         # check that nothing else than the new column has changed in trips df
         assert all(trips_out.iloc[:, :6] == trips)
         # check that the two tours were found
@@ -161,9 +161,22 @@ class TestGenerate_tours:
         user_1_df = trips_out[trips_out["user_id"] == 1]
         assert all(pd.isna(user_1_df["tour_id"]))
 
+    def test_parallel_computing(self, example_trip_data):
+        """The result obtained with parallel computing should be identical."""
+        trips, _ = example_trip_data
+
+        # without parallel computing code
+        trips_ori, tours_ori = ti.preprocessing.trips.generate_tours(trips, n_jobs=1)
+        # using two cores
+        trips_para, tours_para = ti.preprocessing.trips.generate_tours(trips, n_jobs=2)
+
+        # the result of parallel computing should be identical
+        assert_geodataframe_equal(trips_ori, trips_para)
+        pd.testing.assert_frame_equal(tours_ori, tours_para)
+
     def test_tours_with_gap(self, example_trip_data):
         """Test functionality of max_nr_gaps parameter in tour generation"""
-        trips, sp_locs = example_trip_data
+        trips, _ = example_trip_data
         trips_out, tours = ti.preprocessing.trips.generate_tours(trips, max_nr_gaps=1)
         # new tour was found for user 1
         assert len(tours) == 3
@@ -172,8 +185,8 @@ class TestGenerate_tours:
 
     def test_tour_times(self, example_trip_data):
         """Check whether the start and end times of generated tours are correct"""
-        trips, sp_locs = example_trip_data
-        trips_out, tours = ti.preprocessing.trips.generate_tours(trips, max_nr_gaps=1, max_time="1d")
+        trips, _ = example_trip_data
+        _, tours = ti.preprocessing.trips.generate_tours(trips, max_nr_gaps=1, max_time="1d")
         # check that all times are below the max time
         for i, row in tours.iterrows():
             time_diff = row["finished_at"] - row["started_at"]
@@ -189,16 +202,16 @@ class TestGenerate_tours:
 
     def test_tour_geom(self, example_trip_data):
         """Test whether tour generation is invariant to the name of the geometry column"""
-        trips, sp_locs = example_trip_data
+        trips, _ = example_trip_data
         trips.rename(columns={"geom": "other_geom_name"}, inplace=True)
         trips = trips.set_geometry("other_geom_name")
-        trips_out, tours = ti.preprocessing.trips.generate_tours(trips)
+        trips_out, _ = ti.preprocessing.trips.generate_tours(trips)
         # check that nothing else than the new column has changed in trips df
         assert all(trips_out.iloc[:, :6] == trips)
 
     def test_tour_max_time(self, example_trip_data):
         """Test functionality of max time argument in tour generation"""
-        trips, sp_locs = example_trip_data
+        trips, _ = example_trip_data
         with pytest.warns(UserWarning, match="No tours can be generated, return empty tours"):
             _, tours = ti.preprocessing.trips.generate_tours(trips, max_time="2h")  # only 2 hours allowed
             assert len(tours) == 0
@@ -208,7 +221,7 @@ class TestGenerate_tours:
     def test_tours_locations(self, example_trip_data):
         """Test whether tour generation with locations as input yields correct results as well"""
         trips, sp_locs = example_trip_data
-        trips_out, tours = ti.preprocessing.trips.generate_tours(trips, staypoints=sp_locs, max_nr_gaps=1)
+        _, tours = ti.preprocessing.trips.generate_tours(trips, staypoints=sp_locs, max_nr_gaps=1)
         assert all(tours["location_id"] == pd.Series([1, 2, 2]))
 
         # group trips by tour and check that the locations of start and end of each tour are correct
@@ -263,12 +276,12 @@ class TestGenerate_tours:
 
     def test_print_progress_flag(self, example_trip_data, capsys):
         """Test if the print_progress bar controls the printing behavior."""
-        trips, sp_locs = example_trip_data
-        trips_out, tours = ti.preprocessing.trips.generate_tours(trips, print_progress=True)
+        trips, _ = example_trip_data
+        ti.preprocessing.trips.generate_tours(trips, print_progress=True)
         captured_print = capsys.readouterr()
         assert captured_print.err != ""
 
-        trips_out, tours = ti.preprocessing.trips.generate_tours(trips, print_progress=False)
+        ti.preprocessing.trips.generate_tours(trips, print_progress=False)
         captured_noprint = capsys.readouterr()
         assert captured_noprint.err == ""