1972 replace covidcast

_delphi_utils_python/delphi_utils/__init__.py

@@ -4,6 +4,7 @@
 from __future__ import absolute_import
 
 from .archive import ArchiveDiffer, GitArchiveDiffer, S3ArchiveDiffer
+from .date_utils import convert_apitime_column_to_datetimes, date_to_api_string
 from .export import create_backup_csv, create_export_csv
 from .geomap import GeoMapper
 from .logger import get_structured_logger

_delphi_utils_python/delphi_utils/date_utils.py

@@ -0,0 +1,38 @@
+"""Utility for converting dates to a format accepted by epidata api."""
+
+from datetime import datetime
+
+import pandas as pd
+from epiweeks import Week
+
+
+def date_to_api_string(d: datetime, time_type: str = "day") -> str:
+    """Convert a date object to a YYYYMMDD or YYYYMM string expected by the API."""
+    if time_type == "day":
+        return d.strftime("%Y%m%d")
+    if time_type == "week":
+        return Week.fromdate(d).cdcformat()
+    raise ValueError(f"Unknown time_type: {time_type}")
+
+
+def convert_apitime_column_to_datetimes(
+    df: pd.DataFrame, col: str, date_format: str = "%Y%m%d"
+) -> pd.Series:
+    """Convert a DataFrame date or epiweek column into datetimes.
+
+    Dates are assumed to be in the YYYYMMDD format by default.
+    Weeks are assumed to be in the epiweek CDC format YYYYWW
+    format and return the date of the first day of the week.
+    """
+    df[col] = df[col].astype("str")
+
+    def parse_row(row):
+        if row["time_type"] == "day":
+            return pd.to_datetime(row[col], format=date_format)
+        if row["time_type"] == "week":
+            return pd.to_datetime(
+                Week(int(row[col][:4]), int(row[col][-2:])).startdate()
+            )
+        return row[col]
+
+    return df.apply(parse_row, axis=1)

_delphi_utils_python/delphi_utils/validator/datafetcher.py

@@ -3,13 +3,17 @@
 
 import re
 import threading
+import warnings
 from os import listdir
+
+# pylint: disable=W0707
 from os.path import isfile, join
-import warnings
-import requests
-import pandas as pd
+
 import numpy as np
