switch doctor visits to structured logger

Author: chinandrew

doctor_visits/delphi_doctor_visits/run.py

@@ -6,10 +6,11 @@
 """
 
 # standard packages
-import logging
 from datetime import datetime, timedelta
 from pathlib import Path
 
+from delphi_utils import get_structured_logger
+
 # first party
 from .update_sensor import update_sensor, write_to_csv
 
@@ -37,7 +38,9 @@ def run_module(params):
             - "obfuscated_prefix": str, prefix for signal name if write_se is True.
             - "parallel": bool, whether to update sensor in parallel.
     """
-    logging.basicConfig(level=logging.DEBUG)
+    logger = get_structured_logger(
+        __name__, filename=params["common"].get("log_filename"),
+        log_exceptions=params["common"].get("log_exceptions", True))
 
     ## get end date from input file
     # the filename is expected to be in the format:
@@ -61,30 +64,30 @@ def run_module(params):
     startdate_dt = enddate_dt - timedelta(days=n_backfill_days)
     enddate = str(enddate_dt.date())
     startdate = str(startdate_dt.date())
-    logging.info("drop date:\t\t%s", dropdate)
-    logging.info("first sensor date:\t%s", startdate)
-    logging.info("last sensor date:\t%s", enddate)
-    logging.info("n_backfill_days:\t%s", n_backfill_days)
-    logging.info("n_waiting_days:\t%s", n_waiting_days)
+    logger.info("drop date:\t\t%s", dropdate)
+    logger.info("first sensor date:\t%s", startdate)
+    logger.info("last sensor date:\t%s", enddate)
+    logger.info("n_backfill_days:\t%s", n_backfill_days)
+    logger.info("n_waiting_days:\t%s", n_waiting_days)
 
     ## geographies
     geos = ["state", "msa", "hrr", "county", "hhs", "nation"]
 
 
     ## print out other vars
-    logging.info("outpath:\t\t%s", export_dir)
-    logging.info("parallel:\t\t%s", params["indicator"]["parallel"])
-    logging.info("weekday:\t\t%s", params["indicator"]["weekday"])
-    logging.info("write se:\t\t%s", se)
-    logging.info("obfuscated prefix:\t%s", prefix)
+    logger.info("outpath:\t\t%s", export_dir)
+    logger.info("parallel:\t\t%s", params["indicator"]["parallel"])
+    logger.info("weekday:\t\t%s", params["indicator"]["weekday"])
+    logger.info("write se:\t\t%s", se)
+    logger.info("obfuscated prefix:\t%s", prefix)
 
     ## start generating
     for geo in geos:
         for weekday in params["indicator"]["weekday"]:
             if weekday:
-                logging.info("starting %s, weekday adj", geo)
+                logger.info("starting %s, weekday adj", geo)
             else:
-                logging.info("starting %s, no adj", geo)
+                logger.info("starting %s, no adj", geo)
             sensor = update_sensor(
                 filepath=params["indicator"]["input_file"],
                 startdate=startdate,
@@ -93,19 +96,20 @@ def run_module(params):
                 geo=geo,
                 parallel=params["indicator"]["parallel"],
                 weekday=weekday,
-                se=params["indicator"]["se"]
+                se=params["indicator"]["se"],
+                logger=logger,
             )
             if sensor is None:
-                logging.error("No sensors calculated, no output will be produced")
+                logger.error("No sensors calculated, no output will be produced")
                 continue
             # write out results
             out_name = "smoothed_adj_cli" if weekday else "smoothed_cli"
             if params["indicator"]["se"]:
                 assert prefix is not None, "template has no obfuscated prefix"
                 out_name = prefix + "_" + out_name
 
-            write_to_csv(sensor, geo, se, out_name, export_dir)
-            logging.debug(f"wrote files to {export_dir}")
-        logging.info("finished %s", geo)
+            write_to_csv(sensor, geo, se, out_name, logger, export_dir)
+            logger.debug(f"wrote files to {export_dir}")
+        logger.info("finished %s", geo)
 
-    logging.info("finished all")
+    logger.info("finished all")

doctor_visits/delphi_doctor_visits/sensor.py

@@ -6,9 +6,6 @@
 
 """
 
-# standard packages
-import logging
-
 # third party
 import numpy as np
 import pandas as pd
@@ -162,7 +159,8 @@ def fit(y_data,
             geo_id,
             recent_min_visits,
             min_recent_obs,
-            jeffreys):
+            jeffreys,
+            logger):
         """Fitting routine.
 
         Args:
