sircomplainslot needs more filtering

Author: aysim319

sir_complainsalot/delphi_sir_complainsalot/check_source.py

@@ -8,7 +8,8 @@
 import pandas as pd
 from delphi_epidata import Epidata
 
-covidcast.covidcast._ASYNC_CALL = True  # pylint: disable=protected-access
+from .date_utils import _date_to_api_string, _parse_datetimes
+
 
 @dataclass
 class Complaint:
@@ -34,6 +35,7 @@ def to_md(self):
             message=self.message, updated=self.last_updated.strftime("%Y-%m-%d"))
 
 
+
 def check_source(data_source, meta, params, grace, logger):  # pylint: disable=too-many-locals
     """Iterate over all signals from a source and check for problems.
 
@@ -74,30 +76,28 @@ def check_source(data_source, meta, params, grace, logger):  # pylint: disable=t
             signal=row["signal"],
             start_day=start_date.strftime("%Y-%m-%d"),
             end_day=end_date.strftime("%Y-%m-%d"),
-            geo_type=row["geo_type"])
+            geo_type=row["geo_type"],
+            time_type=row["time_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")),
+            time_values=Epidata.range(_date_to_api_string(start_date), _date_to_api_string(end_date)),
             geo_value="*",
         )
 
-        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
+        current_lag_in_days = (now - row["max_time"]).days
         lag_calculated_from_api = False
+        latest_data = None
+
+        if response["result"] == 1:
+            latest_data = pd.DataFrame.from_dict(response["epidata"])
+            latest_data["issue"] = latest_data.apply(lambda x: _parse_datetimes(x.issue, x.time_type), axis=1)
+            latest_data["time_value"] = latest_data.apply(lambda x: _parse_datetimes(x.time_value, x.time_type), axis=1)
+            latest_data.drop("direction", axis=1, inplace=True)
 
-        if latest_data is not None:
             unique_dates = [pd.to_datetime(val).date()
                             for val in latest_data["time_value"].unique()]
             current_lag_in_days = (datetime.now().date() - max(unique_dates)).days

sir_complainsalot/delphi_sir_complainsalot/date_utils.py

@@ -0,0 +1,35 @@
+from datetime import datetime
+from typing import Union
+
+from epiweeks import Week
+import pandas as pd
+def _date_to_api_string(date: datetime.date, time_type: str = "day") -> str:  # pylint: disable=W0621
+    """Convert a date object to a YYYYMMDD or YYYYMM string expected by the API."""
+    if time_type == "day":
+        date_str = date.strftime("%Y%m%d")
+    elif time_type == "week":
+        date_str = Week.fromdate(date).cdcformat()
+    return date_str
+
+def _parse_datetimes(date_int: str, 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

sir_complainsalot/delphi_sir_complainsalot/run.py

@@ -13,6 +13,7 @@
 from delphi_utils import SlackNotifier, get_structured_logger, read_params
 
 from .check_source import check_source
+from .date_utils import _parse_datetimes
 
 
 def get_logger():
@@ -30,6 +31,7 @@ def run_module():
     params = read_params()
     Epidata.auth = ("epidata", params["api_credentials"])
     meta = pd.DataFrame.from_dict(Epidata.covidcast_meta().get("epidata", dict()))
+    meta["max_time"] = meta.apply(lambda x: _parse_datetimes(x.max_time, x.time_type), axis=1)
     slack_notifier = None
     if "channel" in params and "slack_token" in params:
         slack_notifier = SlackNotifier(params["channel"], params["slack_token"])