Fix multiprocessing in experiments and local models (#64)

* set multiprocessing start method to spawn 
* parallelize local models
* remove import

tests/test_benchmark.py

@@ -124,11 +124,15 @@ def test_2_benchmark_CV():
         test_percentage=0.1,
         num_folds=3,
         fold_overlap_pct=0.5,
-        num_processes=1,
+        num_processes=10,
     )
     results_summary_ol, results_train_ol, results_test_ol = benchmark_cv_overlap.run()
     log.debug("{}".format(results_summary_ol))
     log.info("#### test_2_benchmark_CV")
+    if PLOT:
+        air_passengers = results_summary[results_summary["split"] == "test"]
+        # air_passengers.plot(x="data", y="MASE", kind="barh")
+        plt.show()
 
 
 def test_2_benchmark_manual():

tests/test_models.py

@@ -330,7 +330,7 @@ def test_darts_model():
     print(results_test)
 
 
-def test_darts_local_model():
+def test_darts_StatsForecastAutoETS_model():
     air_passengers_df = pd.read_csv(AIR_FILE, nrows=NROWS)
     ercot_df_aux = pd.read_csv(ERCOT_FILE)
     ercot_df = pd.DataFrame()
@@ -350,11 +350,45 @@ def test_darts_local_model():
     model_classes_and_params = [
         (
             DartsLocalForecastingModel,
-            {"model": StatsForecastAutoARIMA, "lags": 12, "n_forecasts": 4},
+            {"model": StatsForecastAutoETS, "lags": 12, "n_forecasts": 1, "ETS_model": "ZZZ"},
         ),
+    ]
+    log.debug("{}".format(model_classes_and_params))
+
+    benchmark = SimpleBenchmark(
+        model_classes_and_params=model_classes_and_params,
+        datasets=dataset_list,
+        metrics=list(ERROR_FUNCTIONS.keys()),
+        test_percentage=0.25,
+        save_dir=SAVE_DIR,
+        num_processes=1,
+    )
+    results_train, results_test = benchmark.run()
+    log.info("#### test_darts_local_model")
+    print(results_test)
+
+
+def test_darts_StatsForecastAutoARIMA_model():
+    air_passengers_df = pd.read_csv(AIR_FILE, nrows=NROWS)
+    ercot_df_aux = pd.read_csv(ERCOT_FILE)
+    ercot_df = pd.DataFrame()
+    for region in ERCOT_REGIONS:
+        ercot_df = pd.concat(
+            (
+                ercot_df,
+                ercot_df_aux[ercot_df_aux["ID"] == region].iloc[:NROWS].copy(deep=True),
+            ),
+            ignore_index=True,
+        )
+
+    dataset_list = [
+        Dataset(df=air_passengers_df, name="air_passengers", freq="MS"),
+        Dataset(df=ercot_df, name="ercot_df", freq="H"),
+    ]
+    model_classes_and_params = [
         (
             DartsLocalForecastingModel,
-            {"model": StatsForecastAutoETS, "lags": 12, "n_forecasts": 4, "ETS_model": "ZZZ"},
+            {"model": StatsForecastAutoARIMA, "lags": 12, "n_forecasts": 1},
         ),
     ]
     log.debug("{}".format(model_classes_and_params))

tot/benchmark.py

@@ -4,7 +4,7 @@
 import os
 from abc import ABC
 from dataclasses import dataclass
-from multiprocessing.pool import Pool
+from multiprocessing import get_context
 from typing import List, Optional, Tuple
 
 import numpy as np
@@ -103,7 +103,7 @@ def run(self, verbose=True):
                 not all([exp.num_processes == 1 for exp in self.experiments]),
                 "Cannot set multiprocessing in " "Experiments and Benchmark.",
             )
-            with Pool(self.num_processes) as pool:
+            with get_context("spawn").Pool(self.num_processes) as pool:
                 args_list = [(exp, verbose, i + 1) for i, exp in enumerate(self.experiments)]
                 pool.map_async(
                     self._run_exp,

tot/experiment.py

@@ -4,7 +4,7 @@
 import time
 from abc import ABC, abstractmethod
 from dataclasses import dataclass
-from multiprocessing.pool import Pool
+from multiprocessing import get_context
 from typing import List, Optional
 
 import pandas as pd
@@ -442,7 +442,7 @@ def run(self):
         self.fcst_test = []
         time_start = time.time()
         if self.num_processes > 1 and self.num_folds > 1:
-            with Pool(self.num_processes) as pool:
+            with get_context("spawn").Pool(self.num_processes) as pool:
                 args = [
                     (df_train, df_test, current_fold, received_ID_column, received_single_time_series)
                     for current_fold, (df_train, df_test) in enumerate(folds)

tot/models/models_darts.py

@@ -1,6 +1,8 @@
+import gc
 import logging
 from copy import deepcopy
 from dataclasses import dataclass
+from multiprocessing import get_context
 from typing import Type
 
 import pandas as pd
@@ -163,6 +165,11 @@ def maybe_drop_added_values_from_df(self, predicted: pd.DataFrame, df: pd.DataFr
         return predicted
 
 
+def _fit(args):
+    model, series = args
+    model.fit(series)
+
+
 @dataclass
 class DartsLocalForecastingModel(DartsForecastingModel):
     """
@@ -240,11 +247,20 @@ def fit(self, df: pd.DataFrame, freq: str, ids_weights: dict) -> None:
         self.freq = freq
         self.model_params["season_length"] = FREQ_TO_SEASON_LENGTH[freq]
 
-        for df_name, df in df.groupby("ID"):
-            model = self.model(**self.model_params)
-            series = convert_df_to_TimeSeries(df, freq=self.freq)
-            model.fit(series)
-            self.models_list.append(model)
+        self.models_list = [self.model(**self.model_params) for _ in range(len(df.groupby("ID")))]
+
+        with get_context("spawn").Pool() as pool:
+            args = [
+                (model, convert_df_to_TimeSeries(df_i, freq=self.freq))
+                for model, (_, df_i) in zip(self.models_list, df.groupby("ID"))
+            ]
+            pool.map(
+                _fit,
+                args,
+            )
+            pool.close()
+            pool.join()
+        gc.collect()
 
     def predict(
         self, df: pd.DataFrame, received_single_time_series: bool, df_historic: pd.DataFrame = None