nssp patching code

nssp/README.md

@@ -73,3 +73,33 @@ with the percentage of code covered by the tests.
 
 None of the linting or unit tests should fail, and the code lines that are not covered by unit tests should be small and
 should not include critical sub-routines. 
+
+## Running Patches:
+A daily backup of from source in the form of csv files can be found on `bigchunk-dev-02` under `/common/source_backup/nssp`. Talk to your sysadmin for access. 
+
+You can also generate your own backup from source by setting up a cron job that runs the following .py every day when a pipeline outtage is going on on our side but aource api is still available:
+```
+import numpy as np
+import pandas as pd
+from sodapy import Socrata
+from datetime import date
+
+today = date.today()
+socrata_token = 'FILL_YOUR_OWN_TOKEN_HERE'
+client = Socrata("data.cdc.gov", socrata_token)
+results = []
+offset = 0
+limit = 50000  # maximum limit allowed by SODA 2.0
+while True:
+    page = client.get("rdmq-nq56", limit=limit, offset=offset)
+    if not page:
+        break  # exit the loop if no more results
+    results.extend(page)
+    offset += limit
+df_ervisits = pd.DataFrame.from_records(results)
+df_ervisits.to_csv(f'~/{today}.csv', index=False)
+```
+When you're ready to create patching data for a specific date range in batch issue format, adjust `params.json` in accordance with instructions in `patch.py`, move the backup csv files into your chosen `source_dir`, then run
+```
+env/bin/python -m delphi_nssp.patch
+```

nssp/delphi_nssp/patch.py

