Merge pull request #1527 from cmu-delphi/ndefries/cpr-generate-full-history-new-signals

Generate full CPR history for new signals

Author: krivard

dsew_community_profile/delphi_dsew_community_profile/pull.py

@@ -312,12 +312,12 @@ def fetch_listing(params):
             el for el in listing
             if start_date <= el['publish_date'] <= end_date
         ]
-    # reference date is guaranteed to be before publish date, so we can trim
+    # reference date is guaranteed to be on or before publish date, so we can trim
     # reports that are too early
     if 'export_start_date' in params['indicator']:
         listing = [
             el for el in listing
-            if params['indicator']['export_start_date'] < el['publish_date']
+            if params['indicator']['export_start_date'] <= el['publish_date']
         ]
     # can't do the same for export_end_date
     return listing

dsew_community_profile/delphi_dsew_community_profile/run.py

@@ -19,8 +19,9 @@
 from delphi_utils import get_structured_logger
 from delphi_utils.export import create_export_csv
 import pandas as pd
+import covidcast
 
-from .constants import make_signal_name
+from .constants import make_signal_name, SIGNALS
 from .pull import fetch_new_reports
 
 
@@ -71,6 +72,48 @@ def replace_date_param(p):
         if len(dates)>0:
             run_stats.append((max(dates), len(dates)))
 
+    ## If any requested signal is not in metadata, generate it for all dates.
+    #
+    # Only do so if params.reports is set to "new". If set to "all", the
+    # previous fetch_new_reports + CSV loop will already have generated the full
+    # history for new signals. If params.reports is set to a specific date
+    # range, that request overrides automated backfill.
+    if params['indicator']['reports'] == 'new':
+        # Fetch metadata to check how recent signals are
+        metadata = covidcast.metadata()
+        sensor_names = {
+            SIGNALS[key][name_field]: key
+            for key in params["indicator"]["export_signals"]
+            for name_field in ["api_name", "api_prop_name"]
+            if name_field in SIGNALS[key].keys()
+        }
+
+        # Filter to only those we currently want to produce
+        cpr_metadata = metadata[(metadata.data_source == "dsew-cpr") &
+            (metadata.signal.isin(sensor_names.keys()))]
+
+        new_signals = set(sensor_names.keys()).difference(set(cpr_metadata.signal))
+        if new_signals:
+            # If any signal not in metadata yet, we need to backfill its full
+            # history.
+            params['indicator']['reports'] = 'all'
+            params['indicator']['export_signals'] = {sensor_names[key] for key in new_signals}
+
+        dfs = fetch_new_reports(params, logger)
+        for key, df in dfs.items():
+            (geo, sig, is_prop) = key
+            if sig not in params["indicator"]["export_signals"]:
+                continue
+            dates = create_export_csv(
+                df,
+                params['common']['export_dir'],
+                geo,
+                make_signal_name(sig, is_prop),
+                **export_params
+            )
+            if len(dates)>0:
+                run_stats.append((max(dates), len(dates)))
+
     ## log this indicator run
     elapsed_time_in_seconds = round(time.time() - start_time, 2)
     min_max_date = run_stats and min(s[0] for s in run_stats)