Merge pull request #1800 from cmu-delphi/fix_runner

FlaSH Timeout Code in Runner not running FlaSH

Author: krivard

_delphi_utils_python/delphi_utils/flash_eval/eval_day.py

@@ -3,7 +3,8 @@
 import importlib
 import os
 from typing import Any, Callable, Dict, Optional
-import threading
+import multiprocessing
+import time
 from .archive import ArchiveDiffer, archiver_from_params
 from .logger import get_structured_logger
 from .utils import read_params, transfer_files, delete_move_files
@@ -66,13 +67,16 @@ def run_indicator_pipeline(indicator_fn:  Callable[[Params], None],
     validator = validator_fn(params)
     archiver = archiver_fn(params)
 
-    t = threading.Timer(timer, flash_fn, args=[params])
-    t.start()
-    t.join(timer)
-    if t.is_alive():
-        t.cancel()
-        t.join()
-
+    t1 = multiprocessing.Process(target=flash_fn, args=[params])
+    t1.start()
+    start = time.time()
+    while time.time()-start < timer:
+        if not t1.is_alive():
+            break
+        time.sleep(10)
+    else:
+        t1.terminate()
+        t1.join()
     if validator:
         validation_report = validator.validate()
         validation_report.log(logger)

_delphi_utils_python/delphi_utils/runner.py

@@ -12,9 +12,9 @@ def test_flash_input():
 
     lag = 1
     day = pd.to_datetime("1/1/2023", format="%m/%d/%Y")
-    signal = 'confirmed_incidence_num'
+    signal = './confirmed_incidence_num'
     params = {'flash': {'support':[0,1]}}
     initial_7_day_file = pd.read_csv(f'flash_ref/{signal}/last_7_1.csv',
                                                   index_col=0, parse_dates=[0], header=0)
-    flash_eval(lag, day, input_df, signal, params, logger=mock_logger)
+    last_7, type_of_outlier = flash_eval(lag, day, input_df, signal, params, logger=mock_logger, local=True)
     initial_7_day_file.to_csv(f'flash_ref/{signal}/last_7_1.csv')