handling wip signal in python template

_template_python/delphi_NAME/constants.py

@@ -0,0 +1,17 @@
+"""
+Registry for signal names
+"""
+
+
+## example: 
+# FULL_TIME = "full_time_work_prop"
+# PART_TIME = "part_time_work_prop"
+# COVIDNET = "covidnet"
+# 
+# SIGNALS = [
+#     FULL_TIME,
+#     PART_TIME,
+#     COVIDNET
+# ]
+
+SIGNALS = []

_template_python/delphi_NAME/handle_wip_signal.py

@@ -0,0 +1,62 @@
+"""
+Handle signal names.
+
+Author: Vishakha
+Created: 2020-08-07
+"""
+
+import covidcast
+
+
+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 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 + 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

_template_python/delphi_NAME/run.py

@@ -4,11 +4,18 @@
 This module should contain a function called `run_module`, that is executed
 when the module is run with `python -m MODULE_NAME`.
 """
-import numpy as np
-import pandas as pd
 from delphi_utils import read_params
-
+from .handle_wip_signal import add_prefix
+from .constants import SIGNALS
 
 def run_module():
-
+    """
+    Calls the method for handling the wip signals
+    Returns
+    -------
+    prints the updated signal names
+    """
     params = read_params()
+    wip_signal = params["wip_signal"]
+    signal_names = add_prefix(SIGNALS, wip_signal, prefix="wip_")
+    print(signal_names)

_template_python/params.json.template

@@ -1,5 +1,6 @@
 {
   "static_file_dir": "./static",
   "export_dir": "./receiving",
-  "cache_dir": "./cache"
+  "cache_dir": "./cache",
+  "wip_signal": ""
 }

_template_python/setup.py

@@ -7,7 +7,8 @@
     "pytest",
     "pytest-cov",
     "pylint",
-    "delphi-utils"
+    "delphi-utils",
+    "covidcast"
 ]
 
 setup(

_template_python/tests/params.json.template

@@ -1,5 +1,6 @@
 {
   "static_file_dir": "../static",
   "export_dir": "./receiving",
-  "cache_dir": "./cache"
+  "cache_dir": "./cache",
+  "wip_signal": ""
 }

_template_python/tests/test_handle_wip_signal.py

@@ -0,0 +1,27 @@
+import unittest
+from delphi_NAME.handle_wip_signal import add_prefix
+from delphi_NAME.run import SIGNALS
+from delphi_utils import read_params
+
+
+
+def test_handle_wip_signal():
+    # Test wip_signal = True (all signals should receive prefix)
+    signal_names = add_prefix(SIGNALS, 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(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 (only unpublished signals should receive prefix)
+    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:])
+
+
+class MyTestCase(unittest.TestCase):
+    pass
+
+
+if __name__ == '__main__':
+    unittest.main()