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,71 @@
+"""
+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
+
+class AllowAnythingPolicy(paramiko.MissingHostKeyPolicy):
+    def missing_host_key(self, client, hostname, key):
+        return
+
+
+def print_callback(filename, bytes_so_far, bytes_total):
+    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):
+    current_time = datetime.datetime.now()
+    seconds_in_day = 24 * 60 * 60
+
+    # go through files in recieving dir
+    files_to_download = []
+    for fileattr in sftp.listdir_attr():
+        file_time = datetime.datetime.fromtimestamp(fileattr.st_mtime)
+        time_diff_to_current_time = current_time - file_time
+        if time_diff_to_current_time.total_seconds() <= seconds_in_day:
+            files_to_download.append(fileattr.filename)
+
+    filepaths_to_download = {}
+    for file in files_to_download:
+        full_path = path.join(out_path, file)
+        if path.exists(full_path):
+            print(f"{file} exists, skipping")
+        else:
+            filepaths_to_download[file] = full_path
+
+    # make sure we don't download more than 2 files per day
+    assert len(files_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):
+
+    # open client
+    client = paramiko.SSHClient()
+    client.set_missing_host_key_policy(AllowAnythingPolicy())
+
+    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)
+
+    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(