Incorporate George's review feedback

* make _combine_source_signal_pairs logic clearer
* reorder covidcast.py endpoints under a single use_server_side_compute branch
* rename _resolve_all_signals to _resolve_bool_source_signals
* remove ugly one-liner from pad_time_pairs and replace with clearer code
* add a few documentation updates
* a few efficiency updates in smooth_diff

Author: dshemetov

src/server/_params.py

@@ -1,10 +1,11 @@
 from dataclasses import dataclass
-from itertools import groupby, chain
+from itertools import groupby
 from math import inf
 import re
 from typing import List, Optional, Sequence, Tuple, Union
 
 from flask import request
+from more_itertools import flatten
 
 from ._exceptions import ValidationFailedException
 from .utils import days_in_range, weeks_in_range, guess_time_value_is_day
@@ -105,9 +106,9 @@ def _combine_source_signal_pairs(source_signal_pairs: List[SourceSignalPair]) ->
     for source, group in source_signal_pairs_grouped:
         group = list(group)
         if any(x.signal == True for x in group):
-            source_signal_pairs_combined.append(SourceSignalPair(source, True))
-            continue
-        combined_signals = sorted(list(set(chain(*[x.signal for x in group]))))
+            combined_signals = True
+        else:
+            combined_signals = sorted(set(flatten(x.signal for x in group)))
         source_signal_pairs_combined.append(SourceSignalPair(source, combined_signals))
     return source_signal_pairs_combined
 

src/server/endpoints/covidcast.py

@@ -179,37 +179,8 @@ def handle():
     lag = extract_integer("lag")
     is_time_type_week = any(time_pair.time_type == "week" for time_pair in time_pairs)
     is_time_value_true = any(isinstance(time_pair.time_values, bool) for time_pair in time_pairs)
-    use_server_side_compute = not any((issues, lag, is_time_type_week, is_time_value_true)) and JIT_COMPUTE and not jit_bypass
-    if use_server_side_compute:
-        transform_args = parse_transform_args()
-        pad_length = get_pad_length(source_signal_pairs, transform_args.get("smoother_window_length"))
-        time_pairs = pad_time_pairs(time_pairs, pad_length)
-        source_signal_pairs, row_transform_generator = get_basename_signal_and_jit_generator(source_signal_pairs, transform_args=transform_args)
 
-    # build query
-    q = QueryBuilder(latest_table, "t")
-
-    fields_string = ["geo_type", "geo_value", "source", "signal", "time_type"]
-    fields_int = ["time_value", "direction", "issue", "lag", "missing_value", "missing_stderr", "missing_sample_size"]
-    fields_float = ["value", "stderr", "sample_size"]
     is_compatibility = is_compatibility_mode()
-
-    q.set_order("geo_type", "geo_value", "source", "signal", "time_type", "time_value", "issue")
-    q.set_fields(fields_string, fields_int, fields_float)
-
-    # basic query info
-    # data type of each field
-    # build the source, signal, time, and location (type and id) filters
-    q.where_source_signal_pairs("source", "signal", source_signal_pairs)
-    q.where_geo_pairs("geo_type", "geo_value", geo_pairs)
-    q.where_time_pairs("time_type", "time_value", time_pairs)
-
-    q.index = guess_index_to_use(time_pairs, geo_pairs, issues, lag, as_of)
-
-    _handle_lag_issues_as_of(q, issues, lag, as_of)
-
-    p = create_printer()
-
     def alias_row(row):
         if is_compatibility:
             # old api returned fewer fields
@@ -222,7 +193,20 @@ def alias_row(row):
         row["source"] = alias_mapper(row["source"], row["signal"])
         return row
 
+    # build query
+    q = QueryBuilder(latest_table, "t")
+
+    fields_string = ["geo_type", "geo_value", "source", "signal", "time_type"]
+    fields_int = ["time_value", "direction", "issue", "lag", "missing_value", "missing_stderr", "missing_sample_size"]
+    fields_float = ["value", "stderr", "sample_size"]
+
+    use_server_side_compute = not any((issues, lag, is_time_type_week, is_time_value_true)) and JIT_COMPUTE and not jit_bypass
     if use_server_side_compute:
+        transform_args = parse_transform_args()
+        pad_length = get_pad_length(source_signal_pairs, transform_args.get("smoother_window_length"))
+        time_pairs = pad_time_pairs(time_pairs, pad_length)
+        source_signal_pairs, row_transform_generator = get_basename_signal_and_jit_generator(source_signal_pairs, transform_args=transform_args)
+
         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=time_pairs, transform_args=transform_args)
