Server-side compute: add /trend endpoint

Author: dshemetov

integrations/server/test_covidcast_endpoints.py

@@ -296,6 +296,14 @@ def test_compatibility(self):
             out = self._fetch(is_compatibility=True, source=first.source, signal=first.signal, geo=first.geo_pair, time="day:*")
             self.assertEqual(len(out["epidata"]), len(rows))
 
+    def _diff_covidcast_rows(self, rows: List[CovidcastRow]) -> List[CovidcastRow]:
+        new_rows = list()
+        for x, y in zip(rows[1:], rows[:-1]):
+            new_row = copy(x)
+            new_row.value = x.value - y.value
+            new_rows.append(new_row)
+        return new_rows
+
     def test_trend(self):
         """Request a signal the /trend endpoint."""
 
@@ -306,29 +314,66 @@ def test_trend(self):
         ref = rows[num_rows // 2]
         self._insert_rows(rows)
 
-        out = self._fetch("/trend", signal=first.signal_pair, geo=first.geo_pair, date=last.time_value, window="20200401-20201212", basis=ref.time_value)
+        with self.subTest("no server-side compute"):
+            out = self._fetch("/trend", signal=first.signal_pair, geo=first.geo_pair, date=last.time_value, window="20200401-20201212", basis=ref.time_value)
+
+            self.assertEqual(out["result"], 1)
+            self.assertEqual(len(out["epidata"]), 1)
+            trend = out["epidata"][0]
+            self.assertEqual(trend["geo_type"], last.geo_type)
+            self.assertEqual(trend["geo_value"], last.geo_value)
+            self.assertEqual(trend["signal_source"], last.source)
+            self.assertEqual(trend["signal_signal"], last.signal)
+
+            self.assertEqual(trend["date"], last.time_value)
+            self.assertEqual(trend["value"], last.value)
+
+            self.assertEqual(trend["basis_date"], ref.time_value)
+            self.assertEqual(trend["basis_value"], ref.value)
+            self.assertEqual(trend["basis_trend"], "increasing")
+
+            self.assertEqual(trend["min_date"], first.time_value)
+            self.assertEqual(trend["min_value"], first.value)
+            self.assertEqual(trend["min_trend"], "increasing")
+            self.assertEqual(trend["max_date"], last.time_value)
+            self.assertEqual(trend["max_value"], last.value)
+            self.assertEqual(trend["max_trend"], "steady")
+
+        num_rows = 30
+        time_value_pairs = [(20200331, 0)] + [(20200401 + i, v) for i, v in enumerate(accumulate(range(num_rows)))]
+        rows = [CovidcastRow(source="jhu-csse", signal="confirmed_cumulative_num", time_value=t, value=v) for t, v in time_value_pairs]
+        self._insert_rows(rows)
+        diffed_rows = self._diff_covidcast_rows(rows)
+        for row in diffed_rows:
+            row.signal = "confirmed_incidence_num"
+        first = diffed_rows[0]
+        last = diffed_rows[-1]
+        ref = diffed_rows[num_rows // 2]
+        with self.subTest("use server-side compute"):
+            out = self._fetch("/trend", signal="jhu-csse:confirmed_incidence_num", geo=first.geo_pair, date=last.time_value, window="20200401-20201212", basis=ref.time_value)
+
+            self.assertEqual(out["result"], 1)
+            self.assertEqual(len(out["epidata"]), 1)
+            trend = out["epidata"][0]
+            self.assertEqual(trend["geo_type"], last.geo_type)
+            self.assertEqual(trend["geo_value"], last.geo_value)
+            self.assertEqual(trend["signal_source"], last.source)
+            self.assertEqual(trend["signal_signal"], last.signal)
+
+            self.assertEqual(trend["date"], last.time_value)
+            self.assertEqual(trend["value"], last.value)
+
+            self.assertEqual(trend["basis_date"], ref.time_value)
+            self.assertEqual(trend["basis_value"], ref.value)
+            self.assertEqual(trend["basis_trend"], "increasing")
+
+            self.assertEqual(trend["min_date"], first.time_value)
+            self.assertEqual(trend["min_value"], first.value)
+            self.assertEqual(trend["min_trend"], "increasing")
+            self.assertEqual(trend["max_date"], last.time_value)
+            self.assertEqual(trend["max_value"], last.value)
+            self.assertEqual(trend["max_trend"], "steady")
 
-        self.assertEqual(out["result"], 1)
-        self.assertEqual(len(out["epidata"]), 1)
-        trend = out["epidata"][0]
-        self.assertEqual(trend["geo_type"], last.geo_type)
-        self.assertEqual(trend["geo_value"], last.geo_value)
-        self.assertEqual(trend["signal_source"], last.source)
-        self.assertEqual(trend["signal_signal"], last.signal)
-
-        self.assertEqual(trend["date"], last.time_value)
-        self.assertEqual(trend["value"], last.value)
-
-        self.assertEqual(trend["basis_date"], ref.time_value)
-        self.assertEqual(trend["basis_value"], ref.value)
-        self.assertEqual(trend["basis_trend"], "increasing")
-
-        self.assertEqual(trend["min_date"], first.time_value)
-        self.assertEqual(trend["min_value"], first.value)
-        self.assertEqual(trend["min_trend"], "increasing")
-        self.assertEqual(trend["max_date"], last.time_value)
-        self.assertEqual(trend["max_value"], last.value)
-        self.assertEqual(trend["max_trend"], "steady")
 
     def test_trendseries(self):
         """Request a signal the /trendseries endpoint."""

src/server/endpoints/covidcast.py

@@ -231,6 +231,8 @@ def handle_trend():
     daily_signals, weekly_signals = count_signal_time_types(source_signal_pairs)
     source_signal_pairs, alias_mapper = create_source_signal_alias_mapper(source_signal_pairs)
     geo_pairs = parse_geo_pairs()
+    transform_args = parse_transform_args()
+    jit_bypass = parse_jit_bypass()
 
     time_window, is_day = parse_day_or_week_range_arg("window")
     time_value, is_also_day = parse_day_or_week_arg("date")
@@ -244,6 +246,12 @@ def handle_trend():
             base_shift = 7
         basis_time_value = shift_time_value(time_value, -1 * base_shift) if is_day else shift_week_value(time_value, -1 * base_shift)
 
+    use_server_side_compute = not any((not is_day, not is_also_day)) and JIT_COMPUTE and not jit_bypass
+    if use_server_side_compute:
+        pad_length = get_pad_length(source_signal_pairs, transform_args.get("smoother_window_length"))
+        source_signal_pairs, row_transform_generator = get_basename_signals(source_signal_pairs)
+        time_window = pad_time_window(time_window, pad_length)
+
     # build query
     q = QueryBuilder("covidcast", "t")
 
@@ -262,8 +270,20 @@ def handle_trend():
 
     p = create_printer()
 
-    def gen(rows):
-        for key, group in groupby((parse_row(row, fields_string, fields_int, fields_float) for row in rows), lambda row: (row["geo_type"], row["geo_value"], row["source"], row["signal"])):
+    if use_server_side_compute:
+        def gen_transform(rows):
+            parsed_rows = (parse_row(row, fields_string, fields_int, fields_float) for row in rows)
+            transformed_rows = row_transform_generator(parsed_rows=parsed_rows, time_pairs=[TimePair("day", [time_window])], transform_args=transform_args)
+            for row in transformed_rows:
+                yield row
+    else:
+        def gen_transform(rows):
+            parsed_rows = (parse_row(row, fields_string, fields_int, fields_float) for row in rows)
+            for row in parsed_rows:
+                yield row
+
+    def gen_trend(rows):
+        for key, group in groupby(rows, lambda row: (row["geo_type"], row["geo_value"], row["source"], row["signal"])):
             geo_type, geo_value, source, signal = key
             if alias_mapper:
                 source = alias_mapper(source, signal)
@@ -277,7 +297,7 @@ def gen(rows):
         raise DatabaseErrorException(str(e))
 
     # now use a generator for sending the rows and execute all the other queries
-    return p(filter_fields(gen(r)))
+    return p(filter_fields(gen_trend(gen_transform(r))))
 
 
 @bp.route("/trendseries", methods=("GET", "POST"))

src/server/endpoints/covidcast_utils/trend.py

@@ -42,6 +42,8 @@ def compute_trend(geo_type: str, geo_value: str, signal_source: str, signal_sign
 
     # find all needed rows
     for time, value in rows:
+        if value is None:
+            continue
         if time == current_time:
             t.value = value
         if time == basis_time: