Make parallel processing optional, avoids #93

Reviewed-on: https://codeberg.org/movingpandas/trajectools/pulls/99

Author: anitagraser

qgis_processing/extractPtsAlgorithm.py

@@ -25,9 +25,6 @@
 from .qgisUtils import feature_from_gdf_row
 
 
-CPU_COUNT = os.cpu_count()
-
-
 class ExtractODPtsAlgorithm(TrajectoriesAlgorithm):
     ORIGIN_PTS = "ORIGIN_PTS"
     DESTINATIONS_PTS = "DESTINATIONS_PTS"
@@ -206,7 +203,7 @@ def processTc(self, tc, parameters, context):
         min_duration = pd.Timedelta(min_duration).to_pytimedelta()
 
         try: 
-            gdf = TrajectoryStopDetector(tc, n_processes=CPU_COUNT).get_stop_points(
+            gdf = TrajectoryStopDetector(tc, n_processes=self.cpu_count).get_stop_points(
                 max_diameter=max_diameter, min_duration=min_duration
             )
         except TypeError:

qgis_processing/splitTrajectoriesAlgorithm.py

@@ -21,9 +21,6 @@
 from .trajectoriesAlgorithm import TrajectoryManipulationAlgorithm, help_str_base, help_str_traj
 
 
-CPU_COUNT = os.cpu_count()
-
-
 class SplitTrajectoriesAlgorithm(TrajectoryManipulationAlgorithm):
     def __init__(self):
         super().__init__()
@@ -90,12 +87,13 @@ def processTc(self, tc, parameters, context):
         if td_units == "Weeks": 
             td_units = "W"
         time_gap = pd.Timedelta(f"{time_gap} {td_units}").to_pytimedelta()
+
         for traj in tc.trajectories:
             try: 
                 splits = ObservationGapSplitter(traj).split(
                     gap=time_gap, 
                     min_length=tc.min_length, 
-                    n_processes=CPU_COUNT
+                    n_processes=self.cpu_count
                 )
             except TypeError:
                 raise TypeError("TypeError: cannot pickle 'QVariant' object. This error is usually caused by None values in input layer fields. Try to remove None values or run without Add movement metrics.")
@@ -151,7 +149,7 @@ def processTc(self, tc, parameters, context):
             splits = TemporalSplitter(tc).split(
                     mode=split_mode, 
                     min_length=tc.min_length, 
-                    n_processes=CPU_COUNT
+                    n_processes=self.cpu_count
                 )    
         except TypeError:
             raise TypeError("TypeError: cannot pickle 'QVariant' object. This error is usually caused by None values in input layer fields. Try to remove None values or run without Add movement metrics.")
@@ -212,7 +210,7 @@ def processTc(self, tc, parameters, context):
                 max_diameter=max_diameter, 
                 min_duration=min_duration, 
                 min_length=tc.min_length, 
-                n_processes=CPU_COUNT
+                n_processes=self.cpu_count
             )
         except TypeError:
             raise TypeError("TypeError: cannot pickle 'QVariant' object. This error is usually caused by None values in input layer fields. Try to remove None values or run without Add movement metrics.")
@@ -263,7 +261,7 @@ def processTc(self, tc, parameters, context):
                 splits = ValueChangeSplitter(traj).split(
                     col_name=self.field, 
                     min_length=tc.min_length, 
-                    n_processes=CPU_COUNT
+                    n_processes=self.cpu_count
                 )
             except TypeError:
                 raise TypeError("TypeError: cannot pickle 'QVariant' object. This error is usually caused by None values in input layer fields. Try to remove None values or run without Add movement metrics.")

qgis_processing/trajectoriesAlgorithm.py

@@ -40,8 +40,6 @@
     "a": 3600 * 24 * 365,
 }
 
-CPU_COUNT = os.cpu_count()
-
 help_str_base = (
     "<p><b>Trajectory ID field</b> is the input layer field containing the ID "
     "of the moving objects. If no field is specified, all input features are "
@@ -66,6 +64,7 @@ class TrajectoriesAlgorithm(QgsProcessingAlgorithm):
     TRAJ_ID_FIELD = "TRAJ_ID_FIELD"
     TIMESTAMP_FIELD = "TIME_FIELD"
     ADD_METRICS = "ADD_METRICS"
+    USE_PARALLEL_PROCESSING = "USE_PARALLEL_PROCESSING"
     SPEED_UNIT = "SPEED_UNIT"
     MIN_LENGTH = "MIN_LENGTH"
 
@@ -143,6 +142,11 @@ def prepare_parameters(self, parameters, context):
         ).split("/")
         self.min_length = self.parameterAsDouble(parameters, self.MIN_LENGTH, context)
         self.add_metrics = self.parameterAsBoolean(parameters, self.ADD_METRICS, context)
+        self.use_parallel = self.parameterAsBoolean(parameters, self.USE_PARALLEL_PROCESSING, context)
+        if self.use_parallel:
+            self.cpu_count = os.cpu_count()
+        else: 
+            self.cpu_count = 1
 
     def create_tc(self, parameters, context):
         self.prepare_parameters(parameters, context)
@@ -159,8 +163,8 @@ def create_tc(self, parameters, context):
 
         if self.add_metrics:
             try:
-                tc.add_speed(units=tuple(self.speed_units), overwrite=True, n_processes=CPU_COUNT)
-                tc.add_direction(overwrite=True, n_processes=CPU_COUNT)
+                tc.add_speed(units=tuple(self.speed_units), overwrite=True, n_processes=self.cpu_count)
+                tc.add_direction(overwrite=True, n_processes=self.cpu_count)
             except TypeError:  # None values cause TypeError: cannot pickle 'QVariant' object, see issue #93
                 raise TypeError("TypeError: cannot pickle 'QVariant' object. This error is usually caused by None values in input layer fields. Try to remove None values or run without Add movement metrics.")
 
@@ -226,7 +230,15 @@ def initAlgorithm(self, config=None):
                 defaultValue=True,
                 optional=False,
             )
-        )        
+        )   
+        self.addParameter(
+            QgsProcessingParameterBoolean(
+                name=self.USE_PARALLEL_PROCESSING,
+                description=self.tr("Use parallel processing (experimental)"),
+                defaultValue=False,
+                optional=False,
+            )
+        )       
         self.addParameter(
             QgsProcessingParameterString(
                 name=self.SPEED_UNIT,