add signal module to delphi_utils

Author: sgsmob

_delphi_utils_python/delphi_utils/__init__.py

@@ -8,5 +8,6 @@
 from .export import create_export_csv
 from .utils import read_params
 from .geomap import GeoMapper
+from .signal import add_prefix, public_signal
 
 __version__ = "0.1.0"

_delphi_utils_python/delphi_utils/signal.py

@@ -0,0 +1,55 @@
+"""Functions for understanding and creating signal names."""
+import covidcast
+
+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
+        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

_delphi_utils_python/tests/test_signal.py

@@ -0,0 +1,27 @@
+"""Tests for delphi_utils.signal."""
+
+from delphi_utils.signal import add_prefix, public_signal
+
+SIGNALS = ["median_home_dwell_time", "completely_home_prop", "full_time_work_prop"]
+
+class TestSignal:
+    """Tests for signal.py."""
+    def test_handle_wip_signal(self):
+        """Tests that `add_prefix()` derives work-in-progress signals."""
+        # Test wip_signal = True
+        signal_names = SIGNALS
+        signal_names = add_prefix(SIGNALS, True, prefix="wip_")
+        assert all(s.startswith("wip_") for s in signal_names)
+        # Test wip_signal = list
+        signal_names = add_prefix(SIGNALS, [SIGNALS[0]], prefix="wip_")
+        assert signal_names[0].startswith("wip_")
+        assert all(not s.startswith("wip_") for s in signal_names[1:])
+        # Test wip_signal = False
+        signal_names = add_prefix(["xyzzy", SIGNALS[0]], False, prefix="wip_")
+        assert signal_names[0].startswith("wip_")
+        assert all(not s.startswith("wip_") for s in signal_names[1:])
+
+    def test_public_signal(self):
+        """Tests that `public_signal()` identifies public vs. private signals."""
+        assert not public_signal("junk")
+        assert public_signal("covid_ag_smoothed_pct_positive")