Add a missing signal detector helper job

.github/workflows/missing_signals.yaml

@@ -0,0 +1,28 @@
+name: Missing Signal Detector
+
+on:
+  workflow_dispatch:
+
+jobs:
+  build:
+    runs-on: ubuntu-latest
+    steps:
+      - name: Check out code
+        uses: actions/checkout@v2
+        with:
+          ref: dev
+      - name: Set up Python 3.8
+        uses: actions/setup-python@v2
+        with:
+          python-version: 3.8
+      - name: Install Dependencies
+        run: pip install requests pandas
+      - name: Run Missing Signals Detector
+        run: python scripts/report_missing_covidcast_meta.py
+      - name: Upload Missing Artifact
+        if: failure()
+        uses: actions/upload-artifact@v2
+        with:
+          name: missing_db_signal.csv
+          path:  missing_db_signal.csv
+

.gitignore

@@ -7,3 +7,4 @@ __pycache__/
 /build
 /node_modules
 .mypy_cache
+/missing_db_signals.csv

scripts/report_missing_covidcast_meta.py

@@ -0,0 +1,87 @@
+from typing import Dict, List, Tuple, Union
+from requests import get
+import sys
+import pandas as pd
+from pathlib import Path
+
+base_dir = Path(__file__).parent.parent
+base_url = 'https://delphi.cmu.edu/epidata'
+
+def is_known_missing(source: str, signal: str) -> bool:
+    if '7dav_cumulative' in signal:
+        return True
+    if source in ('youtube-survey', 'indicator-combination'):
+        return True
+    return False
+
+def compute_missing_signals() -> List[Tuple[Tuple[str, str], Dict]]:
+    defined_meta = get(f"{base_url}/covidcast/meta").json()
+    defined_signals: Dict[Tuple[str, str], Dict] = {}
+    for source in defined_meta:
+        for signal in source['signals']:
+            defined_signals[(signal['source'], signal['signal'])] = signal
+            defined_signals[(source['db_source'], signal['signal'])] = signal
+
+    computed_meta = get(f"{base_url}/covidcast_meta/?format=json").json()
+    computed_signals: Dict[Tuple[str, str], List[Dict]] = {}
+    for entry in computed_meta:
+        computed_signals.setdefault((entry['data_source'], entry['signal']), []).append(entry)
+
+    missing_signals: List[Tuple[Tuple[str, str], Dict]]  = []
+
+    for key, infos in computed_signals.items():
+        defined_info = defined_signals.get(key)
+        if not defined_info:
+            if not is_known_missing(key[0], key[1]):
+                missing_signals.append((key, infos[0]))
+    return missing_signals
+
+
+def gen_row(source: str, signal: str, info: Dict) -> Dict:
+    is_weighted = signal.startswith('smoothed_w') and not (signal.startswith('smoothed_wa') or signal.startswith('smoothed_we') or signal.startswith('smoothed_wi') or signal.startswith('smoothed_wo') or signal.startswith('smoothed_wu'))
+    base_name = signal.replace('smoothed_w', 'smoothed_') if is_weighted else signal
+    bool_str = lambda x: 'TRUE' if x else 'FALSE'
+
+    return {
+        'Source Subdivision': source,
+        'Signal BaseName': base_name,
+        'base_is_other': bool_str(False),
+        'Signal': signal,
+        'Compute From Base': False,
+        'Name': "{base_name} (Weighted)" if is_weighted else signal,
+        'Active': bool_str(True),
+        'Short Description': 'TODO' if base_name == signal else '',
+        'Description': 'TODO' if base_name == signal else '',
+        'Time Type': info['time_type'],
+        'Time Label': 'Week' if info['time_type'] == 'week' else 'Day',
+        'Value Label': 'Percentage' if source == 'fb-survey' else 'Value',
+        'Format': 'percent' if source == 'fb-survey' else 'raw',
+        'Category': 'public' if source == 'fb-survey' else 'other',
+        'High Values Are': 'neutral',
+        'Is Smoothed': bool_str(signal.startswith('smoothed') or '7dav' in signal),
+        'Is Weighted': bool_str(is_weighted),
+        'Is Cumulative': bool_str('cumulative' in signal),
+        'Has StdErr': 'TRUE' if source == 'fb-survey' else '',
+        'Has Sample Size': 'TRUE' if source == 'fb-survey' else '',
+        'Link': 'TODO'
+    }
+
+def generate_missing_info_hint(missing_signals: List[Tuple[Tuple[str, str], Dict]]) -> None:
+    missing = pd.DataFrame.from_records([gen_row(s[0], s[1], info) for s, info in missing_signals])
+
+    # use the current as base to have the right column order
+    current = pd.read_csv(base_dir / 'src/server/endpoints/covidcast_utils/db_signals.csv')
+    # clear
+    current = current[0:0]
+    guessed: pd.DataFrame = pd.concat([current, missing])
+    guessed.to_csv(base_dir / 'missing_db_signals.csv', index=False)
+
+missing = compute_missing_signals()
+if missing:
+    print(f'found {len(missing)} missing signals')
+    generate_missing_info_hint(missing)
+    sys.exit(1)
+else:
+    print(f'all signals found')
+    sys.exit(0)
+