@@ -0,0 +1,139 @@
+"""
+This module is used for patching data in the delphi_nssp package.
+
+To use this module, you need to turn on the custom_run flag
+and specify the range of issue dates in params.json, like so:
+
+{
+  "common": {
+    "custom_run": true,
+    ...
+  },
+  "validation": {
+    ...
+  },
+  "patch": {
+    "patch_dir": "/Users/minhkhuele/Desktop/delphi/covidcast-indicators/nssp/AprilPatch",
+    "start_issue": "2024-04-20",
+    "end_issue": "2024-04-21",
+    "source_dir": "/Users/minhkhuele/Desktop/delphi/covidcast-indicators/nssp/source_data"
+  }
+}
+
+It will generate data for that range of issue dates, and store them in batch issue format:
+[name-of-patch]/issue_[issue-date]/nssp/actual_data_file.csv
+"""
+
+import sys
+from datetime import datetime, timedelta
+from os import makedirs, path
+
+from delphi_utils import get_structured_logger, read_params
+from epiweeks import Week
+
+from .run import run_module
+
+
+def good_patch_config(params, logger):
+    """
+    Check if the params.json file is correctly configured for patching.
+
+    params: Dict[str, Any]
+        Nested dictionary of parameters, typically loaded from params.json file.
+    logger: Logger object
+        Logger object to log messages.
+    """
+    valid_config = True
+    custom_run = params["common"].get("custom_run", False)
+    if not custom_run:
+        logger.error("Calling patch.py without custom_run flag set true.")
+        valid_config = False
+
+    patch_config = params.get("patch", {})
+    if patch_config == {}:
+        logger.error("Custom flag is on, but patch section is missing.")
+        valid_config = False
+    else:
+        required_patch_keys = ["start_issue", "end_issue", "patch_dir", "source_dir"]
+        missing_keys = [key for key in required_patch_keys if key not in patch_config]
+        if missing_keys:
+            logger.error(f"Patch section is missing required key(s): {', '.join(missing_keys)}")
+            valid_config = False
+        else:
+            try:  # issue dates validity check
+                start_issue = datetime.strptime(patch_config["start_issue"], "%Y-%m-%d")
+                end_issue = datetime.strptime(patch_config["end_issue"], "%Y-%m-%d")
+                if start_issue > end_issue:
+                    logger.error("Start issue date is after end issue date.")
+                    valid_config = False
+            except ValueError:
+                logger.error("Issue dates must be in YYYY-MM-DD format.")
+                valid_config = False
+
+            if not path.isdir(patch_config["source_dir"]):
+                logger.error(f"Source directory {patch_config['source_dir']} does not exist.")
+                valid_config = False
+
+    if valid_config:
+        logger.info("Good patch configuration.")
+        return True
+    logger.info("Bad patch configuration.")
+    return False
+
+
+def patch():
+    """
+    Run the nssp indicator for a range of issue dates.
+
+    The range of issue dates is specified in params.json using the following keys:
+    - "patch": Only used for patching data
+        - "start_date": str, YYYY-MM-DD format, first issue date
+        - "end_date": str, YYYY-MM-DD format, last issue date
+        - "patch_dir": str, directory to write all issues output
+        - "source_dir": str, directory to read source data from.
+    """
+    params = read_params()
+    logger = get_structured_logger("delphi_nssp.patch", filename=params["common"]["log_filename"])
+    if not good_patch_config(params, logger):
+        sys.exit(1)
+
+    start_issue = datetime.strptime(params["patch"]["start_issue"], "%Y-%m-%d")
+    end_issue = datetime.strptime(params["patch"]["end_issue"], "%Y-%m-%d")
+
+    logger.info(f"""Start patching {params["patch"]["patch_dir"]}""")
+    logger.info(f"""Start issue: {start_issue.strftime("%Y-%m-%d")}""")
+    logger.info(f"""End issue: {end_issue.strftime("%Y-%m-%d")}""")
+    logger.info(f"""Source from: {params["patch"]["source_dir"]}""")
+    logger.info(f"""Output to: {params["patch"]["patch_dir"]}""")
+
+    makedirs(params["patch"]["patch_dir"], exist_ok=True)
+
+    current_issue = start_issue
+    while current_issue <= end_issue:
+        logger.info(f"""Running issue {current_issue.strftime("%Y-%m-%d")}""")
+
+        current_issue_source_csv = (
+            f"""{params.get("patch", {}).get("source_dir")}/{current_issue.strftime("%Y-%m-%d")}.csv"""
+        )
+        if not path.isfile(current_issue_source_csv):
+            logger.info(f"No source data at {current_issue_source_csv}")
+            current_issue += timedelta(days=1)
+            continue
+
+        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
+        # db matches with first date of epiweek that the reporting date falls in,
+        # rather than reporting date itself.
+        current_issue_date = Week.fromdate(current_issue).startdate()
+        current_issue_dir = f"""{params["patch"]["patch_dir"]}/issue_{current_issue_date.strftime("%Y%m%d")}/nssp"""
+        makedirs(f"{current_issue_dir}", exist_ok=True)
+        params["common"]["export_dir"] = f"""{current_issue_dir}"""
+
+        run_module(params, logger)
+        current_issue += timedelta(days=1)
+
+
+if __name__ == "__main__":
+    patch()

nssp/delphi_nssp/pull.py

@@ -27,7 +27,7 @@ def warn_string(df, type_dict):
     return warn
 
 
-def pull_nssp_data(socrata_token: str):
+def pull_nssp_data(socrata_token: str, issue_date: str = None, source_dir: str = None) -> pd.DataFrame:
     """Pull the latest NSSP ER visits data, and conforms it into a dataset.
 
     The output dataset has:
@@ -39,6 +39,11 @@ def pull_nssp_data(socrata_token: str):
     ----------
     socrata_token: str
         My App Token for pulling the NWSS data (could be the same as the nchs data)
+    issue_date: Optional[str]
+        (patching mode only) YYYY-MM-DD formatted date of the issue to pull data for
+    source_dir: Optional[str]
+        (patching mode only) Path to the directory containing the source data csv files
+        The files in source_dir are expected to be named yyyy-mm-dd.csv
     test_file: Optional[str]
         When not null, name of file from which to read test data
 
@@ -47,18 +52,21 @@ def pull_nssp_data(socrata_token: str):
     pd.DataFrame
         Dataframe as described above.
     """
