add RVDSS endpoint and python client support

Author: nmdefries

src/client/delphi_epidata.py

@@ -675,6 +675,48 @@ def covid_hosp_facility_lookup(state=None, ccn=None, city=None, zip=None, fips_c
         # Make the API call
         return Epidata._request("covid_hosp_facility_lookup", params)
 
+    # Fetch Canadian respiratory RVDSS data
+    @staticmethod
+    def rvdss(
+        geo_type,
+        time_values,
+        geo_value,
+        as_of=None,
+        issues=None,
+    ):
+        """Fetch RVDSS data."""
+        # Check parameters
+        if None in (geo_type, time_values, geo_value):
+            raise EpidataBadRequestException(
+                "`geo_type`, `time_values`, and `geo_value` are all required"
+            )
+
+        # Set up request
+        params = {
+            "data_source": data_source,
+            "signals": Epidata._list(signals),
+            "time_type": time_type,
+            "geo_type": geo_type,
+            "time_values": Epidata._list(time_values),
+            # Fake a time type param so that we can use some helper functions later.
+            "time_type": "week",
+        }
+
+        if isinstance(geo_value, (list, tuple)):
+            params["geo_values"] = ",".join(geo_value)
+        else:
+            params["geo_value"] = geo_value
+        if as_of is not None:
+            params["as_of"] = as_of
+        if issues is not None:
+            params["issues"] = Epidata._list(issues)
+        # if lag is not None:
+        #     params["lag"] = lag
+
+        # Make the API call
+        return Epidata._request("rvdss", params)
+
+
     @staticmethod
     def async_epidata(param_list, batch_size=50):
         """[DEPRECATED] Make asynchronous Epidata calls for a list of parameters."""

src/server/_query.py

@@ -483,15 +483,21 @@ def apply_issues_filter(self, history_table: str, issues: Optional[TimeValues])
                 self.where_integers("issue", issues)
         return self
 
-    def apply_as_of_filter(self, history_table: str, as_of: Optional[int]) -> "QueryBuilder":
+    def apply_as_of_filter(self, history_table: str, as_of: Optional[int], use_source_signal: Optional[bool] = TRUE) -> "QueryBuilder":
         if as_of is not None:
             self.retable(history_table)
             sub_condition_asof = "(issue <= :as_of)"
             self.params["as_of"] = as_of
-            sub_fields = "max(issue) max_issue, time_type, time_value, `source`, `signal`, geo_type, geo_value"
-            sub_group = "time_type, time_value, `source`, `signal`, geo_type, geo_value"
+
             alias = self.alias
-            sub_condition = f"x.max_issue = {alias}.issue AND x.time_type = {alias}.time_type AND x.time_value = {alias}.time_value AND x.source = {alias}.source AND x.signal = {alias}.signal AND x.geo_type = {alias}.geo_type AND x.geo_value = {alias}.geo_value"
+            source_signal_plain = source_signal_alias = ""
+            if use_source_signal:
+                source_signal_plain = ", `source`, `signal`"
+                source_signal_alias = f"AND x.source = {alias}.source AND x.signal = {alias}.signal"
+
+            sub_fields = f"max(issue) max_issue, time_type, time_value{source_signal_plain}, geo_type, geo_value"
+            sub_group = f"time_type, time_value{source_signal_plain}, geo_type, geo_value"
+            sub_condition = f"x.max_issue = {alias}.issue AND x.time_type = {alias}.time_type AND x.time_value = {alias}.time_value{source_signal_alias} AND x.geo_type = {alias}.geo_type AND x.geo_value = {alias}.geo_value"
             self.subquery = f"JOIN (SELECT {sub_fields} FROM {self.table} WHERE {self.conditions_clause} AND {sub_condition_asof} GROUP BY {sub_group}) x ON {sub_condition}"
         return self
 

src/server/endpoints/rvdss.py

@@ -0,0 +1,108 @@
+from flask import Blueprint, request
+
+from .._params import extract_integers, extract_strings, extract_date, extract_dates
+from .._query import execute_query, QueryBuilder
+from .._validate import require_all
+
+bp = Blueprint("rvdss", __name__)
+
+db_table_name = "rvdss"
+
+def parse_time_set_epiweeks() -> TimeSet:
+    require_all(request, "epiweeks")
+    time_values = extract_dates("epiweeks")
+    if time_values == ["*"]:
+        return TimeSet("week", True)
+    return TimeSet("week", time_values)
+
+    return parse_time_arg()
+
+
+@bp.route("/", methods=("GET", "POST"))
+def handle():
+    require_all(request, "epiweeks", "geo_type", "geo_values")
+
+    # epiweeks = extract_integers("epiweeks")
+    time_set = parse_time_set_epiweeks()
+    geo_sets = parse_geo_sets()
+    # issues = extract_integers("issues") # TODO
+    issues = extract_dates("issues")
+    # as_of = extract_date("as_of")
+
+    # basic query info
+    q = QueryBuilder(db_table_name, "rv")
+
+    fields_string = [
+        "geo_type",
+        "geo_value",
+        "region",
+    ]
+    fields_int = [
+        "epiweek",
+        "week",
+        "weekorder",
+        "year",
+        "time_value",
+        "issue",
+    ]
+    fields_float = [
+        "sarscov2_tests",
+        "sarscov2_positive_tests",
+        "sarscov2_pct_positive",
+        "flu_tests",
+        "flua_tests",
+        "flub_tests",
+        "flu_positive_tests",
+        "flu_pct_positive",
+        "fluah1n1pdm09_positive_tests",
+        "fluah3_positive_tests",
+        "fluauns_positive_tests",
+        "flua_positive_tests",
+        "flua_pct_positive",
+        "flub_positive_tests",
+        "flub_pct_positive",
+        "rsv_tests",
+        "rsv_positive_tests",
+        "rsv_pct_positive",
+        "hpiv_tests",
+        "hpiv1_positive_tests",
+        "hpiv2_positive_tests",
+        "hpiv3_positive_tests",
+        "hpiv4_positive_tests",
+        "hpivother_positive_tests",
+        "hpiv_positive_tests",
+        "hpiv_pct_positive",
+        "adv_tests",
+        "adv_positive_tests",
+        "adv_pct_positive",
+        "hmpv_tests",
+        "hmpv_positive_tests",
+        "hmpv_pct_positive",
+        "evrv_tests",
+        "evrv_positive_tests",
+        "evrv_pct_positive",
+        "hcov_tests",
+        "hcov_positive_tests",
+        "hcov_pct_positive",
+    ]
+
+    q.set_sort_order("epiweek", "time_value" "geo_type", "geo_value", "issue")
+    q.set_fields(fields_string, fields_int, fields_float)
+
+    # q.where_strings("geo_type", geo_type)
+    q.apply_geo_filters("geo_type", "geo_value", geo_sets)
+    # TODO: can only use this if have a time_type field in data
+    # q.apply_time_filter("time_type", "time_value", time_set)
+    q.where_integers("epiweek", epiweeks)
+
+    q.apply_issues_filter(db_table_name, issues)
+    # TODO: can only use this if have a time_type field in data
+    # q.apply_as_of_filter(db_table_name, as_of, use_source_signal = FALSE)
+
+    # if issues is not None:
+    #     q.where_integers("issue", issues)
+    # else:
+    #     q.with_max_issue("epiweek", "geo_value")
+
+    # send query
+    return execute_query(str(q), q.params, fields_string, fields_int, fields_float)