Merge branch 'main' into clean-up

Author: chinandrew

_delphi_utils_python/delphi_utils/export.py

@@ -4,6 +4,7 @@
 from os.path import join
 from typing import Optional
 
+from epiweeks import Week
 import numpy as np
 import pandas as pd
 
@@ -16,7 +17,8 @@ def create_export_csv(
     start_date: Optional[datetime] = None,
     end_date: Optional[datetime] = None,
     remove_null_samples: Optional[bool] = False,
-    write_empty_days: Optional[bool] = False
+    write_empty_days: Optional[bool] = False,
+    weekly_dates = False,
 ):
     """Export data in the format expected by the Delphi API.
 
@@ -65,10 +67,15 @@ def create_export_csv(
         dates = pd.date_range(start_date, end_date)
 
     for date in dates:
+        if weekly_dates:
+            t = Week.fromdate(pd.to_datetime(str(date)))
+            date_str = "weekly_" + str(t.year) + str(t.week).zfill(2)
+        else:
+            date_str = date.strftime('%Y%m%d')
         if metric is None:
-            export_filename = f"{date.strftime('%Y%m%d')}_{geo_res}_{sensor}.csv"
+            export_filename = f"{date_str}_{geo_res}_{sensor}.csv"
         else:
-            export_filename = f"{date.strftime('%Y%m%d')}_{geo_res}_{metric}_{sensor}.csv"
+            export_filename = f"{date_str}_{geo_res}_{metric}_{sensor}.csv"
         export_file = join(export_dir, export_filename)
         export_df = df[df["timestamp"] == date][["geo_id", "val", "se", "sample_size",]]
         if remove_null_samples:

_delphi_utils_python/setup.py

@@ -7,6 +7,7 @@
 required = [
     "boto3",
     "covidcast",
+    "epiweeks",
     "freezegun",
     "gitpython",
     "mock",

cdc_covidnet/delphi_cdc_covidnet/covidnet.py

@@ -6,7 +6,7 @@
 """
 
 import json
-import logging
+from logging import Logger
 import os
 from typing import Tuple, List
 from multiprocessing import cpu_count, Pool