@@ -234,6 +218,22 @@ def gen_transform(rows):
             for row in parsed_rows:
                 yield alias_row(row)
 
+    q.set_order("geo_type", "geo_value", "source", "signal", "time_type", "time_value", "issue")
+    q.set_fields(fields_string, fields_int, fields_float)
+
+    # basic query info
+    # data type of each field
+    # build the source, signal, time, and location (type and id) filters
+    q.where_source_signal_pairs("source", "signal", source_signal_pairs)
+    q.where_geo_pairs("geo_type", "geo_value", geo_pairs)
+    q.where_time_pairs("time_type", "time_value", time_pairs)
+
+    q.index = guess_index_to_use(time_pairs, geo_pairs, issues, lag, as_of)
+
+    _handle_lag_issues_as_of(q, issues, lag, as_of)
+
+    p = create_printer()
+
     # execute first query
     try:
         r = run_query(p, (str(q), q.params))
@@ -263,41 +263,40 @@ def handle_trend():
 
     time_window, is_day = parse_day_or_week_range_arg("window")
     time_value, is_also_day = parse_day_or_week_arg("date")
+
     if is_day != is_also_day:
         raise ValidationFailedException("mixing weeks with day arguments")
+
     _verify_argument_time_type_matches(is_day, daily_signals, weekly_signals)
+
     basis_time_value = extract_date("basis")
     if basis_time_value is None:
         base_shift = extract_integer("basis_shift")
         if base_shift is None:
             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_signal_and_jit_generator(source_signal_pairs)
-        time_window = pad_time_window(time_window, pad_length)
+    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)
+            trend = compute_trend(geo_type, geo_value, source, signal, time_value, basis_time_value, ((row["time_value"], row["value"]) for row in group))
+            yield trend.asdict()
 
     # build query
     q = QueryBuilder(latest_table, "t")
 
     fields_string = ["geo_type", "geo_value", "source", "signal"]
     fields_int = ["time_value"]
     fields_float = ["value"]
-    q.set_fields(fields_string, fields_int, fields_float)
-    q.set_order("geo_type", "geo_value", "source", "signal", "time_value")
-
-    q.where_source_signal_pairs("source", "signal", source_signal_pairs)
-    q.where_geo_pairs("geo_type", "geo_value", geo_pairs)
-    q.where_time_pairs("time_type", "time_value", [TimePair("day" if is_day else "week", [time_window])])
-
-    # fetch most recent issue fast
-    _handle_lag_issues_as_of(q, None, None, None)
-
-    p = create_printer()
 
+    use_server_side_compute = all((is_day, 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_signal_and_jit_generator(source_signal_pairs)
+        time_window = pad_time_window(time_window, pad_length)
+
         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)
@@ -309,13 +308,18 @@ def gen_transform(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)
-            trend = compute_trend(geo_type, geo_value, source, signal, time_value, basis_time_value, ((row["time_value"], row["value"]) for row in group))
-            yield trend.asdict()
+    q.set_fields(fields_string, fields_int, fields_float)
+    q.set_order("geo_type", "geo_value", "source", "signal", "time_value")
+
+    q.where_source_signal_pairs("source", "signal", source_signal_pairs)
+    q.where_geo_pairs("geo_type", "geo_value", geo_pairs)
+    q.where_time_pairs("time_type", "time_value", [TimePair("day" if is_day else "week", [time_window])])
+
+    # fetch most recent issue fast
+    _handle_lag_issues_as_of(q, None, None, None)
+
+    p = create_printer()
+
 
     # execute first query
     try:
@@ -338,40 +342,39 @@ def handle_trendseries():
     jit_bypass = parse_jit_bypass()
 
     time_window, is_day = parse_day_or_week_range_arg("window")
+
     _verify_argument_time_type_matches(is_day, daily_signals, weekly_signals)
+
     basis_shift = extract_integer(("basis", "basis_shift"))
     if basis_shift is None:
         basis_shift = 7
 
-    use_server_side_compute = is_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_signal_and_jit_generator(source_signal_pairs)
-        time_window = pad_time_window(time_window, pad_length)
+    shifter = lambda x: shift_time_value(x, -basis_shift)
+    if not is_day:
+        shifter = lambda x: shift_week_value(x, -basis_shift)
+
+    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)
+            trends = compute_trends(geo_type, geo_value, source, signal, shifter, ((row["time_value"], row["value"]) for row in group))
+            for t in trends:
+                yield t.asdict()
 
     # build query
     q = QueryBuilder(latest_table, "t")
 
     fields_string = ["geo_type", "geo_value", "source", "signal"]
     fields_int = ["time_value"]
     fields_float = ["value"]
