signal naming using wip_signal parameter

Author: vishakha1812

cdc_covidnet/delphi_cdc_covidnet/update_sensor.py

@@ -12,10 +12,17 @@
 import numpy as np
 import pandas as pd
 
+from delphi_utils import read_params
+from delphi_epidata import Epidata
+
 from .api_config import APIConfig
 from .covidnet import CovidNet
 from .geo_maps import GeoMaps
 
+
+SIGNALS = ["covidnet"]
+
+
 def write_to_csv(data: pd.DataFrame, out_name: str, output_path: str):
     """
     Write sensor values to csv.
@@ -72,14 +79,12 @@ def update_sensor(
                             right_on=["year", "weeknumber"])
 
     # Select relevant columns and standardize naming
-    hosp_df = hosp_df.loc[:, APIConfig.HOSP_RENAME_COLS.keys()]\
+    hosp_df = hosp_df.loc[:, APIConfig.HOSP_RENAME_COLS.keys()] \
         .rename(columns=APIConfig.HOSP_RENAME_COLS)
 
     # Restrict to start and end date
     hosp_df = hosp_df[
-        (hosp_df["date"] >= start_date) & (
-            hosp_df["date"] < end_date)
-    ]
+        (hosp_df["date"] >= start_date) & (hosp_df["date"] < end_date)]
 
     # Set state id to two-letter abbreviation
     geo_map = GeoMaps(static_path)
@@ -93,7 +98,64 @@ def update_sensor(
     hosp_df["sample_size"] = np.nan
 
     # Write results
-    out_name = "wip_covidnet"
-    write_to_csv(hosp_df, out_name, output_path)
+    signals = add_prefix(SIGNALS, wip_signal=read_params()["wip_signal"], prefix="wip_")
+    for signal in signals:
+        write_to_csv(hosp_df, signal, output_path)
 
     return hosp_df
+
+
+def add_prefix(signal_names, wip_signal, prefix: str):
+    """Adds prefix to signal if there is a WIP signal
+    Parameters
+    ----------
+    signal_names: List[str]
+        Names of signals to be exported
+    prefix : 'wip_'
+        prefix for new/non public signals
+    wip_signal : List[str] or bool
+        Either takes a list of wip signals: [], OR
+        incorporated all signals in the registry: True OR
+        no signals: False
+    Returns
+    -------
+    List of signal names
+        wip/non wip signals for further computation
+    """
+
+    if wip_signal in ("", False):
+        return signal_names
+    elif wip_signal and isinstance(wip_signal, bool):
+        return [
+            (prefix + signal) if public_signal(signal)
+            else signal
+            for signal in signal_names
+        ]
+    elif isinstance(wip_signal, list):
+        for signal in wip_signal:
+            if public_signal(signal):
+                signal_names.append(prefix + signal)
+                signal_names.remove(signal)
+        return signal_names
+    else:
+        raise ValueError("Supply True | False or '' or [] | list()")
+
+
+def public_signal(signal_):
+    """Checks if the signal name is already public using Epidata
+    Parameters
+    ----------
+    signal_ : str
+        Name of the signal
+    Returns
+    -------
+    bool
+        True if the signal is not present
+        False if the signal is present
+    """
+    epidata_df = Epidata.covidcast_meta()
+    for index in range(len(epidata_df['epidata'])):
+        if 'signal' in epidata_df['epidata'][index]:
+            if epidata_df['epidata'][index]['signal'] == signal_:
+                return False
+    return True

cdc_covidnet/params.json.template

@@ -4,5 +4,6 @@
   "cache_dir": "./cache",
   "start_date": "2020-03-07",
   "end_date": "",
-  "parallel": false
+  "parallel": false,
+   "wip_signal": ""
 }