@@ -217,7 +215,7 @@ def fit(y_data,
             # if all rates are zero, don't bother
             if code_vals.sum() == 0:
                 if jeffreys:
-                    logging.error("p is 0 even though we used Jefferys estimate")
+                    logger.error("p is 0 even though we used Jefferys estimate")
                 new_rates.append(np.zeros((n_dates,)))
                 continue
 
@@ -240,7 +238,7 @@ def fit(y_data,
         se[include] = np.sqrt(
             np.divide((new_rates[include] * (1 - new_rates[include])), den[include]))
 
-        logging.debug(f"{geo_id}: {new_rates[-1]:.3f},[{se[-1]:.3f}]")
+        logger.debug(f"{geo_id}: {new_rates[-1]:.3f},[{se[-1]:.3f}]")
 
         included_indices = [x for x in final_sensor_idxs if include[x]]
 

doctor_visits/delphi_doctor_visits/update_sensor.py

@@ -9,7 +9,6 @@
 """
 
 # standard packages
-import logging
 from datetime import timedelta
 from multiprocessing import Pool, cpu_count
 
@@ -24,7 +23,7 @@
 from .weekday import Weekday
 
 
-def write_to_csv(output_df: pd.DataFrame, geo_level, se, out_name, output_path="."):
+def write_to_csv(output_df: pd.DataFrame, geo_level, se, out_name, logger, output_path="."):
     """Write sensor values to csv.
 
     Args:
@@ -34,7 +33,7 @@ def write_to_csv(output_df: pd.DataFrame, geo_level, se, out_name, output_path="
       output_path: outfile path to write the csv (default is current directory)
     """
     if se:
-        logging.info(f"========= WARNING: WRITING SEs TO {out_name} =========")
+        logger.info(f"========= WARNING: WRITING SEs TO {out_name} =========")
 
     out_n = 0
     for d in set(output_df["date"]):
@@ -64,12 +63,12 @@ def write_to_csv(output_df: pd.DataFrame, geo_level, se, out_name, output_path="
                     outfile.write(
                         "%s,%f,%s,%s,%s\n" % (geo_id, sensor, "NA", "NA", "NA"))
                 out_n += 1
-    logging.debug(f"wrote {out_n} rows for {geo_level}")
+    logger.debug(f"wrote {out_n} rows for {geo_level}")
 
 
 def update_sensor(
         filepath, startdate, enddate, dropdate, geo, parallel,
-        weekday, se
+        weekday, se, logger
 ):
     """Generate sensor values.
 
@@ -82,6 +81,7 @@ def update_sensor(
       parallel: boolean to run the sensor update in parallel
       weekday: boolean to adjust for weekday effects
       se: boolean to write out standard errors, if true, use an obfuscated name
+      logger: the structured logger
     """
     # as of 2020-05-11, input file expected to have 10 columns
     # id cols: ServiceDate, PatCountyFIPS, PatAgeGroup, Pat HRR ID/Pat HRR Name
@@ -125,7 +125,7 @@ def update_sensor(
         (burn_in_dates >= startdate) & (burn_in_dates <= enddate))[0][:len(sensor_dates)]
 
     # handle if we need to adjust by weekday
-    params = Weekday.get_params(data) if weekday else None
+    params = Weekday.get_params(data, logger) if weekday else None
     if weekday and np.any(np.all(params == 0,axis=1)):
         # Weekday correction failed for at least one count type
         return None
@@ -155,13 +155,14 @@ def update_sensor(
                 geo_id,
                 Config.MIN_RECENT_VISITS,
                 Config.MIN_RECENT_OBS,
-                jeffreys
+                jeffreys,
+                logger
             )
             out.append(res)
 
     else:
         n_cpu = min(10, cpu_count())
-        logging.debug(f"starting pool with {n_cpu} workers")
+        logger.debug(f"starting pool with {n_cpu} workers")
 
         with Pool(n_cpu) as pool:
             pool_results = []
@@ -182,6 +183,7 @@ def update_sensor(
                             Config.MIN_RECENT_VISITS,
                             Config.MIN_RECENT_OBS,
                             jeffreys,
+                            logger
                         ),
                     )
                 )

doctor_visits/delphi_doctor_visits/weekday.py

@@ -4,8 +4,7 @@
 Created: 2020-05-06
 """
 
-# standard packages
-import logging
+
 
 # third party
 import cvxpy as cp
@@ -19,7 +18,7 @@ class Weekday:
     """Class to handle weekday effects."""
 
     @staticmethod
-    def get_params(data):
+    def get_params(data, logger):
         r"""Correct a signal estimated as numerator/denominator for weekday effects.
 
         The ordinary estimate would be numerator_t/denominator_t for each time point
@@ -98,7 +97,7 @@ def get_params(data):
                     pass
             else:
                 # Leaving params[i,:] = 0 is equivalent to not performing weekday correction
-                logging.error("Unable to calculate weekday correction")
+                logger.error("Unable to calculate weekday correction")
 
         return params
 

doctor_visits/tests/test_update_sensor.py

@@ -1,9 +1,11 @@
 """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 TestUpdateSensor:
     def test_update_sensor(self):
         actual = update_sensor(
@@ -14,7 +16,8 @@ def test_update_sensor(self):
             geo="state",
             parallel=False,
             weekday=False,
-            se=False
+            se=False,
+            logger=TEST_LOGGER,
         )
 
         comparison = pd.read_csv("./comparison/update_sensor/all.csv", parse_dates=["date"])