-    q.set_fields(fields_string, fields_int, fields_float)
-    q.set_order("geo_type", "geo_value", "source", "signal", "time_value")
-
-    q.where_source_signal_pairs("source", "signal", source_signal_pairs)
-    q.where_geo_pairs("geo_type", "geo_value", geo_pairs)
-    q.where_time_pairs("time_type", "time_value", [TimePair("day" if is_day else "week", [time_window])])
-
-    # fetch most recent issue fast
-    _handle_lag_issues_as_of(q, None, None, None)
-
-    p = create_printer()
-
-    shifter = lambda x: shift_time_value(x, -basis_shift)
-    if not is_day:
-        shifter = lambda x: shift_week_value(x, -basis_shift)
 
+    use_server_side_compute = is_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_signal_and_jit_generator(source_signal_pairs)
+        time_window = pad_time_window(time_window, pad_length)
+
         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)
@@ -383,14 +386,17 @@ def gen_transform(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)
-            trends = compute_trends(geo_type, geo_value, source, signal, shifter, ((row["time_value"], row["value"]) for row in group))
-            for t in trends:
-                yield t.asdict()
+    q.set_fields(fields_string, fields_int, fields_float)
+    q.set_order("geo_type", "geo_value", "source", "signal", "time_value")
+
+    q.where_source_signal_pairs("source", "signal", source_signal_pairs)
+    q.where_geo_pairs("geo_type", "geo_value", geo_pairs)
+    q.where_time_pairs("time_type", "time_value", [TimePair("day" if is_day else "week", [time_window])])
+
+    # fetch most recent issue fast
+    _handle_lag_issues_as_of(q, None, None, None)
+
+    p = create_printer()
 
     # execute first query
     try:
@@ -483,9 +489,12 @@ def handle_export():
     start_day, is_day = parse_day_or_week_arg("start_day", 202001 if weekly_signals > 0 else 20200401)
     end_day, is_end_day = parse_day_or_week_arg("end_day", 202020 if weekly_signals > 0 else 20200901)
     time_window = (start_day, end_day)
+
     if is_day != is_end_day:
         raise ValidationFailedException("mixing weeks with day arguments")
+
     _verify_argument_time_type_matches(is_day, daily_signals, weekly_signals)
+
     transform_args = parse_transform_args()
     jit_bypass = parse_jit_bypass()
 
@@ -499,19 +508,30 @@ def handle_export():
     if is_day != is_as_of_day:
         raise ValidationFailedException("mixing weeks with day arguments")
 
+    # build query
+    q = QueryBuilder(latest_table, "t")
+
+    fields_string = ["geo_value", "signal", "geo_type", "source"]
+    fields_int = ["time_value", "issue", "lag"]
+    fields_float = ["value", "stderr", "sample_size"]
+
     use_server_side_compute = all([is_day, is_end_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_signal_and_jit_generator(source_signal_pairs)
         time_window = pad_time_window(time_window, pad_length)
 
-    # build query
-    q = QueryBuilder(latest_table, "t")
+        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):
+            for row in rows:
+                yield row
 
-    fields_string = ["geo_value", "signal", "geo_type", "source"]
-    fields_int = ["time_value", "issue", "lag"]
-    fields_float = ["value", "stderr", "sample_size"]
-    q.set_fields(fields_string + fields_int + fields_float, [], [])
+    q.set_fields(fields_string, fields_int, fields_float)
     q.set_order("time_value", "geo_value")
     q.where_source_signal_pairs("source", "signal", source_signal_pairs)
     q.where_time_pairs("time_type", "time_value", [TimePair("day" if is_day else "week", [time_window])])
@@ -541,17 +561,6 @@ def parse_csv_row(i, row):
             "data_source": alias_mapper(row["source"], row["signal"]) if alias_mapper else row["source"],
         }
 
-    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):
-            for row in rows:
-                yield row
-
     def gen_parse(rows):
         for i, row in enumerate(rows):
             yield parse_csv_row(i, row)