changed so only the latest day in epiweek gets patched in

Author: aysim319

nhsn/delphi_nhsn/patch.py

@@ -15,56 +15,44 @@
   }
 }
 
-It will generate data for the range of issue dates corresponding to source data files available in "backup_dir" specified under "common", and store them in batch issue format under "patch_dir":
+It will generate data for the range of issue dates corresponding to source data files available in "backup_dir"
+specified under "common", and store them in batch issue format under "patch_dir":
 [name-of-patch]/issue_[issue-date]/nhsn/actual_data_file.csv
 """
 
 from datetime import datetime
 from os import makedirs
 from pathlib import Path
+from typing import List
 
 from delphi_utils import get_structured_logger, read_params
 from epiweeks import Week
 
 from .run import run_module
 
 
-def group_source_files(source_files):
+def filter_source_files(source_files: List[Path]):
     """
-    Group patch files such that each lists contains unique epiweek issue date.
-
-    This allows for acquisitions break down patches files per unique epiweek
-    NHSN has not been updating their data in a consistent fashion
-    and in order to properly capture all the changes that happened, the patch files needs
-
+    Filter patch files such that each element in the list is an unique epiweek with the latest issue date.
 
     Parameters
     ----------
     source_files
 
     Returns
     -------
-    list of list of dates where the inner list represents issue dates with a corresponding weekday
-    the content of list contains issue date with the corresponding weekday
+    list of issue dates
 
-    ie:
-    [
-    [datetime.datetime(2024, 9, 9, 0, 0), datetime.datetime(2024, 11, 18, 0, 0)], # (weekday = 0)
-    [datetime.datetime(2024, 11, 20, 0, 0)] # (weekday = 2)
-    ]
-
-    the index may not represent the weekday integer if the sources files does not have issues dates for all 7 days
     """
-    days_in_week = 7
-    patch_list = [[] for _ in range(days_in_week)]
+    epiweek_dict = dict()
 
     for file in source_files:
         if "prelim" not in file.stem:
             current_issue_date = datetime.strptime(file.name.split(".")[0], "%Y%m%d")
-            weekday = current_issue_date.weekday()
-            patch_list[weekday].append(current_issue_date)
+            epiweek = Week.fromdate(current_issue_date)
+            epiweek_dict[epiweek] = file
 
-    filtered_patch_list = [lst for lst in patch_list if lst]
+    filtered_patch_list = list(epiweek_dict.values())
     return filtered_patch_list
 
 
@@ -74,41 +62,31 @@ def patch(params):
 
     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
     """
     logger = get_structured_logger("delphi_nhsn.patch", filename=params["common"]["log_filename"])
 
     source_files = sorted(Path(params["common"]["backup_dir"]).glob("*.csv.gz"))
-
-    patch_directory_prefix = params["patch"]["patch_dir"]
-    patch_list = group_source_files(source_files)
-    for idx, patch_dates in enumerate(patch_list):
-        start_issue = patch_dates[0]
-        end_issue = patch_dates[-1]
-
-        patch_directory = f"{patch_directory_prefix}_{idx}"
-        params["patch"]["patch_dir"] = patch_directory
-
-        logger.info(
-            "Starting patching",
-            patch_directory=patch_directory,
-            start_issue=start_issue.strftime("%Y-%m-%d"),
-            end_issue=end_issue.strftime("%Y-%m-%d"),
-        )
-
-        makedirs(patch_directory, exist_ok=True)
-
-        for issue_date in patch_dates:
-            current_issue_ew = Week.fromdate(issue_date)
-            logger.info("Running issue", issue_date=issue_date.strftime("%Y-%m-%d"))
-            params["patch"]["issue_date"] = issue_date.strftime("%Y%m%d")
-            current_issue_dir = f"{params['patch']['patch_dir']}/issue_{current_issue_ew}/nhsn"
-            makedirs(current_issue_dir, exist_ok=True)
-            params["common"]["export_dir"] = current_issue_dir
-            params["common"]["custom_run"] = True
-            run_module(params, logger)
+    makedirs(params["patch"]["patch_dir"], exist_ok=True)
+
+    logger.info(
+        "Starting patching",
+        patch_directory=params["patch"]["patch_dir"],
+        start_issue=source_files[0].name.split(".")[0],
+        end_issue=source_files[-1].name.split(".")[0],
+    )
+
+    patch_list = filter_source_files(source_files)
+    for file in patch_list:
+        issue_date = datetime.strptime(file.name.split(".")[0], "%Y%m%d")
+        current_issue_ew = Week.fromdate(issue_date)
+        logger.info("Running issue", issue_date=issue_date.strftime("%Y-%m-%d"))
+        params["patch"]["issue_date"] = issue_date.strftime("%Y%m%d")
+        current_issue_dir = f"{params['patch']['patch_dir']}/issue_{current_issue_ew}/nhsn"
+        makedirs(current_issue_dir, exist_ok=True)
+        params["common"]["export_dir"] = current_issue_dir
+        params["common"]["custom_run"] = True
+        run_module(params, logger)
 
 
 if __name__ == "__main__":

nhsn/tests/test_patch.py

@@ -1,6 +1,6 @@
 import glob
 import os
-import tempfile
+from collections import defaultdict
 from pathlib import Path
 import shutil
 from unittest.mock import patch as mock_patch
@@ -10,7 +10,7 @@
 
 from epiweeks import Week
 
-from delphi_nhsn.patch import group_source_files, patch
+from delphi_nhsn.patch import filter_source_files, patch
 from delphi_nhsn.constants import TOTAL_ADMISSION_COVID_API, TOTAL_ADMISSION_FLU_API
 from conftest import TEST_DATA, PRELIM_TEST_DATA, TEST_DIR
 
@@ -51,12 +51,18 @@ def generate_dummy_file_names(self):
             file_list.append(custom_filename)
         return file_list
 
-    def test_group_source_files(self):
+    def test_filter_source_files(self):
         filelist = self.generate_dummy_file_names()
-        processed_file_list = group_source_files(filelist)
-        for file_list in processed_file_list:
-            converted_file_list = [Week.fromdate(date) for date in file_list]
-            assert len(converted_file_list) == len(set(converted_file_list))
+        epiweek_dict = defaultdict(list)
+        for file in filelist:
+            issue_dt = datetime.strptime(file.name.split(".")[0], "%Y%m%d")
+            issue_epiweek = Week.fromdate(issue_dt)
+            epiweek_dict[issue_epiweek].append(issue_dt)
+        patch_issue_list = filter_source_files(filelist)
+        for file in patch_issue_list:
+            issue_dt = datetime.strptime(file.name.split(".")[0], "%Y%m%d")
+            issue_epiweek = Week.fromdate(issue_dt)
+            assert max(epiweek_dict[issue_epiweek]) == issue_dt
 
     def generate_test_source_files(self):
         start_date = datetime(2024, 8, 1)
@@ -102,19 +108,15 @@ def test_patch(self, params_w_patch):
             file_list, prelim_file_list = self.generate_test_source_files()
             patch(params_w_patch)
 
-        for idx in range(7):
-            patch_paths = [Path(dir) for dir in glob.glob(f"{TEST_DIR}/patch_dir_{idx}/*")]
-            for patch_path in patch_paths:
-                # epiweek + the index of the patch files should equal the issue date (which is set as the value of the csv)
-                issue_dt = Week.fromstring(patch_path.name.replace("issue_", "")).daydate(idx).strftime("%Y%m%d")
-                for patch_file in Path(patch_path / "nhsn").iterdir():
-                    df = pd.read_csv(str(patch_file))
-                    val = str(int(df["val"][0]))
-                    assert issue_dt == val
+            for issue_path in Path(f"{TEST_DIR}/patch_dir").glob("*"):
+                issue_dt_str = issue_path.name.replace("issue_", "")
+                for file in Path(issue_path / "nhsn").iterdir():
+                    df = pd.read_csv(file)
+                    val = Week.fromdate(datetime.strptime(str(int(df["val"][0])), "%Y%m%d"))
+                    assert issue_dt_str == str(val)
 
         # clean up
-        for idx in range(7):
-            shutil.rmtree(f"{TEST_DIR}/patch_dir_{idx}")
+        shutil.rmtree(f"{TEST_DIR}/patch_dir")
 
         for file in file_list:
             os.remove(file)