removing wrapper and directly converting in other places and moving test as a script

Author: aysim319

_delphi_utils_python/delphi_utils/__init__.py

@@ -4,7 +4,6 @@
 from __future__ import absolute_import
 
 from .archive import ArchiveDiffer, GitArchiveDiffer, S3ArchiveDiffer
-from .covidcast_wrapper import metadata, signal
 from .export import create_export_csv
 from .geomap import GeoMapper
 from .logger import get_structured_logger

_delphi_utils_python/delphi_utils/validator/datafetcher.py

@@ -6,17 +6,42 @@
 import warnings
 from os import listdir
 from os.path import isfile, join
+from typing import Union
 
 import numpy as np
 import pandas as pd
 import requests
-from delphi_utils import covidcast_wrapper
+from delphi_epidata import Epidata
+from epiweeks import Week
 
 from .errors import APIDataFetchError, ValidationFailure
 
 FILENAME_REGEX = re.compile(
     r'^(?P<date>\d{8})_(?P<geo_type>\w+?)_(?P<signal>\w+)\.csv$')
 
+def _parse_datetimes(date_int: int, time_type: str, date_format: str = "%Y%m%d") -> Union[pd.Timestamp, None]:
+    """Convert a date or epiweeks string into timestamp objects.
+
+    Datetimes (length 8) are converted to their corresponding date, while epiweeks (length 6)
+    are converted to the date of the start of the week. Returns nan otherwise
+
+    Epiweeks use the CDC format.
+
+    date_int: Int representation of date.
+    time_type: The temporal resolution to request this data. Most signals
+      are available at the "day" resolution (the default); some are only
+      available at the "week" resolution, representing an MMWR week ("epiweek").
+    date_format: String of the date format to parse.
+    :returns: Timestamp.
+    """
+    date_str = str(date_int)
+    if time_type == "day":
+        return pd.to_datetime(date_str, format=date_format)
+    if time_type == "week":
+        epiwk = Week(int(date_str[:4]), int(date_str[-2:]))
+        return pd.to_datetime(epiwk.startdate())
+    return None
+
 def make_date_filter(start_date, end_date):
     """
     Create a function to filter dates in the specified date range (inclusive).
@@ -118,7 +143,16 @@ def get_geo_signal_combos(data_source, api_key):
     source_signal_mappings = {i['source']:i['db_source'] for i in
         meta_response.json()}
 
-    meta = covidcast_wrapper.metadata()
+    response = Epidata.covidcast_meta()
+
+    if response["result"] != 1:
+        # Something failed in the API and we did not get real metadata
+        raise RuntimeError("Error when fetching metadata from the API", response["message"])
+
+    meta = pd.DataFrame.from_dict(response["epidata"])
+    meta["min_time"] = meta.apply(lambda x: _parse_datetimes(x.min_time, x.time_type), axis=1)
+    meta["max_time"] = meta.apply(lambda x: _parse_datetimes(x.max_time, x.time_type), axis=1)
+    meta["last_update"] = pd.to_datetime(meta["last_update"], unit="s")
 
     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.
@@ -162,8 +196,28 @@ 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():
-        api_df = covidcast_wrapper.signal(data_source, signal_type, start_date, end_date, geo_type)
+    if start_date > end_date:
+        raise ValueError(
+            "end_day must be on or after start_day, but " f"start_day = '{start_date}', end_day = '{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=geo_type,
+    )
+    if response["result"] != 1:
+        # Something failed in the API and we did not get real metadata
+        raise RuntimeError("Error when fetching signal data from the API", response["message"])
+
+    api_df = pd.DataFrame.from_dict(response["epidata"])
+    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
 
 
     error_context = f"when fetching reference data from {start_date} to {end_date} " +\

_delphi_utils_python/tests/test_covidcast_wrapper.py

@@ -5,20 +5,16 @@
 when the module is run with `python -m delphi_google_symptoms`.
 """
 import time
-from datetime import datetime, date
+from datetime import date, datetime
 from itertools import product
-import covidcast
 
 import numpy as np
-from pandas import to_datetime
-from delphi_utils import (
-    create_export_csv,
-    get_structured_logger
-)
+from delphi_epidata import Epidata
+from delphi_utils import create_export_csv, get_structured_logger
 from delphi_utils.validator.utils import lag_converter
