Server-side compute: add cumulative->incidence

Author: dshemetov

integrations/server/test_covidcast_endpoints.py

@@ -191,6 +191,75 @@ def test_basic(self):
             expected_values = [float(row.value) for row in rows]
             self.assertEqual(out_values, expected_values)
 
+    def test_derived_signals(self):
+        time_value_pairs = [(20200401 + i, i ** 2) for i in range(10)]
+        rows01 = [CovidcastRow(source="jhu-csse", signal="confirmed_cumulative_num", time_value=time_value, value=value, geo_value="01") for time_value, value in time_value_pairs]
+        rows02 = [CovidcastRow(source="jhu-csse", signal="confirmed_cumulative_num", time_value=time_value, value=2 * value, geo_value="02") for time_value, value in time_value_pairs]
+        first = rows01[0]
+        self._insert_rows(rows01 + rows02)
+
+        with self.subTest("diffed signal"):
+            out = self._fetch("/", signal="jhu-csse:confirmed_incidence_num", geo=first.geo_pair, time="day:*")
+            assert out['result'] == -2
+            out = self._fetch("/", signal="jhu-csse:confirmed_incidence_num", geo=first.geo_pair, time="day:20200401-20200410")
+            self.assertEqual(len(out["epidata"]), len(rows01))
+            out_values = [row["value"] for row in out["epidata"]]
+            values = [None] + [value for _, value in time_value_pairs]
+            expected_values = self._diff_rows(values)
+            self.assertAlmostEqual(out_values, expected_values)
+
+        with self.subTest("diffed signal, multiple geos"):
+            out = self._fetch("/", signal="jhu-csse:confirmed_incidence_num", geo="county:01,02", time="day:20200401-20200410")
+            self.assertEqual(len(out["epidata"]), 2*(len(rows01)))
+            out_values = [row["value"] for row in out["epidata"]]
+            values1 = [None] + [value for _, value in time_value_pairs]
+            values2 = [None] + [2 * value for _, value in time_value_pairs]
+            expected_values = self._diff_rows(values1) + self._diff_rows(values2)
+            self.assertAlmostEqual(out_values, expected_values)
+
+        with self.subTest("diffed signal, multiple geos using geo:*"):
+            out = self._fetch("/", signal="jhu-csse:confirmed_incidence_num", geo="county:*", time="day:20200401-20200410")
+            self.assertEqual(len(out["epidata"]), 2*(len(rows01)))
+            values1 = [None] + [value for _, value in time_value_pairs]
+            values2 = [None] + [2 * value for _, value in time_value_pairs]
+            expected_values = self._diff_rows(values1) + self._diff_rows(values2)
+            self.assertAlmostEqual(out_values, expected_values)
+
+
+        with self.subTest("diffing with time window resizing"):
+            # should fetch 1 extra day
+            out = self._fetch("/", signal="jhu-csse:confirmed_incidence_num", geo=first.geo_pair, time="day:20200402-20200410")
+            self.assertEqual(len(out["epidata"]), len(rows01) - 1)
+            out_values = [row["value"] for row in out["epidata"]]
+            values = [value for _, value in time_value_pairs]
+            expected_values = self._diff_rows(values)
+            self.assertAlmostEqual(out_values, expected_values)
+
+        time_value_pairs = [(20200401 + i, i ** 2) for i in chain(range(10), range(15, 20))]
+        rows = [CovidcastRow(source="jhu-csse", signal="confirmed_cumulative_num", geo_value="03", time_value=time_value, value=value) for time_value, value in time_value_pairs]
+        first = rows[0]
+        self._insert_rows(rows)
+
+        with self.subTest("diffing with a time gap"):
+            # should fetch 1 extra day
+            out = self._fetch("/", signal="jhu-csse:confirmed_incidence_num", geo=first.geo_pair, time="day:20200401-20200420")
+            self.assertEqual(len(out["epidata"]), len(rows) + 5)
+            out_values = [row["value"] for row in out["epidata"]]
+            values = [None] + [value for _, value in time_value_pairs][:10] + [None] * 5 + [value for _, value in time_value_pairs][10:]
+            expected_values = self._diff_rows(values)
+            self.assertAlmostEqual(out_values, expected_values)
+
+
+    def test_compatibility(self):
+        """Request at the /api.php endpoint."""
+        rows = [CovidcastRow(source="src", signal="sig", time_value=20200401 + i, value=i) for i in range(10)]
+        first = rows[0]
+        self._insert_rows(rows)
+
+        with self.subTest("simple"):
+            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 test_trend(self):
         """Request a signal the /trend endpoint."""
 

