only procesing once and passing along the dataframe

Author: aysim319

doctor_visits/delphi_doctor_visits/process_data.py

@@ -63,9 +63,7 @@ def csv_to_df(filepath: str, startdate: datetime, enddate: datetime, dropdate: d
       enddate: last sensor date (YYYY-mm-dd)
       dropdate: data drop date (YYYY-mm-dd)
 
-    Returns
     -------
-    cleaned dataframe
     '''
     filename = Path(filepath).name
     logger.info(f"Processing {filename}")

doctor_visits/delphi_doctor_visits/run.py

@@ -14,7 +14,8 @@
 from delphi_utils import get_structured_logger
 
 # first party
-from .update_sensor import update_sensor, write_to_csv
+from .update_sensor import update_sensor
+from .process_data import csv_to_df, write_to_csv
 from .download_claims_ftp_files import download
 from .get_latest_claims_name import get_latest_filename
 
@@ -85,6 +86,8 @@ def run_module(params):  # pylint: disable=too-many-statements
     ## geographies
     geos = ["state", "msa", "hrr", "county", "hhs", "nation"]
 
+    claims_df = csv_to_df(claims_file, startdate_dt, enddate_dt, dropdate_dt, logger)
+
     ## print out other vars
     logger.info("outpath:\t\t%s", export_dir)
     logger.info("parallel:\t\t%s", params["indicator"]["parallel"])
@@ -102,7 +105,7 @@ def run_module(params):  # pylint: disable=too-many-statements
             else:
                 logger.info("starting %s, no adj", geo)
             sensor = update_sensor(
-                filepath=claims_file,
+                data=claims_df,
                 startdate=startdate_dt,
                 enddate=enddate_dt,
                 dropdate=dropdate_dt,

doctor_visits/delphi_doctor_visits/update_sensor.py

@@ -15,24 +15,24 @@
 # third party
 import numpy as np
 import pandas as pd
+import dask.dataframe as dd
 
 
 # first party
 from delphi_utils import Weekday
 from .config import Config
 from .geo_maps import GeoMaps
-from .process_data import csv_to_df
 from .sensor import DoctorVisitsSensor
 
 
 def update_sensor(
-        filepath:str, startdate:datetime, enddate:datetime, dropdate:datetime, geo:str, parallel: bool,
+        data:pd.DataFrame, startdate:datetime, enddate:datetime, dropdate:datetime, geo:str, parallel: bool,
         weekday:bool, se:bool, logger
 ):
     """Generate sensor values.
 
     Args:
-      filepath: path to the aggregated doctor-visits data
+      data: dataframe of the cleaned claims file
       startdate: first sensor date (YYYY-mm-dd)
       enddate: last sensor date (YYYY-mm-dd)
       dropdate: data drop date (YYYY-mm-dd)
@@ -42,8 +42,6 @@ def update_sensor(
       se: boolean to write out standard errors, if true, use an obfuscated name
       logger: the structured logger
     """
-    data = csv_to_df(filepath, startdate, enddate, dropdate, 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"

doctor_visits/tests/teset_process_data.py

@@ -0,0 +1,24 @@
+"""Tests for update_sensor.py."""
+import logging
+import pandas as pd
+
+from delphi_doctor_visits.update_sensor import update_sensor
+
+TEST_LOGGER = logging.getLogger()
+
+class TestProcessData:
+    def test_csv_to_df(self):
+        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",
+            geo="state",
+            parallel=False,
+            weekday=False,
+            se=False,
+            logger=TEST_LOGGER,
+        )
+
+        comparison = pd.read_csv("./comparison/update_sensor/all.csv", parse_dates=["date"])
+        pd.testing.assert_frame_equal(actual.reset_index(drop=True), comparison)