Merge pull request #1693 from cmu-delphi/release/indicators_v0.3.23_utils_v0.3.6

Release covidcast-indicators 0.3.23

Author: krivard

.bumpversion.cfg

@@ -1,5 +1,5 @@
 [bumpversion]
-current_version = 0.3.22
+current_version = 0.3.23
 commit = True
 message = chore: bump covidcast-indicators to {new_version}
 tag = False

changehc/delphi_changehc/backfill.py

@@ -9,7 +9,9 @@
 from datetime import datetime
 # third party
 import pandas as pd
+from delphi_utils import GeoMapper
 
+gmpr = GeoMapper()
 
 def store_backfill_file(df, _end_date, backfill_dir, numtype, geo, weekday):
     """
@@ -35,17 +37,19 @@ def store_backfill_file(df, _end_date, backfill_dir, numtype, geo, weekday):
         return
 
     backfilldata = df.reset_index().copy()
+    backfilldata = gmpr.add_geocode(backfilldata, from_code="fips", new_code="state_id",
+                           from_col="fips", new_col="state_id")
     backfilldata.rename({"timestamp": "time_value"}, axis=1, inplace=True)
     #Store one year's backfill data
     _start_date = _end_date.replace(year=_end_date.year-1)
-    selected_columns = ['time_value', 'fips',
+    selected_columns = ['time_value', 'fips', 'state_id',
                         'num', 'den']
     backfilldata = backfilldata.loc[backfilldata["time_value"] >= _start_date,
                                     selected_columns]
     path = backfill_dir + \
         "/changehc_%s_as_of_%s.parquet"%(numtype, datetime.strftime(_end_date, "%Y%m%d"))
     # Store intermediate file into the backfill folder
-    backfilldata.to_parquet(path)
+    backfilldata.to_parquet(path, index=False)
 
 def merge_backfill_file(backfill_dir, numtype, geo, weekday, backfill_merge_day,
                         today, test_mode=False, check_nd=25):
@@ -83,6 +87,9 @@ def merge_backfill_file(backfill_dir, numtype, geo, weekday, backfill_merge_day,
 
     new_files = glob.glob(backfill_dir + "/changehc_%s_as_of_*"%numtype)
 
+    if len(new_files) == 0: # if no any daily file is stored
+        return
+
     def get_date(file_link):
         # Keep the function here consistent with the backfill path in
         # function `store_backfill_file`
@@ -111,7 +118,7 @@ def get_date(file_link):
         numtype,
         datetime.strftime(earliest_date, "%Y%m%d"),
         datetime.strftime(latest_date, "%Y%m%d"))
-    merged_file.to_parquet(path)
+    merged_file.to_parquet(path, index=False)
 
     # Delete daily files once we have the merged one.
     if not test_mode:

changehc/tests/backfill/.gitignore

@@ -0,0 +1 @@
+*.csv

changehc/tests/backfill/changehc_covid_as_of_20200601.parquet

@@ -68,7 +68,7 @@ def test_store_backfill_file(self):
         fn = "changehc_covid_as_of_20200101.parquet"
         backfill_df = pd.read_parquet(backfill_dir + "/"+ fn, engine='pyarrow')
 
-        selected_columns = ['time_value', 'fips',
+        selected_columns = ['time_value', 'fips', 'state_id',
                         'num', 'den']
         assert set(selected_columns) == set(backfill_df.columns)  
 
@@ -77,35 +77,51 @@ def test_store_backfill_file(self):
 
     def test_merge_backfill_file(self):
 
-        today = datetime.today()
         geo = "county"
         weekday = False
         numtype = "covid"
 
-        new_files = glob.glob(backfill_dir + "/changehc_%s*.parquet"%numtype)
+        today = datetime(2020, 6, 4)
         fn = "changehc_covid_from_20200601_to_20200604.parquet"
         assert fn not in os.listdir(backfill_dir)
 
+        merge_backfill_file(backfill_dir, numtype, geo, weekday, today.weekday(),
+                            today, test_mode=True, check_nd=2)         
+        assert fn not in os.listdir(backfill_dir)
+        
+        # Generate backfill daily files     
+        for d in range(1, 5):
+            dropdate = datetime(2020, 6, d)        
+            store_backfill_file(combined_data, dropdate, backfill_dir, \
+                                numtype, geo, weekday)
+
+        
         # Check the when the merged file is not generated
         today = datetime(2020, 6, 4)
         merge_backfill_file(backfill_dir, numtype, geo, weekday, 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
+        # Generate the merged file, but not delete it
         merge_backfill_file(backfill_dir, numtype, geo, weekday, today.weekday(),
                             today, test_mode=True, check_nd=2)         
         assert fn in os.listdir(backfill_dir)
 
         # Read daily file
+        new_files = glob.glob(backfill_dir + "/changehc_%s*.parquet"%numtype)
         pdList = []        
         for file in new_files:
+            if "from" in file:
+                continue
             df = pd.read_parquet(file, engine='pyarrow')
             issue_date = datetime.strptime(file[-16:-8], "%Y%m%d")
             df["issue_date"] = issue_date
             df["lag"] = [(issue_date - x).days for x in df["time_value"]]
             pdList.append(df)
-            
+            os.remove(file)
+        new_files = glob.glob(backfill_dir + "/changehc_%s*.parquet"%numtype)
+        assert len(new_files) == 1
+        
         expected = pd.concat(pdList).sort_values(["time_value", "fips"])
 
         # Read the merged file