Flu signal

Author: rumackaaron

changehc/delphi_changehc/constants.py

@@ -3,7 +3,9 @@
 SMOOTHED_ADJ = "smoothed_adj_outpatient_covid"
 SMOOTHED_CLI = "smoothed_outpatient_cli"
 SMOOTHED_ADJ_CLI = "smoothed_adj_outpatient_cli"
-SIGNALS = [SMOOTHED, SMOOTHED_ADJ, SMOOTHED_CLI, SMOOTHED_ADJ_CLI]
+SMOOTHED_FLU = "smoothed_outpatient_flu"
+SMOOTHED_ADJ_FLU = "smoothed_adj_outpatient_flu"
+SIGNALS = [SMOOTHED, SMOOTHED_ADJ, SMOOTHED_CLI, SMOOTHED_ADJ_CLI, SMOOTHED_FLU, SMOOTHED_ADJ_FLU]
 NA = "NA"
 HRR = "hrr"
 FIPS = "fips"

changehc/delphi_changehc/download_ftp_files.py

@@ -93,3 +93,30 @@ def download_cli(filedate, out_path, ftp_conn):
     finally:
         if client:
             client.close()
+
+
+def download_flu(filedate, out_path, ftp_conn):
+    """Download files necessary to create chng-flu signal from ftp server.
+
+    Args:
+        filedate: YYYYmmdd string for which the files are named
+        out_path: Path to local directory into which to download the files
+        ftp_conn: Dict containing login credentials to ftp server
+    """
+    # open client
+    try:
+        client = paramiko.SSHClient()
+        client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
+
+        client.connect(ftp_conn["host"], username=ftp_conn["user"],
+                       password=ftp_conn["pass"],
+                       port=ftp_conn["port"],
+                       allow_agent=False, look_for_keys=False)
+        sftp = client.open_sftp()
+
+        sftp.chdir('/countproducts')
+        get_files_from_dir(sftp, filedate, out_path)
+
+    finally:
+        if client:
+            client.close()

changehc/delphi_changehc/load_data.py

@@ -153,3 +153,36 @@ def load_cli_data(denom_filepath, flu_filepath, mixed_filepath, flu_like_filepat
     data = data[["num", "den"]]
 
     return data
+
+
+def load_flu_data(denom_filepath, flu_filepath, dropdate, base_geo):
+    """Load in denominator and flu data, and combine them.
+
+    Args:
+        denom_filepath: path to the aggregated denominator data
+        flu_filepath: path to the aggregated flu data
+        dropdate: data drop date (datetime object)
+        base_geo: base geographic unit before aggregation ('fips')
+
+    Returns:
+        combined multiindexed dataframe, index 0 is geo_base, index 1 is date
+    """
+    assert base_geo == "fips", "base unit must be 'fips'"
+
+    # load each data stream
+    denom_data = load_chng_data(denom_filepath, dropdate, base_geo,
+                     Config.DENOM_COLS, Config.DENOM_DTYPES, Config.DENOM_COL)
+    flu_data = load_chng_data(flu_filepath, dropdate, base_geo,
+                     Config.FLU_COLS, Config.FLU_DTYPES, Config.FLU_COL)
+
+    # merge data
+    data = denom_data.merge(flu_data, how="outer", left_index=True, right_index=True)
+    assert data.isna().all(axis=1).sum() == 0, "entire row is NA after merge"
+
+    # calculate combined numerator and denominator
+    data.fillna(0, inplace=True)
+    data["num"] = data[Config.FLU_COL]
+    data["den"] = data[Config.DENOM_COL]
+    data = data[["num", "den"]]
+
+    return data

changehc/delphi_changehc/run.py

@@ -14,8 +14,8 @@
 from delphi_utils import get_structured_logger
 
 # first party
-from .download_ftp_files import download_covid, download_cli
-from .load_data import load_combined_data, load_cli_data
+from .download_ftp_files import download_covid, download_cli, download_flu
+from .load_data import load_combined_data, load_cli_data, load_flu_data
 from .update_sensor import CHCSensorUpdater
 
 
@@ -30,6 +30,8 @@ def retrieve_files(params, filedate, logger):
             download_covid(filedate, params["indicator"]["input_cache_dir"], params["indicator"]["ftp_conn"])
         if "cli" in params["indicator"]["types"]:
             download_cli(filedate, params["indicator"]["input_cache_dir"], params["indicator"]["ftp_conn"])
+        if "flu" in params["indicator"]["types"]:
+            download_flu(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)
         covid_file = "%s/%s_Counts_Products_Covid.dat.gz" % (params["indicator"]["input_cache_dir"],filedate)
@@ -53,6 +55,8 @@ def retrieve_files(params, filedate, logger):
         file_dict["mixed"] = mixed_file
         file_dict["flu_like"] = flu_like_file
         file_dict["covid_like"] = covid_like_file
+    if "flu" in params["indicator"]["types"]:
+        file_dict["flu"] = flu_file
     return file_dict
 
 
@@ -75,6 +79,9 @@ def make_asserts(params):
                     files["flu_like"] is not None and \
                     files["covid_like"] is not None,\
                     "files must be all present or all absent"
+    if "flu" in params["indicator"]["types"]:
+        assert (files["denom"] is None) == (files["flu"] is None), \
+            "exactly one of denom and flu files are provided"
 
 
 def run_module(params: Dict[str, Dict[str, Any]]):
@@ -182,6 +189,9 @@ def run_module(params: Dict[str, Dict[str, Any]]):
                 elif numtype == "cli":
                     data = load_cli_data(file_dict["denom"],file_dict["flu"],file_dict["mixed"],
                              file_dict["flu_like"],file_dict["covid_like"],dropdate_dt,"fips")
+                elif numtype == "flu":
+                    data = load_flu_data(file_dict["denom"],file_dict["flu"],
+                             dropdate_dt,"fips")
                 more_stats = su_inst.update_sensor(
                     data,
                     params["common"]["export_dir"],

changehc/delphi_changehc/update_sensor.py

@@ -15,7 +15,8 @@
 
 # first party
 from .config import Config
-from .constants import SMOOTHED, SMOOTHED_ADJ, SMOOTHED_CLI, SMOOTHED_ADJ_CLI, NA
+from .constants import SMOOTHED, SMOOTHED_ADJ, SMOOTHED_CLI, SMOOTHED_ADJ_CLI,\
+                       SMOOTHED_FLU, SMOOTHED_ADJ_FLU, NA
 from .sensor import CHCSensor
 
 
@@ -114,9 +115,11 @@ def __init__(self,
             signal_name = SMOOTHED_ADJ if self.weekday else SMOOTHED
         elif self.numtype == "cli":
             signal_name = SMOOTHED_ADJ_CLI if self.weekday else SMOOTHED_CLI
+        elif self.numtype == "flu":
+            signal_name = SMOOTHED_ADJ_FLU if self.weekday else SMOOTHED_FLU
         else:
             raise ValueError(f'Unsupported numtype received "{numtype}",'
-                             f' must be one of ["covid", "cli"]')
+                             f' must be one of ["covid", "cli", "flu"]')
         self.signal_name = add_prefix([signal_name], wip_signal=wip_signal)[0]
 
         # initialize members set in shift_dates().