-import covidcast
+import pandas as pd
+import requests
+from delphi_epidata import Epidata
+
 from .errors import APIDataFetchError, ValidationFailure
 
 FILENAME_REGEX = re.compile(
@@ -115,7 +119,13 @@ def get_geo_signal_combos(data_source, api_key):
     meta_response.raise_for_status()
     source_signal_mappings = {i['source']:i['db_source'] for i in
         meta_response.json()}
-    meta = covidcast.metadata()
+
+    response = Epidata.covidcast_meta()
+
+    meta = pd.DataFrame.from_dict(Epidata.check(response))
+    # note: this will fail for signals with weekly data, but currently not supported for validation
+    meta = meta[meta["time_type"] == "day"]
+
     source_meta = meta[meta['data_source'] == data_source]
     # Need to convert np.records to tuples so they are hashable and can be used in sets and dicts.
     geo_signal_combos = list(map(tuple,
@@ -158,18 +168,40 @@ def fetch_api_reference(data_source, start_date, end_date, geo_type, signal_type
 
     Formatting is changed to match that of source data CSVs.
     """
-    with warnings.catch_warnings():
-        warnings.simplefilter("ignore")
-        api_df = covidcast.signal(
-            data_source, signal_type, start_date, end_date, geo_type)
+    if start_date > end_date:
+        raise ValueError(
+            "end_date must be on or after start_date, but "
+            + f"start_date = '{start_date}', end_date = '{end_date}'"
+        )
+    response = Epidata.covidcast(
+        data_source,
+        signal_type,
+        time_type="day",
+        geo_type=geo_type,
+        time_values=Epidata.range(
+            start_date.strftime("%Y%m%d"), end_date.strftime("%Y%m%d")
+        ),
+        geo_value="*",
+    )
 
-    error_context = f"when fetching reference data from {start_date} to {end_date} " +\
-        f"for data source: {data_source}, signal type: {signal_type}, geo type: {geo_type}"
+    try:
+        epidata_dict = Epidata.check(response)
+    except Exception as e:
+        raise APIDataFetchError(str(e))
 
-    if api_df is None:
+    if len(response["epidata"]) == 0:
+        error_context = f"when fetching reference data from {start_date} to {end_date} " + \
+                        f"for data source: {data_source}, signal type: {signal_type}, geo type: {geo_type}"
         raise APIDataFetchError("Error: no API data was returned " + error_context)
-    if not isinstance(api_df, pd.DataFrame):
-        raise APIDataFetchError("Error: API return value was not a dataframe " + error_context)
+
+    api_df = pd.DataFrame.from_dict(epidata_dict)
+    # note: this will fail for signals with weekly data, but currently not supported for validation
+    api_df["issue"] = pd.to_datetime(api_df["issue"], format="%Y%m%d")
+    api_df["time_value"] = pd.to_datetime(api_df["time_value"], format="%Y%m%d")
+    api_df.drop("direction", axis=1, inplace=True)
+    api_df["data_source"] = data_source
+    api_df["signal"] = signal_type
+
 
     column_names = ["geo_id", "val",
                     "se", "sample_size", "time_value"]

_delphi_utils_python/delphi_utils/validator/dynamic.py

@@ -1,14 +1,16 @@
 """Dynamic file checks."""
+
+import re
 from dataclasses import dataclass
 from datetime import date, timedelta
 from typing import Dict, Set
-import re
-import pandas as pd
+
 import numpy as np
-import covidcast
-from .errors import ValidationFailure
+import pandas as pd
+
 from .datafetcher import get_geo_signal_combos, threaded_api_calls
-from .utils import relative_difference_by_min, TimeWindow, lag_converter
+from .errors import ValidationFailure
+from .utils import TimeWindow, lag_converter, relative_difference_by_min
 
 
 class DynamicValidator:
@@ -78,8 +80,6 @@ def validate(self, all_frames, report):
         # Get 14 days prior to the earliest list date
         outlier_lookbehind = timedelta(days=14)
 
-        # Authenticate API
-        covidcast.use_api_key(self.params.api_key)
 
         # Get all expected combinations of geo_type and signal.
         geo_signal_combos = get_geo_signal_combos(self.params.data_source,

_delphi_utils_python/delphi_utils/validator/run.py

@@ -5,8 +5,10 @@
 when the module is run with `python -m delphi_utils.validator`.
 """
 import argparse as ap
-import covidcast
-from .. import read_params, get_structured_logger
+
+from delphi_epidata import Epidata
+
+from .. import get_structured_logger, read_params
 from .validate import Validator
 
 
@@ -18,7 +20,7 @@ def run_module():
     args = parser.parse_args()
     params = read_params()
     assert "validation" in params
-    covidcast.use_api_key(params["validation"]["common"]["api_credentials"])
+    Epidata.auth = ("epidata", params["validation"]["common"]["api_credentials"])
     dry_run_param = params["validation"]["common"].get("dry_run", False)
     params["validation"]["common"]["dry_run"] = args.dry_run or dry_run_param
     validator = Validator(params)

_delphi_utils_python/pyproject.toml

@@ -17,13 +17,13 @@ classifiers = [
 ]
 dependencies = [
     "boto3",
-    "covidcast",
     "cvxpy",
+    "delphi-epidata",
     "epiweeks",
     "gitpython",
     "importlib_resources>=1.3",
     "numpy",
-    "pandas>=1.1.0",
+    "pandas==1.5.3",
     "requests",
     "slackclient",
     "scs<3.2.6",                # TODO: remove this ; it is a cvxpy dependency, and the excluded version appears to break our jenkins build. see: https://github.com/cvxgrp/scs/issues/283

_delphi_utils_python/tests/test_data/sample_epidata_metadata.json

@@ -0,0 +1,28 @@
+{"data_source": ["chng", "chng", "chng",
+                                                    "covid-act-now",
+                                                    "covid-act-now",
+                                                    "covid-act-now",
+                                                    "chng"],
+                                    "signal": ["smoothed_outpatient_cli",
+                                               "smoothed_outpatient_covid",
+                                               "smoothed_outpatient_covid",
+                                               "pcr_specimen_positivity_rate",
+                                               "pcr_specimen_positivity_rate",
+                                               "pcr_specimen_total_tests",
+                                               "inactive"],
+                                    "geo_type": ["state", "state", "county",
+                                                 "hrr", "msa", "msa",
+                                                 "state"],
+                                   "min_time": ["20200101", "20200101", "20200101",
+                                                "20200101", "20200101", "20200101",
+                                                "20200101"],
+                                   "max_time": ["20240101", "20240101", "20240101",
+                                                "20240101", "20240101", "20240101",
+                                                "20240101"],
+                                   "last_update": [1711963480, 1711963480, 1711963480,
+                                    1711963480, 1711963480, 1711963480,
+                                    1711963480],
+                                   "time_type": ["day", "day", "day",
+                                                   "day", "day", "day",
+                                                   "day"]
+                                    }

_delphi_utils_python/tests/test_data/sample_epidata_signal_a.json

@@ -0,0 +1,9 @@
+[{"geo_value": "1044",
+  "stderr": null,
+  "value": 3,
+  "issue": 20200101,
+  "lag": 7,
+  "sample_size": null,
+  "time_value": 20200101,
+  "direction": null
+}]

_delphi_utils_python/tests/test_data/sample_epidata_signal_county.json

@@ -0,0 +1,9 @@
+[{"geo_value": "0888",
+"stderr": 2,
+"value": 14,
+"issue": 20200101,
+"lag": 1,
+"sample_size": 100,
+"time_value": 20200101,
+"direction": null
+}]