Merge branch 'main' into remove-wip

Author: chinandrew

.github/workflows/json-check.yml

@@ -0,0 +1,28 @@
+# This workflow will verify all JSON files in the directory, excluding venv files
+
+name: JSON check
+
+on:
+  push:
+    branches: [ main, prod ]
+  pull_request:
+    types: [ opened, synchronize, reopened, ready_for_review ]
+    branches: [ main, prod ]
+
+jobs:
+  build:
+    runs-on: ubuntu-20.04
+    if: github.event.pull_request.draft == false
+    steps:
+    - uses: actions/checkout@v2
+    - name: Set up Python 3.8
+      uses: actions/setup-python@v2
+      with:
+        python-version: 3.8
+    - name: Check JSONs
+      run: |
+        python3 -c "import json; from pathlib import Path
+        for i in Path('.').glob('**/*.json*'):
+            if '/env/' not in str(i):
+                print(i)
+                json.load(open(str(i)))"

_delphi_utils_python/data_proc/geomap/geo_data_proc.py

@@ -400,7 +400,14 @@ def create_fips_population_table():
     df_pr = df_pr[~df_pr["fips"].isin(census_pop["fips"])]
     census_pop_pr = pd.concat([census_pop, df_pr])
 
-    census_pop_pr.to_csv(join(OUTPUT_DIR, FIPS_POPULATION_OUT_FILENAME), index=False)
+    # Filled from https://www.census.gov/data/tables/2010/dec/2010-island-areas.html
+    territories_pop = pd.DataFrame({
+        "fips": ["60010", "60020", "60030", "60040", "60050", "66010", "78010", "78020", "78030", "69085", "69100", "69110", "69120"],
+        "pop": [23030, 1143, 0, 17, 31329, 159358, 50601, 4170, 51634, 0, 2527, 48220, 3136]
+    })
+    census_pop_territories = pd.concat([census_pop_pr, territories_pop])
+
+    census_pop_territories.to_csv(join(OUTPUT_DIR, FIPS_POPULATION_OUT_FILENAME), index=False)
 
 
 def derive_zip_population_table():

_delphi_utils_python/delphi_utils/data/fips_hhs_table.csv

@@ -3273,3 +3273,16 @@ fips,hhs
 72149,2
 72151,2
 72153,2
+60010,9
+60020,9
+60030,9
+60040,9
+60050,9
+66010,9
+78010,2
+78020,2
+78030,2
+69085,9
+69100,9
+69110,9
+69120,9

_delphi_utils_python/delphi_utils/data/fips_pop.csv

@@ -3273,3 +3273,16 @@ fips,pop
 72149,26073
 72151,37941
 72153,42043
+60010,23030
+60020,1143
+60030,0
+60040,17
+60050,31329
+66010,159358
+78010,50601
+78020,4170
+78030,51634
+69085,0
+69100,2527
+69110,48220
+69120,3136

_delphi_utils_python/delphi_utils/data/fips_state_table.csv

@@ -3273,3 +3273,16 @@ fips,state_code,state_id,state_name
 72149,72,pr,Puerto Rico
 72151,72,pr,Puerto Rico
 72153,72,pr,Puerto Rico
+60010,60,as,American Samoa
+60020,60,as,American Samoa
+60030,60,as,American Samoa
+60040,60,as,American Samoa
+60050,60,as,American Samoa
+66010,66,gu,Guam
+78010,78,vi,U.S. Virgin Islands
+78020,78,vi,U.S. Virgin Islands
+78030,78,vi,U.S. Virgin Islands
+69085,69,mp,Northern Mariana Islands
+69100,69,mp,Northern Mariana Islands
+69110,69,mp,Northern Mariana Islands
+69120,69,mp,Northern Mariana Islands

_delphi_utils_python/delphi_utils/geomap.py

@@ -125,6 +125,11 @@ def __init__(self):
     # Utility functions
     def _load_crosswalk(self, from_code, to_code):
         """Load the crosswalk from from_code -> to_code."""
+        assert from_code in self.crosswalk_filepaths, \
+            f"No crosswalk files for {from_code}; try {'; '.join(self.crosswalk_filepaths.keys())}"
+        assert to_code in self.crosswalk_filepaths[from_code], \
+            f"No crosswalk file from {from_code} to {to_code}; try" \
+            f"{'; '.join(self.crosswalk_filepaths[from_code].keys())}"
         stream = pkg_resources.resource_stream(
             __name__, self.crosswalk_filepaths[from_code][to_code]
         )
