handling wip signal in python template

Author: vishakha1812

_template_python/delphi_NAME/handle_wip_signal.py

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

_template_python/delphi_NAME/run.py

@@ -4,11 +4,25 @@
 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
 
+# Sample signals
+SIGNALS = ["full_time_work_prop",
+           "covidnet",
+           "part_time_work_prop",
+           "completely_home_prop"
+           ]
 
-def run_module():
 
+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/tests/test_handle_wip_signal.py

@@ -0,0 +1,22 @@
+import unittest
+from delphi_NAME.handle_wip_signal import add_prefix
+from delphi_NAME.run import SIGNALS
+from delphi_utils import read_params
+
+wip_signal = read_params()["wip_signal"]
+
+
+def test_handle_wip_signal():
+    assert isinstance(wip_signal, (list, bool)) or wip_signal == "", "Supply True | False or "" or [] | list()"
+    if isinstance(wip_signal, list):
+        assert set(wip_signal).issubset(set(SIGNALS)), "signal in params don't belong in the registry"
+    updated_signal_names = add_prefix(SIGNALS, wip_signal, prefix='wip_')
+    assert (len(updated_signal_names) >= len(SIGNALS))
+
+
+class MyTestCase(unittest.TestCase):
+    pass
+
+
+if __name__ == '__main__':
+    unittest.main()