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,87 @@
+"""
+This module is used for patching data in the delphi_nssp package.
+
+To use this module, you need to specify the range of issue dates in params.json, like so:
+
+{
+  "common": {
+    ...
+  },
+  "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
+"""
+
+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 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"])
+
+    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

@@ -62,7 +62,7 @@ def logging(start_time, run_stats, logger):
     )
 
 
-def run_module(params):
+def run_module(params, logger=None):
     """
     Run the indicator.
 
@@ -72,18 +72,21 @@ 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),
-    )
+    issue_date = params.get("patch", {}).get("current_issue", None)
+    source_dir = params.get("patch", {}).get("source_dir", None)
+    if not logger:
+        logger = get_structured_logger(
+            __name__,
+            filename=params["common"].get("log_filename"),
+            log_exceptions=params["common"].get("log_exceptions", True),
+        )
     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)
+    df_pull = pull_nssp_data(socrata_token, issue_date, source_dir)
     ## 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

nssp/tests/test_patch.py

@@ -0,0 +1,47 @@
+import unittest
+from unittest.mock import patch as mock_patch, call
+from delphi_nssp.patch import patch
+import os
+import shutil
+from datetime import datetime, timedelta
+
+class TestPatchModule(unittest.TestCase):
+    def test_patch(self):
+        with mock_patch('delphi_nssp.patch.run_module') as mock_run_module, \
+             mock_patch('delphi_nssp.patch.get_structured_logger') as mock_get_structured_logger, \
+             mock_patch('delphi_nssp.patch.read_params') as mock_read_params:
+
+            mock_read_params.return_value = {
+                "common": {
+                    "log_filename": "test.log"
+                },
+                "patch": {
+                    "start_issue": "2021-01-01",
+                    "end_issue": "2021-01-02",
+                    "patch_dir": "./patch_dir",
+                    "source_dir": "./source_dir"
+                }
+            }
+
+            patch()
+
+            self.assertIn('current_issue', mock_read_params.return_value['patch'])
+            self.assertEqual(mock_read_params.return_value['patch']['current_issue'], '2021-01-02')
+
+            self.assertTrue(os.path.isdir('./patch_dir'))
+            self.assertTrue(os.path.isdir('./patch_dir/issue_20201227/nssp'))
+
+            start_date = datetime(2020, 12, 28)
+            end_date = datetime(2021, 1, 3)
+            date = start_date
+
+            while date <= end_date:
+                date_str = date.strftime("%Y%m%d")
+                self.assertFalse(os.path.isdir(f'./patch_dir/issue_{date_str}/nssp'))
+                date += timedelta(days=1)
+
+            # Clean up the created directories after the test
+            shutil.rmtree(mock_read_params.return_value["patch"]["patch_dir"])
+
+if __name__ == '__main__':
+    unittest.main()