-    # Pull data from Socrata API
-    client = Socrata("data.cdc.gov", socrata_token)
-    results = []
-    offset = 0
-    limit = 50000  # maximum limit allowed by SODA 2.0
-    while True:
-        page = client.get("rdmq-nq56", limit=limit, offset=offset)
-        if not page:
-            break  # exit the loop if no more results
-        results.extend(page)
-        offset += limit
-    df_ervisits = pd.DataFrame.from_records(results)
+    if not issue_date:
+        # Pull data from Socrata API
+        client = Socrata("data.cdc.gov", socrata_token)
+        results = []
+        offset = 0
+        limit = 50000  # maximum limit allowed by SODA 2.0
+        while True:
+            page = client.get("rdmq-nq56", limit=limit, offset=offset)
+            if not page:
+                break  # exit the loop if no more results
+            results.extend(page)
+            offset += limit
+        df_ervisits = pd.DataFrame.from_records(results)
+    else:
+        df_ervisits = pd.read_csv(f"{source_dir}/{issue_date}.csv")
     df_ervisits = df_ervisits.rename(columns={"week_end": "timestamp"})
     df_ervisits = df_ervisits.rename(columns=SIGNALS_MAP)
 

nssp/delphi_nssp/run.py

@@ -34,7 +34,6 @@
 from .constants import AUXILIARY_COLS, CSV_COLS, GEOS, SIGNALS
 from .pull import pull_nssp_data
 
-
 def add_needed_columns(df, col_names=None):
     """Short util to add expected columns not found in the dataset."""
     if col_names is None:
@@ -45,7 +44,6 @@ def add_needed_columns(df, col_names=None):
     df = add_default_nancodes(df)
     return df
 
-
 def logging(start_time, run_stats, logger):
     """Boilerplate making logs."""
     elapsed_time_in_seconds = round(time.time() - start_time, 2)
@@ -62,7 +60,7 @@ def logging(start_time, run_stats, logger):
     )
 
 
-def run_module(params):
+def run_module(params, logger=None):
     """
     Run the indicator.
 
@@ -72,18 +70,28 @@ def run_module(params):
         Nested dictionary of parameters.
     """
     start_time = time.time()
-    logger = get_structured_logger(
-        __name__,
-        filename=params["common"].get("log_filename"),
-        log_exceptions=params["common"].get("log_exceptions", True),
-    )
+    custom_run = params["common"].get("custom_run", False)
+    # logger doesn't exist yet means run_module is called from normal indicator run (instead of patching)
+    if not logger:
+        logger = get_structured_logger(
+            __name__,
+            filename=params["common"].get("log_filename"),
+            log_exceptions=params["common"].get("log_exceptions", True),
+        )
+        if custom_run:
+            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"]
     socrata_token = params["indicator"]["socrata_token"]
-
     run_stats = []
     ## 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
-    df_pull = pull_nssp_data(socrata_token)
+    if custom_run:
+        issue_date = params.get("patch", {}).get("current_issue", None)
+        source_dir = params.get("patch", {}).get("source_dir", None)
+        df_pull = pull_nssp_data(socrata_token, issue_date, source_dir)
+    else:
+        df_pull = pull_nssp_data(socrata_token)
     ## aggregate
     geo_mapper = GeoMapper()
     for signal in SIGNALS:

nssp/tests/source_dir/2021-01-02.csv

@@ -0,0 +1,2 @@
+week_end,geography,county,percent_visits_combined,percent_visits_covid,percent_visits_influenza,percent_visits_rsv,percent_visits_smoothed,percent_visits_smoothed_covid,percent_visits_smoothed_1,percent_visits_smoothed_rsv,ed_trends_covid,ed_trends_influenza,ed_trends_rsv,hsa,hsa_counties,hsa_nci_id,fips,trend_source
+2022-10-01T00:00:00.000,United States,All,2.84,1.84,0.48,0.55,2.83,2.07,0.34,0.44,Decreasing,Increasing,Increasing,All,All,All,0,United States