Fix unstructured logging variable formatting

changehc/delphi_changehc/run.py

@@ -25,7 +25,7 @@ def retrieve_files(params, filedate, logger):
     if files["denom"] is None:
 
         ## download recent files from FTP server
-        logger.info("downloading recent files through SFTP")
+        logger.info("Downloading recent files through SFTP")
         download_counts(filedate, params["indicator"]["input_cache_dir"], params["indicator"]["ftp_conn"])
 
         denom_file = "%s/%s_Counts_Products_Denom.dat.gz" % (params["indicator"]["input_cache_dir"],filedate)
@@ -157,28 +157,30 @@ def run_module(params: Dict[str, Dict[str, Any]]):
 
     startdate, enddate = process_dates(params, startdate_dt, enddate_dt)
 
-    logger.info("generating signal and exporting to CSV",
-        first_sensor_date = startdate,
-        last_sensor_date = enddate,
-        drop_date = dropdate,
-        n_backfill_days = n_backfill_days,
-        n_waiting_days = n_waiting_days,
-        geos = params["indicator"]["geos"],
-        export_dir = params["common"]["export_dir"],
-        parallel = params["indicator"]["parallel"],
-        weekday = params["indicator"]["weekday"],
-        types = params["indicator"]["types"],
-        se = params["indicator"]["se"])
+    logger.info(
+        "Generating signal and exporting to CSV",
+        first_sensor_date=startdate,
+        last_sensor_date=enddate,
+        drop_date=dropdate,
+        n_backfill_days=n_backfill_days,
+        n_waiting_days=n_waiting_days,
+        geos=params["indicator"]["geos"],
+        export_dir=params["common"]["export_dir"],
+        parallel=params["indicator"]["parallel"],
+        weekday=params["indicator"]["weekday"],
+        types=params["indicator"]["types"],
+        se=params["indicator"]["se"],
+    )
 
     ## start generating
     stats = []
     for geo in params["indicator"]["geos"]:
         for numtype in params["indicator"]["types"]:
             for weekday in params["indicator"]["weekday"]:
                 if weekday:
-                    logger.info("starting weekday adj", geo = geo, numtype = numtype)
+                    logger.info("Starting weekday adj", geo=geo, numtype=numtype)
                 else:
-                    logger.info("starting no adj", geo = geo, numtype = numtype)
+                    logger.info("Starting no adj", geo=geo, numtype=numtype)
                 su_inst = CHCSensorUpdater(
                     startdate,
                     enddate,
@@ -211,7 +213,7 @@ def run_module(params: Dict[str, Dict[str, Any]]):
                 )
                 stats.extend(more_stats)
 
-            logger.info("finished processing", geo = geo)
+            logger.info("Finished processing", geo=geo)
 
     elapsed_time_in_seconds = round(time.time() - start_time, 2)
     min_max_date = stats and min(s[0] for s in stats)

changehc/delphi_changehc/update_sensor.py

@@ -41,7 +41,7 @@ def write_to_csv(df, geo_level, write_se, day_shift, out_name, logger, output_pa
     assert df[suspicious_se_mask].empty, " se contains suspiciously large values"
     assert not df["se"].isna().any(), " se contains nan values"
     if write_se:
-        logger.info("========= WARNING: WRITING SEs TO {0} =========".format(out_name))
+        logger.info("WARNING: WRITING SEs", filename=out_name)
     else:
         df["se"] = np.nan
 

claims_hosp/delphi_claims_hosp/download_claims_ftp_files.py

@@ -57,7 +57,7 @@ def download(ftp_credentials, out_path, logger):
     """Pull the latest raw files."""
     current_time = datetime.datetime.now()
     seconds_in_day = 24 * 60 * 60
-    logger.info("starting download", time=current_time)
+    logger.info("Starting download")
 
     # open client
     client = paramiko.SSHClient()

claims_hosp/delphi_claims_hosp/modify_claims_drops.py

@@ -57,5 +57,5 @@ def modify_and_write(data_path, logger, test_mode=False):
             dfs_list.append(dfs)
         else:
             dfs.to_csv(out_path, index=False)
-            logger.info(f"Wrote {out_path}")
+            logger.info("Wrote modified csv", filename=out_path)
     return files, dfs_list

claims_hosp/delphi_claims_hosp/run.py

@@ -116,9 +116,9 @@ def run_module(params):
     for geo in params["indicator"]["geos"]:
         for weekday in params["indicator"]["weekday"]:
             if weekday:
-                logger.info("starting weekday adj", geo = geo)
+                logger.info("Starting weekday adj", geo=geo)
             else:
-                logger.info("starting no weekday adj", geo =  geo)
+                logger.info("Starting no weekday adj", geo=geo)
 
             signal_name = Config.signal_weekday_name if weekday else Config.signal_name
             if params["indicator"]["write_se"]:
@@ -135,16 +135,16 @@ def run_module(params):
                 params["indicator"]["parallel"],
                 weekday,
                 params["indicator"]["write_se"],
-                signal_name
+                signal_name,
+                logger,
             )
             updater.update_indicator(
                 claims_file,
                 params["common"]["export_dir"],
-                logger,
             )
             max_dates.append(updater.output_dates[-1])
             n_csv_export.append(len(updater.output_dates))