@@ -100,7 +100,7 @@ def download_hosp_data(
 
     @staticmethod
     def download_all_hosp_data(
-            mappings_file: str, cache_path: str, parallel: bool = False
+            mappings_file: str, cache_path: str, logger: Logger, parallel: bool = False
         ) -> List[str]:
         """
         Download hospitalization data for all states listed in the mappings JSON file to disk.
@@ -146,7 +146,7 @@ def download_all_hosp_data(
         else:
             for args in state_args:
                 CovidNet.download_hosp_data(*args)
-                logging.debug("Downloading for nid=%s, cid=%s", args[0], args[1])
+                logger.debug("Downloading for nid=%s, cid=%s", args[0], args[1])
 
         return state_files
 

cdc_covidnet/delphi_cdc_covidnet/run.py

@@ -4,12 +4,13 @@
 This module should contain a function called `run_module`, that is executed
 when the module is run with `python -m delphi_cdc_covidnet`.
 """
-import logging
 from datetime import datetime
 from os import remove
 from os.path import join
 from typing import Dict, Any
 
+from delphi_utils import get_structured_logger
+
 from .covidnet import CovidNet
 from .update_sensor import update_sensor
 
@@ -32,7 +33,9 @@ def run_module(params: Dict[str, Dict[str, Any]]):
             - "wip_signal": list of str or bool, to be passed to delphi_utils.add_prefix.
             - "input_cache_dir": str, directory to download source files.
     """
-    logging.basicConfig(level=logging.DEBUG)
+    logger = get_structured_logger(
+        __name__, filename=params["common"].get("log_filename"),
+        log_exceptions=params["common"].get("log_exceptions", True))
 
     start_date = datetime.strptime(params["indicator"]["start_date"], "%Y-%m-%d")
 
@@ -42,23 +45,24 @@ def run_module(params: Dict[str, Dict[str, Any]]):
     else:
         end_date = datetime.strptime(params["indicator"]["end_date"], "%Y-%m-%d")
 
-    logging.info("start date:\t%s", start_date.date())
-    logging.info("end date:\t%s", end_date.date())
+    logger.info("start date:\t%s", start_date.date())
+    logger.info("end date:\t%s", end_date.date())
 
-    logging.info("outpath:\t%s", params["common"]["export_dir"])
-    logging.info("parallel:\t%s", params["indicator"]["parallel"])
+    logger.info("outpath:\t%s", params["common"]["export_dir"])
+    logger.info("parallel:\t%s", params["indicator"]["parallel"])
 
     # Only geo is state, and no weekday adjustment for now
     # COVID-NET data is by weeks anyway, not daily
-    logging.info("starting state, no adj")
+    logger.info("starting state, no adj")
 
     # Download latest COVID-NET files into the cache directory first
     mappings_file = join(params["indicator"]["input_cache_dir"], "init.json")
     CovidNet.download_mappings(outfile=mappings_file)
     _, mmwr_info, _ = CovidNet.read_mappings(mappings_file)
     state_files = CovidNet.download_all_hosp_data(
         mappings_file, params["indicator"]["input_cache_dir"],
-        parallel=params["indicator"]["parallel"])
+        parallel=params["indicator"]["parallel"],
+        logger=logger)
 
     update_sensor(
         state_files,
@@ -73,4 +77,4 @@ def run_module(params: Dict[str, Dict[str, Any]]):
     for state_file in state_files:
         remove(state_file)
 
-    logging.info("finished all")
+    logger.info("finished all")

cdc_covidnet/tests/test_covidnet.py

@@ -1,4 +1,5 @@
 import json
+import logging
 from os.path import join, exists
 from tempfile import TemporaryDirectory
 
@@ -7,6 +8,7 @@
 from delphi_cdc_covidnet.api_config import APIConfig
 from delphi_cdc_covidnet.covidnet import CovidNet
 
+TEST_LOGGER = logging.getLogger()
 
 class TestCovidNet:
 
@@ -65,14 +67,14 @@ def test_hosp_data(self):
 
             # Non-parallel
             state_files = CovidNet.download_all_hosp_data(
-                init_file, temp_dir, parallel=False)
+                init_file, temp_dir, TEST_LOGGER, parallel=False)
             assert len(state_files) == num_states
             for state_file in state_files:
                 assert exists(state_file)
 
             # Parallel
             state_files_par = CovidNet.download_all_hosp_data(
-                init_file, temp_dir, parallel=True)
+                init_file, temp_dir, TEST_LOGGER, parallel=True)
             assert set(state_files) == set(state_files_par)
             assert len(state_files_par) == num_states
             for state_file in state_files_par:

changehc/delphi_changehc/run.py

@@ -173,7 +173,8 @@ def run_module(params: Dict[str, Dict[str, Any]]):
                     weekday,
                     numtype,
                     params["indicator"]["se"],
-                    params["indicator"]["wip_signal"]
+                    params["indicator"]["wip_signal"],
+                    logger
                 )
                 if numtype == "covid":
                     data = load_combined_data(file_dict["denom"],

changehc/delphi_changehc/sensor.py

@@ -87,7 +87,7 @@ def backfill(
         return new_num, new_den
 
     @staticmethod
-    def fit(y_data, first_sensor_date, geo_id, num_col="num", den_col="den"):
+    def fit(y_data, first_sensor_date, geo_id, logger, num_col="num", den_col="den"):
         """Fitting routine.
 
         Args:
@@ -121,7 +121,7 @@ def fit(y_data, first_sensor_date, geo_id, num_col="num", den_col="den"):
         se_valid = valid_rates.eval('sqrt(rate * (1 - rate) / den)')
         rate_data['se'] = se_valid
 
-        logging.debug("{0}: {1:.3f},[{2:.3f}]".format(
+        logger.debug("{0}: {1:.3f},[{2:.3f}]".format(
             geo_id, rate_data['rate'][-1], rate_data['se'][-1]
         ))
         return {"geo_id": geo_id,

changehc/delphi_changehc/update_sensor.py

@@ -20,7 +20,7 @@
 from .weekday import Weekday
 
 
-def write_to_csv(df, geo_level, write_se, day_shift, out_name, output_path=".", start_date=None, end_date=None):
+def write_to_csv(df, geo_level, write_se, day_shift, out_name, logger, output_path=".", start_date=None, end_date=None):
     """Write sensor values to csv.
 
     Args:
@@ -43,15 +43,15 @@ def write_to_csv(df, geo_level, write_se, day_shift, out_name, output_path=".",
     assert df[suspicious_se_mask].empty, " se contains suspiciously large values"
     assert not df["se"].isna().any(), " se contains nan values"
     if write_se:
-        logging.info("========= WARNING: WRITING SEs TO {0} =========".format(out_name))
+        logger.info("========= WARNING: WRITING SEs TO {0} =========".format(out_name))
     else:
         df["se"] = np.nan
 
     assert not df["val"].isna().any(), " val contains nan values"
     suspicious_val_mask = df["val"].gt(90)
     if not df[suspicious_val_mask].empty:
         for geo in df.loc[suspicious_val_mask, "geo_id"]:
-            logging.warning("value suspiciously high, {0}: {1}".format(
+            logger.warning("value suspiciously high, {0}: {1}".format(
                 geo, out_name
             ))
 
@@ -64,10 +64,10 @@ def write_to_csv(df, geo_level, write_se, day_shift, out_name, output_path=".",
         sensor=out_name,
         write_empty_days=True
     )
-    logging.debug("wrote {0} rows for {1} {2}".format(
+    logger.debug("wrote {0} rows for {1} {2}".format(
         df.size, df["geo_id"].unique().size, geo_level
     ))
-    logging.debug("wrote files to {0}".format(output_path))
+    logger.debug("wrote files to {0}".format(output_path))
     return dates
 
 
@@ -83,7 +83,8 @@ def __init__(self,
                  weekday,
                  numtype,
                  se,
-                 wip_signal):
+                 wip_signal,
+                 logger):
         """Init Sensor Updater.
 
         Args:
@@ -96,7 +97,9 @@ def __init__(self,
             numtype: type of count data used, one of ["covid", "cli"]
             se: boolean to write out standard errors, if true, use an obfuscated name
             wip_signal: Prefix for WIP signals
+            logger: the structured logger
         """
+        self.logger = logger
         self.startdate, self.enddate, self.dropdate = [
             pd.to_datetime(t) for t in (startdate, enddate, dropdate)]
         # handle dates
@@ -145,7 +148,7 @@ def geo_reindex(self, data):
         geo = self.geo
         gmpr = GeoMapper()
         if geo not in {"county", "state", "msa", "hrr", "nation", "hhs"}:
-            logging.error("{0} is invalid, pick one of 'county', "
+            self.logger.error("{0} is invalid, pick one of 'county', "
                           "'state', 'msa', 'hrr', 'hss','nation'".format(geo))
             return False
         if geo == "county":
@@ -197,12 +200,12 @@ def update_sensor(self,
                 sub_data.reset_index(level=0,inplace=True)
                 if self.weekday:
                     sub_data = Weekday.calc_adjustment(wd_params, sub_data)
-                res = CHCSensor.fit(sub_data, self.burnindate, geo_id)
+                res = CHCSensor.fit(sub_data, self.burnindate, geo_id, self.logger)
                 res = pd.DataFrame(res).loc[final_sensor_idxs]
                 dfs.append(res)
         else:
             n_cpu = min(10, cpu_count())
-            logging.debug("starting pool with {0} workers".format(n_cpu))
+            self.logger.debug("starting pool with {0} workers".format(n_cpu))
             with Pool(n_cpu) as pool:
                 pool_results = []
                 for geo_id, sub_data in data_frame.groupby(level=0,as_index=False):
@@ -211,7 +214,7 @@ def update_sensor(self,
                         sub_data = Weekday.calc_adjustment(wd_params, sub_data)
                     pool_results.append(
                         pool.apply_async(
-                            CHCSensor.fit, args=(sub_data, self.burnindate, geo_id,),
+                            CHCSensor.fit, args=(sub_data, self.burnindate, geo_id, self.logger),
                         )
                     )
                 pool_results = [proc.get() for proc in pool_results]

changehc/tests/test_sensor.py

@@ -1,4 +1,5 @@
 # standard
+import logging
 
 import numpy as np
 import numpy.random as nr
@@ -19,6 +20,7 @@
 COVID_FILEPATH = PARAMS["indicator"]["input_covid_file"]
 DENOM_FILEPATH = PARAMS["indicator"]["input_denom_file"]
 DROP_DATE = pd.to_datetime(PARAMS["indicator"]["drop_date"])
+TEST_LOGGER = logging.getLogger()
 
 class TestLoadData:
     combined_data = load_combined_data(DENOM_FILEPATH, COVID_FILEPATH, DROP_DATE,
@@ -56,7 +58,7 @@ def test_fit_fips(self):
         for fips in all_fips:
             sub_data = self.combined_data.loc[fips]
             sub_data = sub_data.reindex(date_range, fill_value=0)
-            res0 = CHCSensor.fit(sub_data, date_range[0], fips)
+            res0 = CHCSensor.fit(sub_data, date_range[0], fips, TEST_LOGGER)
 
             if np.isnan(res0["rate"]).all():
                 assert res0["incl"].sum() == 0