Merge pull request #1393 from cmu-delphi/release/indicators_v0.2.10_utils_v0.2.6

Release covidcast-indicators 0.2.10

Author: krivard

.bumpversion.cfg

@@ -1,5 +1,5 @@
 [bumpversion]
-current_version = 0.2.9
+current_version = 0.2.10
 commit = True
 message = chore: bump covidcast-indicators to {new_version}
 tag = False

ansible/templates/changehc-params-prod.json.j2

@@ -22,7 +22,7 @@
     "parallel": false,
     "geos": ["state", "msa", "hrr", "county", "hhs", "nation"],
     "weekday": [true, false],
-    "types": ["covid","cli"],
+    "types": ["covid","cli","flu"],
     "wip_signal": "",
     "ftp_conn": {
       "host": "{{ changehc_sftp_host }}",

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

@@ -41,35 +41,8 @@ def get_files_from_dir(sftp, filedate, out_path):
         sftp.get(infile, outfile, callback=callback_for_filename)
 
 
-def download_covid(filedate, out_path, ftp_conn):
-    """Download files necessary to create chng-covid 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()
-
-
-def download_cli(filedate, out_path, ftp_conn):
-    """Download files necessary to create chng-cli signal from ftp server.
+def download_counts(filedate, out_path, ftp_conn):
+    """Download files necessary to create chng- signals from ftp server.
 
     Args:
         filedate: YYYYmmdd string for which the files are named

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_counts
+from .load_data import load_combined_data, load_cli_data, load_flu_data
 from .update_sensor import CHCSensorUpdater
 
 
@@ -26,10 +26,7 @@ def retrieve_files(params, filedate, logger):
 
         ## download recent files from FTP server
         logger.info("downloading recent files through SFTP")
-        if "covid" in params["indicator"]["types"]:
-            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"])
+        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)
         covid_file = "%s/%s_Counts_Products_Covid.dat.gz" % (params["indicator"]["input_cache_dir"],filedate)
@@ -53,6 +50,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 +74,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 +184,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().

changehc/tests/test_load_data.py

@@ -14,10 +14,12 @@
     "indicator": {
         "input_denom_file": "test_data/20200601_Counts_Products_Denom.dat.gz",
         "input_covid_file": "test_data/20200601_Counts_Products_Covid.dat.gz",
+        "input_flu_file": "test_data/20200601_Counts_Products_Covid.dat.gz",
         "drop_date": "2020-06-01"
     }
 }
 COVID_FILEPATH = PARAMS["indicator"]["input_covid_file"]
+FLU_FILEPATH = PARAMS["indicator"]["input_flu_file"]
 DENOM_FILEPATH = PARAMS["indicator"]["input_denom_file"]
 DROP_DATE = pd.to_datetime(PARAMS["indicator"]["drop_date"])
 
@@ -29,6 +31,8 @@ class TestLoadData:
                     Config.COVID_COLS, Config.COVID_DTYPES, Config.COVID_COL)
     combined_data = load_combined_data(DENOM_FILEPATH, COVID_FILEPATH, DROP_DATE,
                                             "fips")
+    flu_data = load_flu_data(DENOM_FILEPATH, FLU_FILEPATH, DROP_DATE,
+                                            "fips")
     gmpr = GeoMapper()
 
     def test_base_unit(self):
@@ -43,6 +47,9 @@ def test_base_unit(self):
         with pytest.raises(AssertionError):
             load_combined_data(DENOM_FILEPATH, COVID_FILEPATH, DROP_DATE, "foo")
 
+        with pytest.raises(AssertionError):
+            load_flu_data(DENOM_FILEPATH, FLU_FILEPATH, DROP_DATE, "foo")
+
     def test_denom_columns(self):
         assert "fips" in self.denom_data.index.names
         assert "timestamp" in self.denom_data.index.names
@@ -71,6 +78,16 @@ def test_combined_columns(self):
         assert len(
             set(self.combined_data.columns) - set(expected_combined_columns)) == 0
 
+    def test_flu_columns(self):
+        assert "fips" in self.flu_data.index.names
+        assert "timestamp" in self.flu_data.index.names
+
+        expected_flu_columns = ["num", "den"]
+        for col in expected_flu_columns:
+            assert col in self.flu_data.columns
+        assert len(
+            set(self.flu_data.columns) - set(expected_flu_columns)) == 0
+
     def test_edge_values(self):
         for data in [self.denom_data,
                      self.covid_data,

facebook/delphiFacebook/NAMESPACE

@@ -112,6 +112,7 @@ importFrom(stats,na.omit)
 importFrom(stats,setNames)
 importFrom(stats,weighted.mean)
 importFrom(stringi,stri_extract)
+importFrom(stringi,stri_extract_first)
 importFrom(stringi,stri_replace)
 importFrom(stringi,stri_replace_all)
 importFrom(stringi,stri_split)

facebook/delphiFacebook/R/aggregate.R

@@ -179,6 +179,15 @@ summarize_indicators_day <- function(day_df, indicators, target_day, geo_level,
       var_weight <- indicators$var_weight[row]
       compute_fn <- indicators$compute_fn[[row]]
 
+      # Prevent smoothed weighted signals from being reported for dates after
+      # the latest available weight data.
+      if (target_day > params$latest_weight_date &&
+          indicators$smooth_days[row] > 1 &&
+          indicators$var_weight[row] != "weight_unif") {
+        
+        next
+      }
+
       # Copy only columns we're using.
       select_cols <- c(metric, var_weight, "weight_in_location")
       # This filter uses `x[, cols, with=FALSE]` rather than the newer