Merge branch 'main' of github.com:cmu-delphi/covidcast-indicators into signals

Author: sgsmob

changehc/delphi_changehc/config.py

@@ -6,7 +6,6 @@
 """
 
 from datetime import datetime, timedelta
-import numpy as np
 
 
 class Config:

changehc/delphi_changehc/constants.py

@@ -1,6 +1,6 @@
 """Registry for signal names and geo types"""
-SMOOTHED = "smoothed_chc"
-SMOOTHED_ADJ = "smoothed_adj_chc"
+SMOOTHED = "smoothed_cli"
+SMOOTHED_ADJ = "smoothed_adj_cli"
 SIGNALS = [SMOOTHED, SMOOTHED_ADJ]
 NA = "NA"
 HRR = "hrr"

changehc/delphi_changehc/download_ftp_files.py

@@ -0,0 +1,73 @@
+"""
+Downloads files modified in the last 24 hours from the specified ftp server."""
+
+# standard
+import datetime
+import functools
+from os import path
+
+# third party
+import paramiko
+
+
+def print_callback(filename, bytes_so_far, bytes_total):
+    """Log file transfer progress"""
+    rough_percent_transferred = int(100 * (bytes_so_far / bytes_total))
+    if (rough_percent_transferred % 25) == 0:
+        print(f'{filename} transfer: {rough_percent_transferred}%')
+
+
+def get_files_from_dir(sftp, out_path):
+    """Download files from sftp server that have been uploaded in last day
+    Args:
+        sftp: SFTP Session from Paramiko client
+        out_path: Path to local directory into which to download the files
+    """
+
+    current_time = datetime.datetime.now()
+
+    # go through files in recieving dir
+    filepaths_to_download = {}
+    for fileattr in sftp.listdir_attr():
+        file_time = datetime.datetime.fromtimestamp(fileattr.st_mtime)
+        filename = fileattr.filename
+        if current_time - file_time < datetime.timedelta(days=1) and \
+                not path.exists(path.join(out_path, filename)):
+            filepaths_to_download[filename] = path.join(out_path, filename)
+
+    # make sure we don't download more than 2 files per day
+    assert len(filepaths_to_download) <= 2, "more files dropped than expected"
+
+    # download!
+    for infile, outfile in filepaths_to_download.items():
+        callback_for_filename = functools.partial(print_callback, infile)
+        sftp.get(infile, outfile, callback=callback_for_filename)
+
+
+def download(out_path, ftp_conn):
+    """Downloads files necessary to create CHC signal from ftp server.
+    Args:
+        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"][1:] + ftp_conn["pass"][0],
+                       port=ftp_conn["port"],
+                       allow_agent=False, look_for_keys=False)
+        sftp = client.open_sftp()
+
+        sftp.chdir('/dailycounts/All_Outpatients_By_County')
+        get_files_from_dir(sftp, out_path)
+
+        sftp.chdir('/dailycounts/Covid_Outpatients_By_County')
+        get_files_from_dir(sftp, out_path)
+
+    finally:
+        if client:
+            client.close()

changehc/delphi_changehc/run.py

@@ -14,6 +14,7 @@
 from delphi_utils import read_params
 
 # first party
+from .download_ftp_files import download
 from .update_sensor import CHCSensorUpdator
 
 
@@ -25,11 +26,15 @@ def run_module():
 
     logging.basicConfig(level=logging.DEBUG)
 
+    ## download recent files from FTP server
+    logging.info("downloading recent files through SFTP")
+    download(params["cache_dir"], params["ftp_conn"])
+
     ## get end date from input file
     # the filenames are expected to be in the format:
     # Denominator: "YYYYMMDD_All_Outpatients_By_County.dat.gz"
     # Numerator: "YYYYMMDD_Covid_Outpatients_By_County.dat.gz"
