test: tweak covidcast_port tests a bit

Author: dshemetov

testing_utils/delphi_utils/.gitignore

@@ -0,0 +1 @@
+covidcast_result

testing_utils/delphi_utils/check_covidcast_port.py

@@ -1,27 +1,30 @@
 """
 script to check converting covidcast api calls with Epidata.covidcast Epidata.covidcast_meta
 """
-import time
+
 from collections import defaultdict
 from pathlib import Path
 from typing import Union, Iterable, Tuple, List, Dict
 from datetime import datetime, timedelta, date
 
-import numpy as np
 import pandas as pd
 import covidcast
+import tqdm
 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')
+API_KEY = os.environ.get("DELPHI_API_KEY", os.environ.get("DELPHI_EPIDATA_KEY"))
 covidcast.use_api_key(API_KEY)
-
-Epidata.debug = True
 Epidata.auth = ("epidata", API_KEY)
-
 CURRENT_DIR = Path(__file__).parent
+if not Path(f"{CURRENT_DIR}/covidcast_result").is_dir():
+    os.mkdir(f"{CURRENT_DIR}/covidcast_result")
+# We will test the first X signals for each data source that we find from the
+# metadata endpoint with this variable.
+NUMBER_SIGNALS_PER_SOURCE = 5
+
 
 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.
@@ -151,22 +154,23 @@ def ported_signal(
 
     return api_df
 
-def check_metadata():
 
+def check_metadata():
     expected_df = covidcast.metadata()
     df = ported_metadata()
     assert_frame_equal(expected_df, df)
 
+
 def ported_signal(
-        data_source: str,
-        signal: str,  # pylint: disable=W0621
-        start_day: date = None,
-        end_day: date = None,
-        geo_type: str = "county",
-        geo_values: Union[str, Iterable[str]] = "*",
-        as_of: date = None,
-        lag: int = None,
-        time_type: str = "day",
+    data_source: str,
+    signal: str,  # pylint: disable=W0621
+    start_day: date = None,
+    end_day: date = None,
+    geo_type: str = "county",
+    geo_values: Union[str, Iterable[str]] = "*",
+    as_of: date = None,
+    lag: int = None,
+    time_type: str = "day",
 ) -> Union[pd.DataFrame, None]:
     """
     Makes covidcast signal api call.
@@ -252,13 +256,16 @@ def generate_start_date_per_signal() -> Dict[Tuple[datetime, datetime, str], Lis
     Dict[Tuple[datetime.datetime, datetime.datetime, str],[List[Tuple[str, str]]]
     """
     meta_df = pd.DataFrame.from_dict(Epidata.covidcast_meta()["epidata"])
-    meta_df["min_time"] = meta_df["min_time"].astype('str')
+    meta_df["min_time"] = meta_df["min_time"].astype("str")
+    meta_df = meta_df.groupby("data_source").head(NUMBER_SIGNALS_PER_SOURCE)
     signal_timeframe_dict = defaultdict(list)
 
     for start_str, data in meta_df.groupby("min_time"):
-
         data_source_groups = data.groupby("data_source")
         for data_source, df in data_source_groups:
+            # TODO: Remove after metadata bug is fixed.
+            if data_source == "google-symptom":
+                continue
             signals = list(df["signal"].unique())
             time_type = df["time_type"].values[0]
             for signal in signals:
@@ -274,8 +281,7 @@ def generate_start_date_per_signal() -> Dict[Tuple[datetime, datetime, str], Lis
                 elif time_type == "week":
                     start_time = Week(year=int(start_str[:4]), week=int(start_str[-2:]))
                     end_time = (start_time + 2).startdate()
-                    date_range = (start_time.startdate(),
-                                  end_time, time_type)
+                    date_range = (start_time.startdate(), end_time, time_type)
                     signal_timeframe_dict[date_range].append((data_source, signal))
 
     return signal_timeframe_dict
@@ -289,39 +295,51 @@ def check_signal():
     """
     signal_timeframe_dict = generate_start_date_per_signal()
     signal_df_dict = dict()
-    for date_range, data_source_signal_list in signal_timeframe_dict.items():
+    for date_range, data_source_signal_list in tqdm.tqdm(signal_timeframe_dict.items()):
         for data_source, signal in data_source_signal_list:
             time_type = date_range[2]
             filename = f"{CURRENT_DIR}/covidcast_result/{data_source}_{signal}.parquet"
             if not Path(filename).is_file():
                 # every signal except google-symptom has geo type of state
                 geo_type = "state"
-                if data_source == "google-symptom":
+                if data_source == "google-symptoms":
                     geo_type = "county"
-
-                expected_df = covidcast.signal(data_source, signal, start_day=date_range[0], end_day=date_range[1],
-                                           geo_type=geo_type, time_type=time_type)
-                if expected_df is None:
-                    raise RuntimeError("Data should exists")
+                expected_df = covidcast.signal(
+                    data_source,
+                    signal,
+                    start_day=date_range[0],
+                    end_day=date_range[1],
+                    geo_type=geo_type,
+                    time_type=time_type,
+                )
+                assert not expected_df.empty, "Received no data from covidcast API."
 
                 expected_df.to_parquet(filename)
             signal_df_dict[(data_source, signal, time_type)] = filename
 
-    for date_range, data_source_signal_list in signal_timeframe_dict.items():
+    for date_range, data_source_signal_list in tqdm.tqdm(signal_timeframe_dict.items()):
         for data_source, signal in data_source_signal_list:
             expected_filename = signal_df_dict.get((data_source, signal, date_range[2]))
             expected_df = pd.read_parquet(expected_filename)
 
             # every signal except google-symptom has geo type of state
             geo_type = "state"
-            if data_source == "google-symptom":
+            if data_source == "google-symptoms":
                 geo_type = "county"
-            df = ported_signal(data_source, signal, start_day=date_range[0], end_day=date_range[1],
-                               time_type=date_range[2],
-                               geo_type=geo_type)
+            df = ported_signal(
+                data_source,
+                signal,
+                start_day=date_range[0],
+                end_day=date_range[1],
+                time_type=date_range[2],
+                geo_type=geo_type,
+            )
+            assert not df.empty, "Received no data from covidcast API."
+
             check = df.merge(expected_df, indicator=True)
             assert (check["_merge"] == "both").all()
 
+
 if __name__ == "__main__":
     check_metadata()
-    check_signal()
+    check_signal()

testing_utils/delphi_utils/requirements.txt

@@ -1,12 +1,7 @@
-boto3
 covidcast
 delphi-epidata
-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
 epiweeks
-importlib_resources>=1.3
-moto~=4.2.14
 numpy
-pandas>=1.1.0
-pylint==2.8.3
-structlog
-xlrd
+pandas==1.5.3
+pyarrow
+tqdm