Merge branch 'main' into claimshosp-hhs-nation

Author: chinandrew

_delphi_utils_python/data_proc/geomap/geo_data_proc.py

@@ -547,7 +547,7 @@ def derive_fips_hhs_crosswalk():
     )
     state_hhs = pd.read_csv(
         join(OUTPUT_DIR, STATE_HHS_OUT_FILENAME),
-        dtype={"state_code": str, "state_id": str, "state_name": str},
+        dtype={"state_code": str, "hhs": str},
     )
 
     fips_pop["state_code"] = fips_pop["fips"].str[:2]
@@ -573,7 +573,7 @@ def derive_zip_hhs_crosswalk():
     )
     state_hhs = pd.read_csv(
         join(OUTPUT_DIR, STATE_HHS_OUT_FILENAME),
-        dtype={"state_code": str, "state_id": str, "state_name": str},
+        dtype={"state_code": str, "hhs": str},
     )
 
     (

_delphi_utils_python/delphi_utils/__init__.py

@@ -7,6 +7,7 @@
 from .export import create_export_csv
 from .utils import read_params
 
+from .logger import get_structured_logger
 from .geomap import GeoMapper
 from .smooth import Smoother
 from .signal import add_prefix, public_signal

_delphi_utils_python/delphi_utils/data/zip_hhs_table.csv

@@ -2780,7 +2780,6 @@ zip,weight,hhs
 08901,1.0,2
 08902,1.0,2
 08904,1.0,2
->>>>>>> add-hhs
 10001,1.0,2
 10002,1.0,2
 10003,1.0,2

_delphi_utils_python/delphi_utils/logger.py

@@ -0,0 +1,64 @@
+"""Structured logger utility for creating JSON logs in Delphi pipelines."""
+import logging
+import sys
+import structlog
+
+def get_structured_logger(name=__name__):
+    """Create a new structlog logger.
+
+    Use the logger returned from this in indicator code using the standard
+    wrapper calls, e.g.:
+
+    logger = get_structured_logger(__name__)
+    logger.warning("Error", type="Signal too low").
+
+    The output will be rendered as JSON which can easily be consumed by logs
+    processors.
+
+    See the structlog documentation for details.
+
+    Parameters
+    ---------
+    name: Name to use for logger (included in log lines), __name__ from caller
+    is a good choice.
+    """
+    # Configure the underlying logging configuration
+    logging.basicConfig(
+        format="%(message)s",
+        stream=sys.stdout,
+        level=logging.INFO)
+
+    # Configure structlog. This uses many of the standard suggestions from
+    # the structlog documentation.
+    structlog.configure(
+        processors=[
+            # Filter out log levels we are not tracking.
+            structlog.stdlib.filter_by_level,
+            # Include logger name in output.
+            structlog.stdlib.add_logger_name,
+            # Include log level in output.
+            structlog.stdlib.add_log_level,
+            # Allow formatting into arguments e.g., logger.info("Hello, %s",
+            # name)
+            structlog.stdlib.PositionalArgumentsFormatter(),
+            # Add timestamps.
+            structlog.processors.TimeStamper(fmt="iso"),
+            # Match support for exception logging in the standard logger.
+            structlog.processors.StackInfoRenderer(),
+            structlog.processors.format_exc_info,
+            # Decode unicode characters
+            structlog.processors.UnicodeDecoder(),
+            # Render as JSON
+            structlog.processors.JSONRenderer()
+        ],
+        # Use a dict class for keeping track of data.
+        context_class=dict,
+        # Use a standard logger for the actual log call.
+        logger_factory=structlog.stdlib.LoggerFactory(),
+        # Use a standard wrapper class for utilities like log.warning()
+        wrapper_class=structlog.stdlib.BoundLogger,
+        # Cache the logger
+        cache_logger_on_first_use=True,
+    )
+
+    return structlog.get_logger(name)

_delphi_utils_python/setup.py

@@ -12,6 +12,7 @@
     "pylint",
     "pytest",
     "pytest-cov",
+    "structlog",
     "xlrd"
 ]
 

sir_complainsalot/delphi_sir_complainsalot/check_source.py

@@ -1,10 +1,13 @@
 from dataclasses import dataclass
 from typing import List
+from delphi_utils import get_structured_logger
 
 import covidcast
 import numpy as np
 import pandas as pd
 
+logger = get_structured_logger(__name__)
+
 @dataclass
 class Complaint:
     message: str
@@ -83,6 +86,14 @@ def check_source(data_source, meta, params, grace):
             # No gap detection for this source
             continue
 
