added/updated tests

Author: aysim319

doctor_visits/delphi_doctor_visits/config.py

@@ -19,16 +19,17 @@ class Config:
     # data columns
     CLI_COLS = ["Covid_like", "Flu_like", "Mixed"]
     FLU1_COL = ["Flu1"]
-    COUNT_COLS = ["Denominator"] + FLU1_COL + CLI_COLS
+    COUNT_COLS = CLI_COLS + FLU1_COL + ["Denominator"]
     DATE_COL = "ServiceDate"
     GEO_COL = "PatCountyFIPS"
     AGE_COL = "PatAgeGroup"
     HRR_COLS = ["Pat HRR Name", "Pat HRR ID"]
-    ID_COLS = [DATE_COL] + [GEO_COL] + HRR_COLS + [AGE_COL]
-    FILT_COLS = ID_COLS + COUNT_COLS
     # as of 2020-05-11, input file expected to have 10 columns
     # id cols: ServiceDate, PatCountyFIPS, PatAgeGroup, Pat HRR ID/Pat HRR Name
     # value cols: Denominator, Covid_like, Flu_like, Flu1, Mixed
+    ID_COLS = [DATE_COL] + [GEO_COL] + HRR_COLS + [AGE_COL]
+    # drop HRR columns - unused for now since we assign HRRs by FIPS
+    FILT_COLS = [DATE_COL] + [GEO_COL] + [AGE_COL] + COUNT_COLS
     DTYPES = {
         "ServiceDate": str,
         "PatCountyFIPS": str,

doctor_visits/delphi_doctor_visits/input/SYNEDI_AGG_OUTPATIENT_18052020_1455CDT.csv.gz

@@ -51,7 +51,6 @@ def write_to_csv(output_df: pd.DataFrame, geo_level: str, se:bool, out_name: str
     logger.debug(f"wrote {out_n} rows for {geo_level}")
 
 
-#TODO clean the date params
 def csv_to_df(filepath: str, startdate: datetime, enddate: datetime, dropdate: datetime, logger) -> pd.DataFrame:
     '''
     Reads csv using Dask and filters out based on date range and currently unused column,
@@ -65,8 +64,9 @@ def csv_to_df(filepath: str, startdate: datetime, enddate: datetime, dropdate: d
 
     -------
     '''
-    filename = Path(filepath).name
-    logger.info(f"Processing {filename}")
+    filepath = Path(filepath)
+    logger.info(f"Processing {filepath}")
+
     ddata = dd.read_csv(
         filepath,
         compression="gzip",
@@ -75,7 +75,9 @@ def csv_to_df(filepath: str, startdate: datetime, enddate: datetime, dropdate: d
     )
 
     ddata = ddata.dropna()
+    # rename inconsistent column names to match config column names
     ddata = ddata.rename(columns=Config.DEVIANT_COLS_MAP)
+
     ddata = ddata[Config.FILT_COLS]
     ddata[Config.DATE_COL] = dd.to_datetime(ddata[Config.DATE_COL])
 
@@ -89,5 +91,11 @@ def csv_to_df(filepath: str, startdate: datetime, enddate: datetime, dropdate: d
     date_filter = ((ddata[Config.DATE_COL] >= Config.FIRST_DATA_DATE) & (ddata[Config.DATE_COL] < dropdate))
 
     df = ddata[date_filter].compute()
-    logger.info(f"Done processing {filename}")
+
+    # aggregate age groups (so data is unique by service date and FIPS)
+    df = df.groupby([Config.DATE_COL, Config.GEO_COL]).sum(numeric_only=True).reset_index()
+    assert np.sum(df.duplicated()) == 0, "Duplicates after age group aggregation"
+    assert (df[Config.COUNT_COLS] >= 0).all().all(), "Counts must be nonnegative"
+
+    logger.info(f"Done processing {filepath}")
     return df

doctor_visits/delphi_doctor_visits/process_data.py

@@ -15,8 +15,6 @@
 # third party
 import numpy as np
 import pandas as pd
-import dask.dataframe as dd
-
 
 # first party
 from delphi_utils import Weekday
@@ -42,10 +40,6 @@ def update_sensor(
       se: boolean to write out standard errors, if true, use an obfuscated name
       logger: the structured logger
     """
-    # aggregate age groups (so data is unique by service date and FIPS)
-    data = data.groupby([Config.DATE_COL, Config.GEO_COL]).sum(numeric_only=True).reset_index()
-    assert np.sum(data.duplicated()) == 0, "Duplicates after age group aggregation"
-    assert (data[Config.COUNT_COLS] >= 0).all().all(), "Counts must be nonnegative"
 
     drange = lambda s, e: np.array([s + timedelta(days=x) for x in range((e - s).days)])
     fit_dates = drange(Config.FIRST_DATA_DATE, dropdate)

doctor_visits/delphi_doctor_visits/update_sensor.py

@@ -0,0 +1,21 @@
+"""Tests for update_sensor.py."""
+from datetime import datetime
+import logging
+import pandas as pd
+
+from delphi_doctor_visits.process_data import csv_to_df
+
+TEST_LOGGER = logging.getLogger()
+
+class TestProcessData:
+    def test_csv_to_df(self):
+        actual = csv_to_df(
+            filepath="./test_data/SYNEDI_AGG_OUTPATIENT_07022020_1455CDT.csv.gz",
+            startdate=datetime(2020, 2, 4),
+            enddate=datetime(2020, 2, 5),
+            dropdate=datetime(2020, 2,6),
+            logger=TEST_LOGGER,
+        )
+
+        comparison = pd.read_pickle("./comparison/process_data/main_after_date_SYNEDI_AGG_OUTPATIENT_07022020_1455CDT.pkl")
+        pd.testing.assert_frame_equal(actual.reset_index(drop=True), comparison)

doctor_visits/tests/comparison/process_data/main_after_date_SYNEDI_AGG_OUTPATIENT_07022020_1455CDT.pkl

@@ -1,4 +1,5 @@
 """Tests for update_sensor.py."""
+from datetime import datetime
 import logging
 import pandas as pd
 
@@ -8,11 +9,12 @@
 
 class TestUpdateSensor:
     def test_update_sensor(self):
+        df = pd.read_pickle("./test_data/SYNEDI_AGG_OUTPATIENT_07022020_1455CDT.pkl")
         actual = update_sensor(
-            filepath="./test_data/SYNEDI_AGG_OUTPATIENT_07022020_1455CDT.csv.gz",
-            startdate="2020-02-04",
-            enddate="2020-02-05",
-            dropdate="2020-02-06",
+            data=df,
+            startdate=datetime(2020, 2, 4),
+            enddate=datetime(2020, 2, 5),
+            dropdate=datetime(2020, 2,6),
             geo="state",
             parallel=False,
             weekday=False,