Merge pull request #368 from sgsmob/signals

Migrate indicators to a common `add_prefix()` library

Author: krivard

_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"

quidel/delphi_quidel/handle_wip_sensor.py renamed to _delphi_utils_python/delphi_utils/signal.py

@@ -1,13 +1,13 @@
-"""This file checks the wip status of signals"""
+"""Functions for understanding and creating signal names."""
 import covidcast
 
-def add_prefix(signal_names, wip_signal, prefix):
+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
@@ -36,11 +36,11 @@ def add_prefix(signal_names, wip_signal, prefix):
     raise ValueError("Supply True | False or '' or [] | list()")
 
 
-def public_signal(signal_):
+def public_signal(signal):
     """Checks if the signal name is already public using COVIDcast
     Parameters
     ----------
-    signal_ : str
+    signal : str
         Name of the signal
     Returns
     -------
@@ -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

_delphi_utils_python/setup.py

@@ -3,6 +3,7 @@
 
 required = [
     "boto3",
+    "covidcast",
     "gitpython",
     "moto",
     "numpy",

_delphi_utils_python/tests/test_signal.py

@@ -0,0 +1,34 @@
+"""Tests for delphi_utils.signal."""
+from unittest.mock import patch
+import pandas as pd
+
+from delphi_utils.signal import add_prefix, public_signal
+
+# Constants for mocking out the call to `covidcast.metadata` within `public_signal()`.
+PUBLIC_SIGNALS = ["sig1", "sig2", "sig3"]
+PUBLIC_SIGNALS_FRAME = pd.DataFrame(data={"signal": PUBLIC_SIGNALS})
+
+class TestSignal:
+    """Tests for signal.py."""
+
+    def test_add_prefix_to_all(self):
+        """Tests that `add_prefix()` derives work-in-progress names for all input signals."""
+        assert add_prefix(["sig1", "sig3"], True, prefix="wip_") == ["wip_sig1", "wip_sig3"]
+    
+    def test_add_prefix_to_specified(self):
+        """Tests that `add_prefix()` derives work-in-progress names for specified signals."""
+        assert add_prefix(["sig1", "sig2", "sig3"], ["sig2"], prefix="wip_") ==\
+            ["sig1", "wip_sig2", "sig3"]
+    
+    @patch("covidcast.metadata")
+    def test_add_prefix_to_non_public(self, metadata):
+        """Tests that `add_prefix()` derives work-in-progress names for non-public signals."""
+        metadata.return_value = PUBLIC_SIGNALS_FRAME
+        assert add_prefix(["sig0", "sig1"], False, prefix="wip_") == ["wip_sig0", "sig1"]
+
+    @patch("covidcast.metadata")
+    def test_public_signal(self, metadata):
+        """Tests that `public_signal()` identifies public vs. private signals."""
+        metadata.return_value = PUBLIC_SIGNALS_FRAME
+        assert not public_signal("sig0")
+        assert public_signal("sig2")

_template_python/delphi_NAME/handle_wip_signal.py

@@ -12,8 +12,7 @@
 import numpy as np
 import pandas as pd
 
-from delphi_utils import read_params, GeoMapper
-import covidcast
+from delphi_utils import read_params, GeoMapper, add_prefix
 from .api_config import APIConfig
 from .covidnet import CovidNet
 from .constants import SIGNALS
@@ -106,57 +105,3 @@ def update_sensor(
     for signal in signals:
         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

@@ -95,8 +95,7 @@ def run_module():
             su_inst.update_sensor(
                 params["input_denom_file"],
                 params["input_covid_file"],
-                params["export_dir"],
-                params["static_file_dir"]
+                params["export_dir"]
             )
         logging.info("finished %s", geo)