@@ -301,6 +306,7 @@ def add_geocode(
         df = df.copy()
         from_col = from_code if from_col is None else from_col
         new_col = new_code if new_col is None else new_col
+        assert from_col != new_col, f"Can't use the same column '{from_col}' for both from_col and to_col"
         state_codes = ["state_code", "state_id", "state_name"]
 
         if not is_string_dtype(df[from_col]):
@@ -319,7 +325,7 @@ def add_geocode(
             return df
         elif new_code == "nation":
             raise ValueError(
-                "Conversion to the nation level is only supported from the FIPS and ZIP codes."
+                f"Conversion to the nation level is not supported from {from_code}; try fips, zip, or state_*"
             )
 
         # state codes are all stored in one table
@@ -522,6 +528,17 @@ def fips_to_megacounty(
         data = data.reset_index().groupby([date_col, mega_col]).sum()
         return data.reset_index()
 
+    def as_mapper_name(self, geo_type, state="state_id"):
+        """
+        Return the mapper equivalent of a region type.
+
+        Human-readable names like 'county' will return their mapper equivalents ('fips').
+        """
+        if geo_type == "state":
+            return state
+        if geo_type == "county":
+            return "fips"
+        return geo_type
     def get_geo_values(self, geo_type):
         """
         Return a set of all values for a given geography type.

_delphi_utils_python/delphi_utils/logger.py

@@ -1,8 +1,26 @@
 """Structured logger utility for creating JSON logs in Delphi pipelines."""
 import logging
+import sys
+import threading
 import structlog
 
-def get_structured_logger(name = __name__, filename = None):
+
+def handle_exceptions(logger):
+    """Handle exceptions using the provided logger."""
+    def exception_handler(etype, value, traceback):
+        logger.exception("Top-level exception occurred",
+                         exc_info=(etype, value, traceback))
+
+    def multithread_exception_handler(args):
+        exception_handler(args.exc_type, args.exc_value, args.exc_traceback)
+
+    sys.excepthook = exception_handler
+    threading.excepthook = multithread_exception_handler
+
+
+def get_structured_logger(name=__name__,
+                          filename=None,
+                          log_exceptions=True):
     """Create a new structlog logger.
 
     Use the logger returned from this in indicator code using the standard
@@ -66,4 +84,9 @@ def get_structured_logger(name = __name__, filename = None):
         cache_logger_on_first_use=True,
     )
 
-    return structlog.get_logger(name)
+    logger = structlog.get_logger(name)
+
+    if log_exceptions:
+        handle_exceptions(logger)
+
+    return logger

_delphi_utils_python/tests/test_geomap.py

@@ -307,6 +307,6 @@ def test_get_geos(self):
         gmpr = GeoMapper()
         assert gmpr.get_geo_values("nation") == {"us"}
         assert gmpr.get_geo_values("hhs") == set(str(i) for i in range(1, 11))
-        assert len(gmpr.get_geo_values("fips")) == 3274
+        assert len(gmpr.get_geo_values("fips")) == 3287
         assert len(gmpr.get_geo_values("state_id")) == 60
         assert len(gmpr.get_geo_values("zip")) == 32976

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

@@ -31,5 +31,6 @@
     "user": "{{ changehc_sftp_user }}",
     "pass": "{{ changehc_sftp_password }}",
     "port": "{{ changehc_sftp_port }}"
-  }
+  },
+  "log_filename": "/var/log/indicators/changehc.log"
 }

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

@@ -11,5 +11,6 @@
     "aws_access_key_id": "{{ delphi_aws_access_key_id }}",
     "aws_secret_access_key": "{{ delphi_aws_secret_access_key }}"
   },
-  "bucket_name": "delphi-covidcast-indicator-output"
+  "bucket_name": "delphi-covidcast-indicator-output",
+  "log_filename": "/var/log/indicators/nchs_mortality.log"
 }

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

@@ -6,10 +6,12 @@
   "export_end_date": "",
   "pull_start_date": "2020-05-26",
   "pull_end_date":"",
-  "mail_server": "{{ quidel_datadrop_mail_server }}",
-  "account": "{{ quidel_datadrop_account }}",
-  "password": "{{ quidel_datadrop_password }}",
-  "sender": "{{ quidel_datadrop_sender }}",
+  "aws_credentials": {
+    "aws_access_key_id": "{{ quidel_aws_access_key_id }}",
+    "aws_secret_access_key": "{{ quidel_aws_secret_access_key }}"
+  },
+  "bucket_name": "{{ quidel_aws_bucket_name }}",
   "wip_signal": [""],
-  "mode": ""
-}
+  "mode": "",
+  "log_filename": "/var/log/indicators/quidel_covidtest.log"
+} 

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

@@ -9,5 +9,6 @@
   "aws_default_region": "us-east-1",
   "aws_endpoint": "https://s3.wasabisys.com",
   "sync": true,
-  "wip_signal" : []
+  "wip_signal" : [],
+  "log_filename": "/var/log/indicators/safegraph.log"
 }

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

@@ -9,5 +9,6 @@
   "aws_default_region": "us-east-1",
   "aws_endpoint": "https://s3.wasabisys.com",
   "sync": true,
-  "wip_signal" : []
+  "wip_signal" : [],
+  "log_filename": "/var/log/indicators/safegraph_patterns.log"
 }

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

@@ -1,5 +1,6 @@
 {
   "channel": "#sir-complains-a-lot",
+  "log_filename": "/var/log/indicators/sircal.log",
   "slack_token": "{{ sir_complainsalot_slack_token }}",
   "sources": {
     "doctor-visits": {

ansible/vars.yaml

@@ -15,10 +15,9 @@ delphi_aws_access_key_id: "{{ vault_delphi_aws_access_key_id }}"
 delphi_aws_secret_access_key: "{{ vault_delphi_aws_secret_access_key }}"
 safegraph_aws_access_key_id: "{{ vault_safegraph_aws_access_key_id }}"
 safegraph_aws_secret_access_key: "{{ vault_safegraph_aws_secret_access_key }}"
-quidel_datadrop_mail_server: "{{ vault_quidel_datadrop_mail_server }}"
-quidel_datadrop_account: "{{ vault_quidel_datadrop_account }}"
-quidel_datadrop_password: "{{ vault_quidel_datadrop_password }}"
-quidel_datadrop_sender: "{{ vault_quidel_datadrop_sender }}"
+quidel_aws_access_key_id: "{{ vault_quidel_aws_access_key_id }}"
+quidel_aws_secret_access_key: "{{ vault_quidel_aws_secret_access_key }}"
+quidel_aws_bucket_name: 'delphi-quidel-data'
 changehc_sftp_host: "{{ vault_changehc_sftp_host }}"
 changehc_sftp_port: "{{ vault_changehc_sftp_port }}"
 changehc_sftp_user: "{{ vault_changehc_sftp_user }}"