-    
+
     if params["drop_date"] is None:
         dropdate_denom = datetime.strptime(
             Path(params["input_denom_file"]).name.split("_")[0], "%Y%m%d"

changehc/delphi_changehc/sensor.py

@@ -129,7 +129,7 @@ def fit(y_data, first_sensor_date, geo_id, num_col="num", den_col="den"):
 
         # checks - due to the smoother, the first value will be NA
         assert (
-                np.sum(np.isnan(smoothed_total_rates[1:]) == True) == 0
+                np.sum(np.isnan(smoothed_total_rates[1:])) == 0
         ), "NAs in rate calculation"
         assert (
                 np.sum(smoothed_total_rates[1:] <= 0) == 0

changehc/delphi_changehc/update_sensor.py

@@ -68,6 +68,8 @@ def write_to_csv(output_dict, write_se, out_name, output_path="."):
 
 
 class CHCSensorUpdator:
+    """Contains methods to update sensor and write results to csv
+    """
 
     def __init__(self,
                  startdate,

changehc/delphi_changehc/weekday.py

@@ -7,6 +7,7 @@
 # third party
 import cvxpy as cp
 import numpy as np
+from cvxpy.error import SolverError
 
 # first party
 from .config import Config
@@ -83,7 +84,7 @@ def get_params(data):
         try:
             prob = cp.Problem(cp.Minimize(-ll + lmbda * penalty))
             _ = prob.solve()
-        except:
+        except SolverError:
             # If the magnitude of the objective function is too large, an error is
             # thrown; Rescale the objective function
             prob = cp.Problem(cp.Minimize((-ll + lmbda * penalty) / 1e5))

changehc/params.json.template

@@ -18,5 +18,11 @@
     "aws_access_key_id": "",
     "aws_secret_access_key": ""
   },
-  "bucket_name": ""
-}
+  "bucket_name": "",
+  "ftp_conn": {
+    "host": "",
+    "user": "",
+    "pass": "",
+    "port": 0
+  }
+}

changehc/setup.py

@@ -11,7 +11,8 @@
     "delphi-utils",
     "covidcast",
     "boto3",
-    "moto"
+    "moto",
+    "paramiko"
 ]
 
 setup(

changehc/tests/params.json.template

@@ -2,8 +2,8 @@
   "static_file_dir": "../static",
   "export_dir": "./receiving",
   "cache_dir": "./cache",
-  "input_emr_file": "test_data/SYNICUE_CMB_INPATIENT_11062020.csv.gz",
-  "input_claims_file": "test_data/SYNEDI_AGG_INPATIENT_11062020_1451CDT.csv.gz",
+  "input_denom_file": "test_data/20200601_All_Outpatients_By_County.dat",
+  "input_covid_file": "test_data/20200601_Covid_Outpatients_By_County.dat",
   "start_date": "2020-02-01",
   "end_date": "2020-02-02",
   "drop_date": "2020-02-02",
@@ -19,4 +19,4 @@
     "aws_secret_access_key": "FAKE_TEST_SECRET_ACCESS_KEY"
   },
   "bucket_name": "test_bucket"
-}
+}

changehc/tests/test_download_ftp_files.py

@@ -0,0 +1,63 @@
+# standard
+import pytest
+import mock
+from datetime import datetime as dt
+from datetime import timedelta
+
+# first party
+from delphi_changehc.download_ftp_files import *
+
+class TestDownloadFTPFiles:
+
+    class MockSFTP:
+
+        # Mocks an SFTP connection
+        def __init__(self, attrs):
+            self.attrs = attrs
+            self.num_gets = 0
+
+        # Attrs are modified time and filename
+        def listdir_attr(self):
+            return self.attrs
+
+        # Don't download anything, just note that method was called
+        def get(self, infile, outfile, callback=None):
+            self.num_gets += 1
+            return
+
+
+    class FileAttr:
+
+        def __init__(self, time, name):
+            self.st_mtime = time
+            self.filename = name
+
+
+    @mock.patch("os.path")
+    def test_get_files(self, mock_path):
+
+        # When one new file is present, one file is downloaded
+        one_new = self.MockSFTP([self.FileAttr(dt.timestamp(dt.now()-timedelta(minutes=1)),"foo")])
+        get_files_from_dir(one_new, "")
+        assert one_new.num_gets == 1
+
+        # When one new file and one old file are present, one file is downloaded
+        one_new_one_old = self.MockSFTP([self.FileAttr(dt.timestamp(dt.now()-timedelta(minutes=1)),"foo"),
+                                         self.FileAttr(dt.timestamp(dt.now()-timedelta(days=10)),"foo")])
+        get_files_from_dir(one_new_one_old, "")
+        assert one_new_one_old.num_gets == 1
+
+        # When three new files are present, AssertionError
+        new_file1 = self.FileAttr(dt.timestamp(dt.now()-timedelta(minutes=1)),"foo1")
+        new_file2 = self.FileAttr(dt.timestamp(dt.now()-timedelta(minutes=1)),"foo2")
+        new_file3 = self.FileAttr(dt.timestamp(dt.now()-timedelta(minutes=1)),"foo3")
+        three_new = self.MockSFTP([new_file1, new_file2, new_file3])
+        with pytest.raises(AssertionError):
+            get_files_from_dir(three_new,"")
+
+        # When the file already exists, no files are downloaded
+        mock_path.exists.return_value = True
+        one_exists = self.MockSFTP([new_file1])
+        get_files_from_dir(one_new, "")
+        assert one_exists.num_gets == 0
+        

