various bug fixes

Author: minhkhul

nssp/delphi_nssp/patch.py

@@ -41,7 +41,7 @@
 
 import sys
 from datetime import datetime, timedelta
-from os import listdir, makedirs, path
+from os import listdir, makedirs, path, getcwd
 from shutil import rmtree
 
 from delphi_utils import get_structured_logger, read_params
@@ -123,15 +123,15 @@ def patch():
 
     current_issue = start_issue
     while current_issue <= end_issue:
-        logger.info("patching issue", issue_date=current_issue.strftime("%Y-%m-%d"))
+        logger.info("patching issue", issue_date=current_issue.strftime("%Y%m%d"))
 
-        current_issue_source_csv = f"""{source_dir}/{current_issue.strftime("%Y-%m-%d")}.csv"""
+        current_issue_source_csv = f"""{source_dir}/{current_issue.strftime("%Y%m%d")}.csv.gz"""
         if not path.isfile(current_issue_source_csv):
             logger.info("No source data at this path", current_issue_source_csv=current_issue_source_csv)
             current_issue += timedelta(days=1)
             continue
 
-        params["patch"]["current_issue"] = current_issue.strftime("%Y-%m-%d")
+        params["patch"]["current_issue"] = current_issue.strftime("%Y%m%d")
 
         # current_issue_date can be different from params["patch"]["current_issue"]
         # due to weekly cadence of nssp data. For weekly sources, issue dates in our

nssp/delphi_nssp/pull.py

@@ -56,7 +56,7 @@ def get_source_data(params, logger):
 
     # Generate file names of source files to download
     dates = pd.date_range(start=params["patch"]["start_issue"], end=params["patch"]["end_issue"])
-    primary_source_files = [f"{date.strftime("%Y%m%d")}.csv.gz" for date in dates]
+    primary_source_files = [f"{date.strftime('%Y%m%d')}.csv.gz" for date in dates]
     secondary_source_files = [f"{date.strftime('%Y%m%d')}_secondary.csv.gz" for date in dates]
     remote_source_files = primary_source_files + secondary_source_files
 
@@ -79,6 +79,7 @@ def get_source_data(params, logger):
             logger.warning(
                 "Source backup for this date does not exist on the remote server.", missing_filename=remote_file_name
             )
+            continue
         sftp.get(remote_file_name, local_file_path, callback=callback_for_filename)
         logger.info("Transfer finished", remote_file_name=remote_file_name, local_file_path=local_file_path)
         num_files_transferred += 1
@@ -136,7 +137,7 @@ def pull_with_socrata_api(socrata_token: str, dataset_id: str):
 
 
 def pull_nssp_data(socrata_token: str, backup_dir: str, custom_run: bool, issue_date: Optional[str] = None, logger: Optional[logging.Logger] = None):
-    """Pull the latest NSSP ER visits primary dataset.
+    """Pull the NSSP ER visits primary dataset.
 
     https://data.cdc.gov/Public-Health-Surveillance/NSSP-Emergency-Department-Visit-Trajectories-by-St/rdmq-nq56/data_preview
 
@@ -213,7 +214,7 @@ def secondary_pull_nssp_data(
     elif custom_run and logger.name == "delphi_nssp.patch":
         if issue_date is None:
             raise ValueError("Issue date is required for patching")
-        source_filename = f"{backup_dir}/secondary_{issue_date}.csv.gz"
+        source_filename = f"{backup_dir}/{issue_date}_secondary.csv.gz"
         df_ervisits = pd.read_csv(source_filename)
         logger.info(
             "Number of records grabbed",

nssp/delphi_nssp/run.py

@@ -82,7 +82,10 @@ def run_module(params, logger=None):
             logger.warning("custom_run flag is on despite direct indicator run call. Normal indicator run continues.")
             custom_run = False
     export_dir = params["common"]["export_dir"]
-    backup_dir = params["common"]["backup_dir"]
+    if custom_run and logger.name == "delphi_nssp.patch":
+        backup_dir = params["patch"]["source_dir"]
+    else:
+        backup_dir = params["common"]["backup_dir"]
     custom_run = params["common"].get("custom_run", False)
     issue_date = params.get("patch", {}).get("current_issue", None)
     socrata_token = params["indicator"]["socrata_token"]
@@ -92,7 +95,7 @@ def run_module(params, logger=None):
     ## build the base version of the signal at the most detailed geo level you can get.
     ## compute stuff here or farm out to another function or file
 
-
+    # breakpoint()
     df_pull = pull_nssp_data(socrata_token, backup_dir, custom_run=custom_run, issue_date=issue_date, logger=logger)
 
     ## aggregate