migrate most indicators to common add_prefix

Author: sgsmob

_delphi_utils_python/delphi_utils/signal.py

@@ -1,13 +1,13 @@
 """Functions for understanding and creating signal names."""
 import covidcast
 
-def add_prefix(signal_names, wip_signal, prefix: str):
+def add_prefix(signal_names, wip_signal, prefix="wip_"):
     """Adds prefix to signal if there is a WIP signal
     Parameters
     ----------
     signal_names: List[str]
         Names of signals to be exported
-    prefix : 'wip_'
+    prefix : "wip_"
         prefix for new/non public signals
     wip_signal : List[str] or bool
         a list of wip signals: [], OR
@@ -50,6 +50,6 @@ def public_signal(signal):
     """
     epidata_df = covidcast.metadata()
     for index in range(len(epidata_df)):
-        if epidata_df['signal'][index] == signal:
+        if epidata_df["signal"][index] == signal:
             return True
     return False

_template_python/delphi_NAME/handle_wip_signal.py

@@ -12,7 +12,7 @@
 import numpy as np
 import pandas as pd
 
-from delphi_utils import read_params
+from delphi_utils import read_params, add_prefix
 import covidcast
 from .api_config import APIConfig
 from .covidnet import CovidNet
@@ -101,57 +101,3 @@ def update_sensor(
         write_to_csv(hosp_df, signal, output_path)
 
     return hosp_df
-
-
-def add_prefix(signal_names, wip_signal, prefix):
-    """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
-        a list of wip signals: [], OR
-        all signals in the registry: True OR
-        only signals that have never been published: False
-    Returns
-    -------
-    List of signal names
-        wip/non wip signals for further computation
-    """
-
-    if wip_signal is True:
-        return [prefix + signal for signal in signal_names]
-    if isinstance(wip_signal, list):
-        make_wip = set(wip_signal)
-        return [
-            (prefix if signal in make_wip else "") + signal
-            for signal in signal_names
-        ]
-    if wip_signal in {False, ""}:
-        return [
-            signal if public_signal(signal)
-            else prefix + signal
-            for signal in signal_names
-        ]
-    raise ValueError("Supply True | False or '' or [] | list()")
-
-
-def public_signal(signal_):
-    """Checks if the signal name is already public using COVIDcast
-    Parameters
-    ----------
-    signal_ : str
-        Name of the signal
-    Returns
-    -------
-    bool
-        True if the signal is present
-        False if the signal is not present
-    """
-    epidata_df = covidcast.metadata()
-    for index in range(len(epidata_df)):
-        if epidata_df['signal'][index] == signal_:
-            return True
-    return False

_template_python/tests/test_handle_wip_signal.py

@@ -13,9 +13,8 @@
 import covidcast
 import pandas as pd
 
-from delphi_utils import read_params
+from delphi_utils import read_params, add_prefix
 from .constants import METRICS, SMOOTH_TYPES, SENSORS, GEO_RESOLUTIONS
-from .handle_wip_signal import add_prefix
 
 
 def check_none_data_frame(data_frame, label, date_range):

cdc_covidnet/delphi_cdc_covidnet/update_sensor.py

@@ -5,7 +5,6 @@
 import pandas as pd
 
 from delphi_combo_cases_and_deaths.run import extend_raw_date_range, sensor_signal, combine_usafacts_and_jhu, COLUMN_MAPPING
-from delphi_combo_cases_and_deaths.handle_wip_signal import add_prefix
 from delphi_utils import read_params
 from delphi_combo_cases_and_deaths.constants import METRICS, SMOOTH_TYPES, SENSORS, GEO_RESOLUTIONS
 
@@ -36,28 +35,6 @@ def test_issue_dates():
 """
 
 
-def test_handle_wip_signal():
-    """Verify that "wip_" prefixes are being applied appropriately.
-    """
-    signal_list = [sensor_signal(metric, sensor, smoother)[1]
-                   for (metric, sensor, smoother) in
-                   product(METRICS, SENSORS, SMOOTH_TYPES)]
-
-    # Test wip_signal = True (all signals should receive prefix)
-    signal_names = add_prefix(signal_list, True, prefix="wip_")
-    assert all(s.startswith("wip_") for s in signal_names)
-
-    # Test wip_signal = list (only listed signals should receive prefix)
-    signal_names = add_prefix(signal_list, [signal_list[0]], prefix="wip_")
-    print(signal_names)
-    assert signal_names[0].startswith("wip_")
-    assert all(not s.startswith("wip_") for s in signal_names[1:])
-
-    # Test wip_signal = False (only unpublished signals should receive prefix)
-    signal_names = add_prefix(["xyzzy", signal_list[0]], False, prefix="wip_")
-    assert signal_names[0].startswith("wip_")
-    assert all(not s.startswith("wip_") for s in signal_names[1:])
-
 def test_unstable_sources():
     """Verify that combine_usafacts_and_jhu assembles the combined data
     frame correctly for all cases where 0, 1, or both signals are

cdc_covidnet/tests/test_handle_wip_signal.py

@@ -8,7 +8,7 @@
 from datetime import timedelta
 from multiprocessing import Pool, cpu_count
 import covidcast
-from delphi_utils import GeoMapper, S3ArchiveDiffer, read_params
+from delphi_utils import GeoMapper, S3ArchiveDiffer, read_params, add_prefix
 
 # third party
 import numpy as np
@@ -70,59 +70,6 @@ def write_to_csv(output_dict, write_se, out_name, output_path="."):
     logging.debug(f"wrote {out_n} rows for {len(geo_ids)} {geo_level}")
 
 
-def add_prefix(signal_names, wip_signal, prefix="wip_"):
-    """Adds prefix to signal if there is a WIP signal
-    Parameters
-    ----------
-    signal_names: List[str]
-        Names of signals to be exported
-    wip_signal : List[str] or bool
-        a list of wip signals: [], OR
-        all signals in the registry: True OR
-        only signals that have never been published: False
-    prefix : 'wip_'
-        prefix for new/non public signals
-    Returns
-    -------
-    List of signal names
-        wip/non wip signals for further computation
-    """
-    if wip_signal is True:
-        return [prefix + signal for signal in signal_names]
-    if isinstance(wip_signal, list):
-        make_wip = set(wip_signal)
-        return [
-            prefix + signal if signal in make_wip else signal
-            for signal in signal_names
-        ]
-    if wip_signal in {False, ""}:
-        return [
-            signal if public_signal(signal)
-            else prefix + signal
-            for signal in signal_names
-        ]
-    raise ValueError("Supply True | False or '' or [] | list()")
-
-
-def public_signal(signal_):
-    """Checks if the signal name is already public using COVIDcast
-    Parameters
-    ----------
-    signal_ : str
-        Name of the signal
-    Returns
-    -------
-    bool
-        True if the signal is present
-        False if the signal is not present
-    """
-    epidata_df = covidcast.metadata()
-    for index in range(len(epidata_df)):
-        if epidata_df['signal'][index] == signal_:
-            return True
-    return False
-
-
 class EMRHospSensorUpdator:
 
     def __init__(self,