+        logger.info("Retrieving signal",
+                    source=data_source,
+                    signal=row["signal"],
+                    start_day=(row["max_time"] -
+                               gap_window).strftime("%Y-%m-%d"),
+                    end_day=row["max_time"].strftime("%Y-%m-%d"),
+                    geo_type=row["geo_type"])
+
         latest_data = covidcast.signal(
             data_source, row["signal"],
             start_day=row["max_time"] - gap_window,

sir_complainsalot/delphi_sir_complainsalot/run.py

@@ -5,6 +5,8 @@
 when the module is run with `python -m delphi_sir_complainsalot`.
 """
 
+import logging
+import structlog
 import sys
 
 from itertools import groupby
@@ -13,10 +15,12 @@
 from slack.errors import SlackApiError
 
 from delphi_utils import read_params
+from delphi_utils import get_structured_logger
 import covidcast
 
 from .check_source import check_source
 
+logger = get_structured_logger(__name__)
 
 def run_module():
 
@@ -30,7 +34,12 @@ def run_module():
 
     if len(complaints) > 0:
         for complaint in complaints:
-            print(complaint)
+            logger.critical(event="signal out of SLA",
+                            message=complaint.message,
+                            data_source=complaint.data_source,
+                            signal=complaint.signal,
+                            geo_types=complaint.geo_types,
+                            last_updated=complaint.last_updated.strftime("%Y-%m-%d"))
 
         report_complaints(complaints, params)
 
@@ -46,14 +55,14 @@ def split_complaints(complaints, n=49):
 def report_complaints(all_complaints, params):
     """Post complaints to Slack."""
     if not params["slack_token"]:
-        print("\b (dry-run)")
+        logger.info("(dry-run)")
         return
 
     client = WebClient(token=params["slack_token"])
 
     for complaints in split_complaints(all_complaints):
         blocks = format_complaints_aggregated_by_source(complaints)
-        print(f"blocks: {len(blocks)}")
+        logger.info(f"blocks: {len(blocks)}")
         try:
             client.chat_postMessage(
                 channel=params["channel"],

validator/delphi_validator/report.py

@@ -0,0 +1,98 @@
+"""Validation output reports."""
+import sys
+from datetime import date, datetime
+from typing import List, Tuple
+
+class ValidationReport:
+    """Class for reporting the results of validation."""
+    def __init__(self, errors_to_suppress: List[Tuple[str]]):
+        """Initialize a ValidationReport.
+        Parameters
+        ----------
+        errors_to_suppress: List[Tuple[str]]
+            List of error identifications to ignore.
+
+        Attributes
+        ----------
+        errors_to_suppress: List[Tuple[str]]
+            See above
+        num_suppressed: int
+            Number of errors suppressed
+        total_checks: int
+            Number of validation checks performed
+        raised_errors: List[Exception]
+            Errors raised from validation failures
+        raised_warnings: List[Exception]
+            Warnings raised from validation execution
+        unsuppressed_errors: List[Exception]
+            Errors raised from validation failures not found in `self.errors_to_suppress`
+        """
+        self.errors_to_suppress = errors_to_suppress.copy()
+        self.num_suppressed = 0
+        self.total_checks = 0
+        self.raised_errors = []
+        self.raised_warnings = []
+        self.unsuppressed_errors = []
+
+    def add_raised_error(self, error):
+        """Add an error to the report.
+        Parameters
+        ----------
+        error: Exception
+            Error raised in validation
+
+        Returns
+        -------
+        None
+        """
+        self.raised_errors.append(error)
+        # Convert any dates in check_data_id to strings for the purpose of comparing
+        # to manually suppressed errors.
+        raised_check_id = tuple([
+            item.strftime("%Y-%m-%d") if isinstance(item, (date, datetime))
+            else item for item in error.check_data_id])
+
+        if raised_check_id in self.errors_to_suppress:
+            self.errors_to_suppress.remove(raised_check_id)
+            self.num_suppressed += 1
+        else:
+            self.unsuppressed_errors.append(error)
+
+    def increment_total_checks(self):
+        """Records a check."""
+        self.total_checks += 1
+
+    def add_raised_warning(self, warning):
+        """Add a warning to the report.
+        Parameters
+        ----------
+        warning: Warning
+            Warning raised in validation
+
+        Returns
+        -------
+        None
+        """
+        self.raised_warnings.append(warning)
+
+    def __str__(self):
+        """String representation of report."""
+        out_str = f"{self.total_checks} checks run\n"
+        out_str += f"{len(self.unsuppressed_errors)} checks failed\n"
+        out_str += f"{self.num_suppressed} checks suppressed\n"
+        out_str += f"{len(self.raised_warnings)} warnings\n"
+        for message in self.unsuppressed_errors:
+            out_str += f"{message}\n"
+        for message in self.raised_warnings:
+            out_str += f"{message}\n"
+        return out_str
+
+    def print_and_exit(self):
+        """
+        Print results and, if any not-suppressed exceptions were raised, exit with non-zero status.
+        """
+        print(self)
+        if len(self.unsuppressed_errors) != 0:
+            sys.exit(1)
+        else:
+            sys.exit(0)

validator/delphi_validator/run.py

@@ -14,4 +14,4 @@ def run_module():
     params = parent_params['validation']
 
     validator = Validator(params)
-    validator.validate(parent_params["export_dir"])
+    validator.validate(parent_params["export_dir"]).print_and_exit()