claims_hosp/delphi_claims_hosp/weekday.py

@@ -7,6 +7,7 @@
 # third party
 import cvxpy as cp
 import numpy as np
+from cvxpy.error import SolverError
 
 # first party
 from .config import Config
@@ -83,7 +84,7 @@ def get_params(data):
         try:
             prob = cp.Problem(cp.Minimize(-ll + lmbda * penalty))
             _ = prob.solve()
-        except:
+        except SolverError:
             # If the magnitude of the objective function is too large, an error is
             # thrown; Rescale the objective function
             prob = cp.Problem(cp.Minimize((-ll + lmbda * penalty) / 1e5))

emr_hosp/delphi_emr_hosp/weekday.py

@@ -7,6 +7,7 @@
 # third party
 import cvxpy as cp
 import numpy as np
+from cvxpy.error import SolverError
 
 # first party
 from .config import Config
@@ -83,7 +84,7 @@ def get_params(data):
         try:
             prob = cp.Problem(cp.Minimize(-ll + lmbda * penalty))
             _ = prob.solve()
-        except:
+        except SolverError:
             # If the magnitude of the objective function is too large, an error is
             # thrown; Rescale the objective function
             prob = cp.Problem(cp.Minimize((-ll + lmbda * penalty) / 1e5))

safegraph/delphi_safegraph/process.py

@@ -15,6 +15,7 @@
 # Base file name for raw data CSVs.
 CSV_NAME = 'social-distancing.csv.gz'
 
+
 def validate(df):
     """Confirms that a data frame has only one date."""
     timestamps = df['date_range_start'].apply(date_from_timestamp)
@@ -181,13 +182,13 @@ def process_window(df_list: List[pd.DataFrame],
                 f'{signal}_se': 'se',
                 f'{signal}_n': 'sample_size',
             }, axis=1)
-            df_export.to_csv(f'{export_dir}/{date}_{geo_res}_{signal}.csv',
+            date_str = date.strftime('%Y%m%d')
+            df_export.to_csv(f'{export_dir}/{date_str}_{geo_res}_{signal}.csv',
                              na_rep='NA',
                              index=False, )
 
 
-def process(current_filename: str,
-            previous_filenames: List[str],
+def process(filenames: List[str],
             signal_names: List[str],
             wip_signal,
             geo_resolutions: List[str],
@@ -196,11 +197,11 @@ def process(current_filename: str,
     as averaged over the previous week.
     Parameters
     ----------
-    current_filename: str
-        path to file holding the target date's data.
-    previous_filenames: List[str]
-        paths to files holding data from each day in the week preceding the
-        target date.
+    current_filename: List[str]
+        paths to files holding data.
+        The first entry of the list should correspond to the target date while
+        the remaining entries should correspond to the dates from each day in
+        the week preceding the target date.
     signal_names: List[str]
         signal names to be processed for a single date.
         A second version of each such signal named {SIGNAL}_7d_avg will be
@@ -220,8 +221,8 @@ def process(current_filename: str,
     one for the data averaged over the previous week to
     {export_dir}/{date}_{resolution}_{signal}_7d_avg.csv.
     """
-    past_week = [pd.read_csv(current_filename)]
-    for fname in previous_filenames:
+    past_week = []
+    for fname in filenames:
         if os.path.exists(fname):
             past_week.append(pd.read_csv(fname))
 
@@ -232,8 +233,8 @@ def process(current_filename: str,
                    export_dir)
     # ...then as part of the whole window.
     process_window(past_week,
-                  add_prefix(add_suffix(signal_names, '_7d_avg'),
-                             wip_signal,
-                             'wip_'),
-                  geo_resolutions,
-                  export_dir)
+                   add_prefix(add_suffix(signal_names, '_7d_avg'),
+                              wip_signal,
+                              'wip_'),
+                   geo_resolutions,
+                   export_dir)