move automated backfill to run.py

Author: nmdefries

dsew_community_profile/delphi_dsew_community_profile/pull.py

@@ -5,7 +5,6 @@
 import os
 import re
 from urllib.parse import quote_plus as quote_as_url
-import covidcast
 
 import pandas as pd
 import requests
@@ -361,23 +360,6 @@ def nation_from_state(df, sig, geomapper):
 
 def fetch_new_reports(params, logger=None):
     """Retrieve, compute, and collate all data we haven't seen yet."""
-    # Fetch metadata to check how recent each signal is
-    metadata = covidcast.metadata()
-    sensor_names = {
-        SIGNALS[key][name_field]
-        for key in params["indicator"]["export_signals"]
-        for name_field in ["api_name", "api_prop_name"]
-        if name_field in SIGNALS[key]
-    }
-
-    # Filter to only those we currently want to produce, ignore any old or deprecated signals
-    cpr_metadata = metadata[(metadata.data_source == "dsew-cpr") &
-        (metadata.signal.isin(sensor_names))]
-
-    if sensor_names.difference(set(cpr_metadata.signal)):
-        # If any signal not in metadata yet, we need to backfill its full history.
-        params['indicator']['reports'] = 'all'
-
     listing = fetch_listing(params)
 
     # download and parse individual reports

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)