CHC SFTP Downloads

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,74 @@
+"""
+Downloads files modified in the last 24 hours from the specified ftp server."""
+
+# standard
+import datetime
+import functools
+import sys
+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(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,6 +26,10 @@ 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"

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/test_update_sensor.py

@@ -268,7 +268,6 @@ def test_handle_wip_signal(self):
         assert signal_names[0].startswith("wip_")
         assert all(not s.startswith("wip_") for s in signal_names[1:])
         # Test wip_signal = False (only unpublished signals should receive prefix)
-        # No CHC signal is published now, so both should get prefix
         signal_names = add_prefix(["xyzzy", SIGNALS[0]], False)
         assert signal_names[0].startswith("wip_")
-        assert all(s.startswith("wip_") for s in signal_names[1:])
+        assert all(not s.startswith("wip_") for s in signal_names[1:])