+from pandas import to_datetime
 
-from .constants import (COMBINED_METRIC,
-                        GEO_RESOLUTIONS, SMOOTHERS, SMOOTHERS_MAP)
+from .constants import COMBINED_METRIC, GEO_RESOLUTIONS, SMOOTHERS, SMOOTHERS_MAP
 from .geo import geo_map
 from .pull import pull_gs_data
 
@@ -66,8 +62,7 @@ def run_module(params):
         )
 
         # Fetch metadata to check how recent each signal is
-        covidcast.use_api_key(params["indicator"]["api_credentials"])
-        metadata = covidcast.metadata()
+        metadata = Epidata.covidcast_meta()
         # Filter to only those we currently want to produce, ignore any old or deprecated signals
         gs_metadata = metadata[(metadata.data_source == "google-symptoms") &
             (metadata.signal.isin(sensor_names))]

_delphi_utils_python/tests/test_data/covidcast_metadata.pkl

@@ -6,6 +6,7 @@
 
 import covidcast
 import pandas as pd
+from delphi_epidata import Epidata
 
 covidcast.covidcast._ASYNC_CALL = True  # pylint: disable=protected-access
 
@@ -64,22 +65,36 @@ def check_source(data_source, meta, params, grace, logger):  # pylint: disable=t
     age_complaints = {}
     gap_complaints = {}
 
+    start_date = datetime.now() - timedelta(days=14)
+    end_date = datetime.now()
+
     for _, row in signals.iterrows():
         logger.info("Retrieving signal",
             data_source=data_source,
             signal=row["signal"],
-            start_day=(datetime.now() - timedelta(days = 14)).strftime("%Y-%m-%d"),
-            end_day=datetime.now().strftime("%Y-%m-%d"),
+            start_day=start_date.strftime("%Y-%m-%d"),
+            end_day=end_date.strftime("%Y-%m-%d"),
             geo_type=row["geo_type"])
 