src/server/endpoints/covidcast_utils/smooth_diff.py

@@ -72,4 +72,15 @@ def generate_row_diffs(rows: Iterable[Dict], nan_fill_value: Number = nan, **kwa
         Container for non-shared parameters with other computation functions.
     """
     for window in windowed(rows, 2):
-        yield diffed_row(window)
+        if all(e is None for e in window):
+            continue # should only occur if rows is empty
+        try:
+            first_value = window[0].get("value", nan_fill_value) if window[0] is not None else nan_fill_value
+            second_value = window[1].get("value", nan_fill_value) if window[1] is not None else nan_fill_value
+            value = round(second_value - first_value, 8)
+            value = float(value) if not isnan(value) else None
+        except TypeError:
+            value = None
+        item = list(filter(lambda e: e is not None, window))[-1].copy() # inherit values of last time stamp
+        item.update({"value": value, "stderr": None, "sample_size": None, "missing_value": Nans.NOT_MISSING if value is not None else Nans.NOT_APPLICABLE, "missing_stderr": Nans.NOT_APPLICABLE, "missing_sample_size": Nans.NOT_APPLICABLE})
+        yield item

tests/server/endpoints/covidcast_utils/test_smooth_diff.py

@@ -0,0 +1,48 @@
+from pandas import DataFrame, date_range
+from pandas.testing import assert_frame_equal
+from numpy import nan
+from itertools import chain
+from more_itertools import windowed
+
+from delphi.epidata.server.utils import date_to_time_value, CovidcastRecords
+from delphi.epidata.server.endpoints.covidcast_utils.smooth_diff import generate_row_diffs, generate_smooth_rows
+
+
+class TestStreaming:
+
+
+    def test_generate_row_diffs(self):
+        # an empty dataframe should return an empty dataframe
+        data = DataFrame({})
+        diffs_df = DataFrame.from_records(generate_row_diffs(data.to_dict(orient='records')))
+        expected_df = DataFrame({})
+        assert_frame_equal(diffs_df, expected_df)
+
+        # a dataframe with a single entry should return a single nan value
+        data = CovidcastRecords(
+            time_values=[20210501],
+            values=[1.0]
+        ).as_dataframe()
+        diffs_df = DataFrame.from_records(generate_row_diffs(data.to_dict(orient='records')))
+        expected_df = CovidcastRecords(
+            time_values=[20210501],
+            values=[None],
+            stderrs=[None],
+            sample_sizes=[None]
+        ).as_dataframe()
+        assert_frame_equal(diffs_df, expected_df)
+
+        data = CovidcastRecords(
+            time_values=date_range("2021-05-01", "2021-05-10"),
+            values=chain(range(7), [None, 2., 1.])
+        ).as_dataframe()
+
+        # no fill
+        diffs_df = DataFrame.from_records(generate_row_diffs(data.to_dict(orient='records')))
+        expected_df = CovidcastRecords(
+            time_values=date_range("2021-05-02", "2021-05-10"),
+            values=[1.] * 6 + [None, None, -1.],
+            stderrs=[None] * 9,
+            sample_sizes=[None] * 9
+        ).as_dataframe()
+        assert_frame_equal(diffs_df, expected_df)