making backfill to monthly in progess

Author: aysim319

changehc/delphi_changehc/patch.py

@@ -10,11 +10,12 @@
 import os
 import re
 import shutil
-from datetime import datetime
+from datetime import datetime, timedelta
 from typing import Union
 
 # third party
 import pandas as pd
+import pytz
 from delphi_utils import GeoMapper
 
 from .config import Config
@@ -94,11 +95,12 @@ def merge_existing_backfill_files(backfill_dir, backfill_file, issue_date, logge
 
     def get_file_with_date(files) -> Union[str, None]:
         for filename in files:
-            pattern = re.findall(r"\d{8}", filename)
-            if len(pattern) == 2:
-                start_date = datetime.strptime(pattern[0], "%Y%m%d")
-                end_date = datetime.strptime(pattern[1], "%Y%m%d")
-                if start_date <= issue_date or end_date <= issue_date:
+            pattern = re.findall(r"\d{6}", filename)
+            if len(pattern) == 1:
+                file_month = datetime.strptime(pattern[0], "%Y%m")
+                start_date = file_month.replace(day=1)
+                end_date = (start_date + timedelta(days=32)).replace(day=1)
+                if issue_date >= start_date and issue_date < end_date:
                     return filename
         return ""
 
@@ -125,27 +127,17 @@ def get_file_with_date(files) -> Union[str, None]:
     return
 
 
-def merge_backfill_file(backfill_dir, backfill_merge_day, most_recent, test_mode=False, check_nd=25):
+def merge_backfill_file(backfill_dir, most_recent, logger, test_mode=False):
     """
-    Merge ~4 weeks' backfill data into one file.
+    Merge a month's source data into one file.
 
-    Usually this function should merge 28 days' data into a new file so as to
-    save the reading time when running the backfill pipelines. We set a softer
-    threshold to allow flexibility in data delivery.
     Parameters
     ----------
     most_recent : datetime
         The most recent date when the raw data is received
     backfill_dir : str
         specified path to store backfill files.
-    backfill_merge_day: int
-        The day of a week that we used to merge the backfill files. e.g. 0
-        is Monday.
     test_mode: bool
-    check_nd: int
-        The criteria of the number of unmerged files. Ideally, we want the
-        number to be 28, but we use a looser criteria from practical
-        considerations
     """
     new_files = glob.glob(backfill_dir + "/claims_hosp_as_of_*")
     if len(new_files) == 0: # if no any daily file is stored
@@ -158,23 +150,19 @@ def get_date(file_link):
         return datetime.strptime(fn, "%Y%m%d")
 
     date_list = list(map(get_date, new_files))
-    earliest_date = min(date_list)
     latest_date = max(date_list)
-
-    # Check whether to merge
-    # Check the number of files that are not merged
-    if most_recent.weekday() != backfill_merge_day or (most_recent - earliest_date).days <= check_nd:
+    if latest_date.month == most_recent.month:
+        logger.info("Not a new month; skipping merging")
         return
 
+
     # Start to merge files
     pdList = []
     for fn in new_files:
         df = pd.read_parquet(fn, engine='pyarrow')
         pdList.append(df)
     merged_file = pd.concat(pdList).sort_values(["time_value", "fips"])
-    path = backfill_dir + "/claims_hosp_from_%s_to_%s.parquet"%(
-        datetime.strftime(earliest_date, "%Y%m%d"),
-        datetime.strftime(latest_date, "%Y%m%d"))
+    path = f"{backfill_dir}/claims_hosp_{datetime.strftime(latest_date, '%Y%m')}.parquet"
     merged_file.to_parquet(path, index=False)
 
     # Delete daily files once we have the merged one.

claims_hosp/delphi_claims_hosp/backfill.py

@@ -109,7 +109,7 @@ def run_module(params, logger=None):
             merge_existing_backfill_files(backfill_dir, backfilled_filepath, issue_date, logger)
 
         else:
-            merge_backfill_file(backfill_dir, backfill_merge_day, datetime.today())
+            merge_backfill_file(backfill_dir, datetime.today())
             store_backfill_file(claims_file, dropdate_dt, backfill_dir)
 
     # print out information

claims_hosp/delphi_claims_hosp/run.py

@@ -3,7 +3,6 @@
 import glob
 from datetime import datetime
 from pathlib import Path
-import shutil
 
 # third party
 import pandas as pd
@@ -49,32 +48,24 @@ def test_store_backfill_file(self):
         assert fn not in os.listdir(backfill_dir)
 
     def test_merge_backfill_file(self):
-        
-        today = datetime.today()
-        
-        fn = "claims_hosp_from_20200611_to_20200614.parquet"
+        fn = "claims_hosp_202006.parquet"
         assert fn not in os.listdir(backfill_dir)
 
         # Check when there is no daily file to merge.
         today = datetime(2020, 6, 14)
-        merge_backfill_file(backfill_dir, today.weekday(), today, 
-                            test_mode=True, check_nd=8)
+        merge_backfill_file(backfill_dir, today, TEST_LOGGER,
+                            test_mode=True)
         assert fn not in os.listdir(backfill_dir)
 
         # Generate backfill daily files     
         for d in range(11, 15):
             dropdate = datetime(2020, 6, d)        
             store_backfill_file(DATA_FILEPATH, dropdate, backfill_dir)
 
-        # Check the when the merged file is not generated
-        today = datetime(2020, 6, 14)
-        merge_backfill_file(backfill_dir, today.weekday(), today, 
-                            test_mode=True, check_nd=8)
-        assert fn not in os.listdir(backfill_dir)
-        
-         # Generate the merged file, but not delete it
-        merge_backfill_file(backfill_dir, today.weekday(), today, 
-                            test_mode=True, check_nd=2)         
+        # Check when the merged file is not generated
+        today = datetime(2020, 7, 1)
+        merge_backfill_file(backfill_dir, today, TEST_LOGGER,
+                            test_mode=True)
         assert fn in os.listdir(backfill_dir)
 
         # Read daily file
@@ -112,15 +103,15 @@ def prep_backfill_data():
 
             today = datetime(2020, 6, 14)
             # creating expected file
-            merge_backfill_file(backfill_dir, today.weekday(), today,
-                                test_mode=True, check_nd=2)
+            merge_backfill_file(backfill_dir, today, TEST_LOGGER,
+                                test_mode=True)
             original = f"{backfill_dir}/claims_hosp_from_20200611_to_20200614.parquet"
             os.rename(original, f"{backfill_dir}/expected.parquet")
 
             # creating backfill without issue date
             os.remove(f"{backfill_dir}/claims_hosp_as_of_{issue_date_str}.parquet")
             today = datetime(2020, 6, 14)
-            merge_backfill_file(backfill_dir, today.weekday(), today,
+            merge_backfill_file(backfill_dir, today,
                                 test_mode=True, check_nd=2)
 
             old_files = glob.glob(backfill_dir + "/claims_hosp_as_of_*")
@@ -141,7 +132,7 @@ def prep_backfill_data():
 
 
     def test_merge_existing_backfill_files_no_call(self):
-        issue_date = datetime(year=2020, month=6, day=20)
+        issue_date = datetime(year=2020, month=5, day=20)
         issue_date_str = issue_date.strftime("%Y%m%d")
         def prep_backfill_data():
             # Generate backfill daily files
@@ -151,8 +142,8 @@ def prep_backfill_data():
 
             today = datetime(2020, 6, 14)
             # creating expected file
-            merge_backfill_file(backfill_dir, today.weekday(), today,
-                                test_mode=True, check_nd=8)
+            merge_backfill_file(backfill_dir, today, TEST_LOGGER,
+                                test_mode=True)
 
         prep_backfill_data()
         file_to_add = store_backfill_file(DATA_FILEPATH, issue_date, backfill_dir)