-        logger.info("finished updating", geo = geo)
+        logger.info("Finished updating", geo=geo)
 
     # Remove all the raw files
     for fn in os.listdir(params["indicator"]["input_dir"]):

claims_hosp/delphi_claims_hosp/update_indicator.py

@@ -7,7 +7,6 @@
 """
 
 # standard packages
-import logging
 from multiprocessing import Pool, cpu_count
 
 # third party
@@ -28,8 +27,7 @@ class ClaimsHospIndicatorUpdater:
     # pylint: disable=too-many-instance-attributes, too-many-arguments
     # all variables are used
 
-    def __init__(self, startdate, enddate, dropdate, geo, parallel, weekday,
-                 write_se, signal_name):
+    def __init__(self, startdate, enddate, dropdate, geo, parallel, weekday, write_se, signal_name, logger):
         """
         Initialize updater for the claims-based hospitalization indicator.
 
@@ -53,6 +51,7 @@ def __init__(self, startdate, enddate, dropdate, geo, parallel, weekday,
         # init in shift_dates, declared here for pylint
         self.burnindate, self.fit_dates, self.burn_in_dates, self.output_dates = \
             [None] * 4
+        self.logger = logger
 
         assert (
                 self.startdate > (Config.FIRST_DATA_DATE + Config.BURN_IN_PERIOD)
@@ -114,9 +113,9 @@ def geo_reindex(self, data):
         elif self.geo == "hrr":
             data_frame = data  # data is already adjusted in aggregation step above
         else:
-            logging.error(
-                "%s is invalid, pick one of 'county', 'state', 'msa', 'hrr', 'hhs', nation'",
-                self.geo)
+            self.logger.error(
+                "geo is invalid, pick one of 'county', 'state', 'msa', 'hrr', 'hhs', nation'", geo=self.geo
+            )
             return False
 
         unique_geo_ids = pd.unique(data_frame[self.geo])
@@ -133,7 +132,7 @@ def geo_reindex(self, data):
         data_frame.fillna(0, inplace=True)
         return data_frame
 
-    def update_indicator(self, input_filepath, outpath, logger):
+    def update_indicator(self, input_filepath, outpath):
         """
         Generate and output indicator values.
 
@@ -159,7 +158,7 @@ def update_indicator(self, input_filepath, outpath, logger):
                 ["num"],
                 Config.DATE_COL,
                 [1, 1e5],
-                logger,
+                self.logger,
             )
             if self.weekday
             else None
@@ -182,7 +181,7 @@ def update_indicator(self, input_filepath, outpath, logger):
                 valid_inds[geo_id] = np.array(res.loc[final_output_inds, "incl"])
         else:
             n_cpu = min(Config.MAX_CPU_POOL, cpu_count())
-            logging.debug("starting pool with %d workers", n_cpu)
+            self.logger.debug("Starting pool", n_workers=n_cpu)
             with Pool(n_cpu) as pool:
                 pool_results = []
                 for geo_id, sub_data in data_frame.groupby(level=0, as_index=False):
@@ -217,7 +216,7 @@ def update_indicator(self, input_filepath, outpath, logger):
         }
 
         self.write_to_csv(output_dict, outpath)
-        logging.debug("wrote files to %s", outpath)
+        self.logger.debug("Wrote files", directory=outpath)
 
     def write_to_csv(self, output_dict, output_path="./receiving"):
         """
@@ -229,8 +228,7 @@ def write_to_csv(self, output_dict, output_path="./receiving"):
 
         """
         if self.write_se:
-            logging.info("========= WARNING: WRITING SEs TO %s =========",
-                         self.signal_name)
+            self.logger.info("WARNING: WRITING SEs", signal=self.signal_name)
 
         geo_level = output_dict["geo_level"]
         dates = output_dict["dates"]
@@ -255,7 +253,7 @@ def write_to_csv(self, output_dict, output_path="./receiving"):
                         assert not np.isnan(val), "value for included value is nan"
                         assert not np.isnan(se), "se for included rate is nan"
                         if val > 90:
-                            logging.warning("value suspicious, %s: %d", geo_id, val)
+                            self.logger.warning("value suspicious", geo=geo_id, value=val)
                         assert se < 5, f"se suspicious, {geo_id}: {se}"
                         if self.write_se:
                             assert val > 0 and se > 0, "p=0, std_err=0 invalid"
@@ -267,4 +265,4 @@ def write_to_csv(self, output_dict, output_path="./receiving"):
                                 "%s,%f,%s,%s,%s\n" % (geo_id, val, "NA", "NA", "NA"))
                         out_n += 1
 
-        logging.debug("wrote %d rows for %d %s", out_n, len(geo_ids), geo_level)
+        self.logger.debug("Wrote rows", num_rows=out_n, num_geo_ids=len(geo_ids), geo_level=geo_level)

doctor_visits/delphi_doctor_visits/download_claims_ftp_files.py

@@ -59,7 +59,7 @@ def download(ftp_credentials, out_path, logger, issue_date=None):
     else:
         current_time = datetime.datetime.strptime(issue_date, "%Y-%m-%d").replace(hour=23, minute=59, second=59)
 
-    logger.info("starting download", time=current_time)
+    logger.info("Starting download")
     seconds_in_day = 24 * 60 * 60
 
     # open client

doctor_visits/delphi_doctor_visits/modify_claims_drops.py

@@ -48,5 +48,5 @@ def modify_and_write(f, logger, test_mode=False):
 
     if not test_mode:
         dfs.to_csv(out_path, index=False)
-        logger.info(f"Wrote {out_path}")
+        logger.info("Wrote modified csv", filename=out_path)
     return dfs

doctor_visits/delphi_doctor_visits/patch.py

@@ -45,16 +45,19 @@ def patch():
     start_issue = datetime.strptime(params["patch"]["start_issue"], "%Y-%m-%d")
     end_issue = datetime.strptime(params["patch"]["end_issue"], "%Y-%m-%d")
 
-    logger.info(f"""Start patching {params["patch"]["patch_dir"]}""")
-    logger.info(f"""Start issue: {start_issue.strftime("%Y-%m-%d")}""")
-    logger.info(f"""End issue: {end_issue.strftime("%Y-%m-%d")}""")
+    logger.info(
+        "Starting patching",
+        patch_directory=params["patch"]["patch_dir"],
+        start_issue=start_issue.strftime("%Y-%m-%d"),
+        end_issue=end_issue.strftime("%Y-%m-%d"),
+    )
 
     makedirs(params["patch"]["patch_dir"], exist_ok=True)
 
     current_issue = start_issue
 
     while current_issue <= end_issue:
-        logger.info(f"""Running issue {current_issue.strftime("%Y-%m-%d")}""")
+        logger.info("Running issue", issue_date=current_issue.strftime("%Y-%m-%d"))
 
         params["patch"]["current_issue"] = current_issue.strftime("%Y-%m-%d")
 

doctor_visits/delphi_doctor_visits/run.py

@@ -88,32 +88,33 @@ def run_module(params, logger=None):  # pylint: disable=too-many-statements
     startdate_dt = enddate_dt - timedelta(days=n_backfill_days)
     enddate = str(enddate_dt.date())
     startdate = str(startdate_dt.date())
-    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)
+
+    logger.info(
+        "Using params",
+        startdate=startdate,
+        enddate=enddate,
+        dropdate=dropdate,
+        n_backfill_days=n_backfill_days,
+        n_waiting_days=n_waiting_days,
+        outpath=export_dir,
+        parallel=params["indicator"]["parallel"],
+        weekday=params["indicator"]["weekday"],
+        write_se=se,
+        prefix=prefix,
+    )
 
     ## geographies
     geos = ["state", "msa", "hrr", "county", "hhs", "nation"]
 
-
-    ## print out other vars
-    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)
-
     max_dates = []
     n_csv_export = []
     ## start generating
     for geo in geos:
         for weekday in params["indicator"]["weekday"]:
             if weekday:
-                logger.info("starting %s, weekday adj", geo)
+                logger.info("Starting with weekday adj", geo=geo)
             else:
-                logger.info("starting %s, no adj", geo)
+                logger.info("Starting with no adj", geo=geo)
             sensor = update_sensor(
                 filepath=claims_file,
                 startdate=startdate,
@@ -137,8 +138,8 @@ def run_module(params, logger=None):  # pylint: disable=too-many-statements
             write_to_csv(sensor, geo, se, out_name, logger, export_dir)
             max_dates.append(sensor.date.max())
             n_csv_export.append(sensor.date.unique().shape[0])
-            logger.debug(f"wrote files to {export_dir}")
-        logger.info("finished updating", geo = geo)
+            logger.debug("Wrote files", directory=export_dir)
+        logger.info("Finished updating", geo=geo)
 
     # Remove all the raw files
     for fn in os.listdir(params["indicator"]["input_dir"]):

doctor_visits/delphi_doctor_visits/update_sensor.py

@@ -34,7 +34,7 @@ def write_to_csv(output_df: pd.DataFrame, geo_level, se, out_name, logger, outpu
       output_path: outfile path to write the csv (default is current directory)
     """
     if se:
-        logger.info(f"========= WARNING: WRITING SEs TO {out_name} =========")
+        logger.info("WARNING: WRITING SEs", filename=out_name)
 
     out_n = 0
     for d in set(output_df["date"]):

google_symptoms/delphi_google_symptoms/date_utils.py

@@ -98,7 +98,7 @@ def generate_num_export_days(params: Dict, logger) -> [int]:
                 expected_date_diff += global_max_expected_lag
 
             if latest_date_diff > expected_date_diff:
-                logger.info(f"Missing dates from: {to_datetime(min(gs_metadata.max_time)).date()}")
+                logger.info("Missing date", date=to_datetime(min(gs_metadata.max_time)).date())
 
             num_export_days = expected_date_diff
 

google_symptoms/delphi_google_symptoms/patch.py

@@ -58,16 +58,19 @@ def patch(params):
     issue_date = datetime.strptime(params["patch"]["start_issue"], "%Y-%m-%d")
     end_issue = datetime.strptime(params["patch"]["end_issue"], "%Y-%m-%d")
 
-    logger.info(f"""Start patching {params["patch"]["patch_dir"]}""")
-    logger.info(f"""Start issue: {issue_date.strftime("%Y-%m-%d")}""")
-    logger.info(f"""End issue: {end_issue.strftime("%Y-%m-%d")}""")
+    logger.info(
+        "Starting patching",
+        patch_directory=params["patch"]["patch_dir"],
+        start_issue=issue_date.strftime("%Y-%m-%d"),
+        end_issue=end_issue.strftime("%Y-%m-%d"),
+    )
 
     makedirs(params["patch"]["patch_dir"], exist_ok=True)
 
     patch_dates = generate_patch_dates(params)
 
     while issue_date <= end_issue:
-        logger.info(f"""Running issue {issue_date.strftime("%Y-%m-%d")}""")
+        logger.info("Running issue", issue_date=issue_date.strftime("%Y-%m-%d"))
 
         # Output dir setup
         current_issue_yyyymmdd = issue_date.strftime("%Y%m%d")

google_symptoms/delphi_google_symptoms/run.py

@@ -80,10 +80,7 @@ def run_module(params, logger=None):
         if len(df_pull) == 0:
             continue
         for metric, smoother in product(COMBINED_METRIC, SMOOTHERS):
-            logger.info("generating signal and exporting to CSV",
-                        geo_res=geo_res,
-                        metric=metric,
-                        smoother=smoother)
+            logger.info("Generating signal and exporting to CSV", geo_res=geo_res, metric=metric, smoother=smoother)
             df = df_pull
             df["val"] = df[metric].astype(float)
             df["val"] = df[["geo_id", "val"]].groupby(
@@ -96,7 +93,7 @@ def run_module(params, logger=None):
             df = df.reset_index()
             sensor_name = "_".join([smoother, "search"])
             if len(df) == 0:
-                logger.info("No data for %s_%s_%s", geo_res, metric.lower(), sensor_name)
+                logger.info("No data for signal", signal=f"{geo_res}_{metric.lower()}_{sensor_name}")
                 continue
             exported_csv_dates = create_export_csv(
                 df,

nssp/delphi_nssp/run.py

@@ -90,7 +90,7 @@ def run_module(params):
         for geo in GEOS:
             df = df_pull.copy()
             df["val"] = df[signal]
-            logger.info("Generating signal and exporting to CSV", metric=signal)
+            logger.info("Generating signal and exporting to CSV", geo=geo, signal=signal)
             if geo == "nation":
                 df = df[df["geography"] == "United States"]
                 df["geo_id"] = "us"