-        latest_data = covidcast.signal(
-            data_source, row["signal"],
-            start_day=datetime.now() - timedelta(days = 14),
-            end_day=datetime.now(),
-            geo_type=row["geo_type"]
+        response = Epidata.covidcast(
+            data_source,
+            row["signal"],
+            time_type=row["time_type"],
+            geo_type=row["geo_type"],
+            time_values=Epidata.range(start_date.strftime("%Y%m%d"), end_date.strftime("%Y%m%d")),
+            geo_value="*",
         )
 
-        current_lag_in_days = (now - row["max_time"]).days
+        if response["result"] != 1:
+            # Something failed in the API and we did not get real metadata
+            raise RuntimeError("Error when fetching signal data from the API", response["message"])
+
+        latest_data = pd.DataFrame.from_dict(response["epidata"])
+        latest_data["issue"] = pd.to_datetime(latest_data["issue"], format="%Y%m%d")
+        latest_data["time_value"] = pd.to_datetime(latest_data["time_value"], format="%Y%m%d")
+        latest_data.drop("direction", axis=1, inplace=True)
+
+        current_lag_in_days = (now - datetime.strptime(str(row["max_time"]), "%Y%m%d")).days
         lag_calculated_from_api = False
 
         if latest_data is not None:

google_symptoms/delphi_google_symptoms/run.py

@@ -8,10 +8,9 @@
 import time
 from itertools import groupby
 
-import covidcast
-from delphi_utils import SlackNotifier
-from delphi_utils import get_structured_logger
-from delphi_utils import read_params
+import pandas as pd
+from delphi_epidata import Epidata
+from delphi_utils import SlackNotifier, get_structured_logger, read_params
 
 from .check_source import check_source
 
@@ -29,8 +28,8 @@ def run_module():
     """Run SirCAL."""
     start_time = time.time()
     params = read_params()
-    covidcast.use_api_key(params["api_credentials"])
-    meta = covidcast.metadata()
+    Epidata.auth = ("epidata", params["api_credentials"])
+    meta = pd.DataFrame.from_dict(Epidata.covidcast_meta().get("epidata", dict()))
     slack_notifier = None
     if "channel" in params and "slack_token" in params:
         slack_notifier = SlackNotifier(params["channel"], params["slack_token"])

sir_complainsalot/delphi_sir_complainsalot/check_source.py

@@ -1,37 +1,20 @@
-"""module for covidcast api call wrapper."""
-
-from datetime import date, timedelta
-from typing import Iterable, Union
-
+"""
+script to check converting covidcast api calls with Epidata.covidcast Epidata.covidcast_meta
+"""
+from typing import Union, Iterable
+from datetime import datetime, timedelta, date
 import pandas as pd
+import covidcast
 from delphi_epidata import Epidata
+from pandas.testing import assert_frame_equal
+import os
 from epiweeks import Week
 
+API_KEY = os.environ.get('DELPHI_API_KEY')
+covidcast.use_api_key(API_KEY)
 
-def date_generator(startdate: date, enddate: date, time_type: str) -> Iterable[date]:
-    """
-    Take start date and end date and generates date string.
-
-    Parameters
-    ----------
-    startdate: date
-    enddate: date
-    time_type: str
-
-    Returns
-    -------
-    generator of str
-    """
-    if time_type.lower() == "day":
-        while startdate <= enddate:
-            yield startdate.strftime("%Y-%m-%d")
-            startdate = startdate + timedelta(days=1)
-    elif time_type.lower() == "week":
-        while startdate <= enddate:
-            epiweek = Week.fromdate(startdate)
-            yield epiweek
-            startdate = startdate + timedelta(days=7)
-
+Epidata.debug = True
+Epidata.auth = ("epidata", API_KEY)
 
 def _parse_datetimes(date_int: int, time_type: str, date_format: str = "%Y%m%d") -> Union[pd.Timestamp, None]:
     """Convert a date or epiweeks string into timestamp objects.
@@ -57,7 +40,7 @@ def _parse_datetimes(date_int: int, time_type: str, date_format: str = "%Y%m%d")
     return None
 
 
-def metadata() -> Union[pd.DataFrame, None]:
+def ported_metadata() -> Union[pd.DataFrame, None]:
     """
     Make covidcast metadata api call.
 
@@ -78,7 +61,7 @@ def metadata() -> Union[pd.DataFrame, None]:
     return df
 
 
-def signal(
+def ported_signal(
     data_source: str,
     signal: str,  # pylint: disable=W0621
     start_day: date = None,
@@ -132,14 +115,12 @@ def signal(
             "end_day must be on or after start_day, but " f"start_day = '{start_day}', end_day = '{end_day}'"
         )
 
-    time_values = list(date_generator(start_day, end_day))
-
     response = Epidata.covidcast(
         data_source,
         signal,
         time_type=time_type,
         geo_type=geo_type,
-        time_values=time_values,
+        time_values=Epidata.range(start_day.strftime("%Y%m%d"), end_day.strftime("%Y%m%d")),
         geo_value=geo_values,
         as_of=as_of,
         lag=lag,
@@ -156,3 +137,33 @@ def signal(
     api_df["signal"] = signal
 
     return api_df
+
+def check_metadata():
+    expected_df = covidcast.metadata()
+    df = ported_metadata().metadata()
+    assert_frame_equal(expected_df, df)
+
+def check_signal():
+    meta_df = covidcast.metadata()
+    startdate = datetime(year=2022, month=2, day=1)
+    data_filter = (meta_df["max_time"] >= startdate)
+    signal_df = meta_df[data_filter].groupby("data_source")["signal"].agg(['unique'])
+    enddate = startdate + timedelta(days=3)
+
+    for data_source, row in signal_df.iterrows():
+        signals = list(row[0])
+        for signal in signals:
+            expected_df = covidcast.signal(data_source, signal, start_day=startdate, end_day=enddate, geo_type="state")
+            if expected_df is None:
+                print("%s %s %s %s not existing", data_source, signal, startdate, enddate)
+
+            df = ported_signal(data_source, signal, start_day=startdate, end_day=enddate, geo_type="state")
+
+            check = df.merge(expected_df, indicator=True)
+            assert (check["_merge"] == "both").all()
+            assert check["_left_only"].empty
+            assert check["_right_only"].empty
+
+if __name__ == "__main__":
+    # check_metadata()
+    check_signal()