Repository: legend-exp/pygama (language: python)

Issue:
Add DPLMS DSP processor
DSP processor for LAr instrumentation SiPMs using the DPLMS method for optimum filter calculation (arXiv:2209.09863).
This processor is composed of a factory function "dplms_filter" that is called using the `init_args` argument. The input and output waveforms are passed using `args`.
The function "dplms_filter" input arguments are:
- noise_mat: list noise matrix (from the database)
- reference: list reference signal (from the database)
- length: int filter length
- a1: float penalized coefficient for the noise matrix
- a2: float penalized coefficient for the reference matrix
- a3: float penalized coefficient for the zero area matrix
- ff: intflat top length for the reference signal
A test config file for SiPM processing has been included.


Discussion / hints:
Add a test please.
> Add a test please.

See https://pygama.readthedocs.io/en/latest/developer.html#testing-numba-wrapped-functions

Fix patch:
diff --git a/docs/source/developer.rst b/docs/source/developer.rst
index d69f7c9e6..5d3c85dcf 100644
--- a/docs/source/developer.rst
+++ b/docs/source/developer.rst
@@ -192,6 +192,59 @@ required imports in the snippet for verbosity.
 In this case, the general idea is to use :func:`pytest.raises` twice, once with
 ``func(*inputs)``, and again with ``inspect.unwrap(func)(*inputs)``.
 
+Testing Factory Functions that Return Numba-Wrapped Functions
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+As in the previous section, we also have processors that are first initialized
+with a factory function, which then returns a callable Numba-wrapped function.
+In this case, there is a slightly different way of testing the function to ensure
+full code coverage when using ``compare_numba_vs_python``, as the function
+signature is generally different.
+
+As an example, we show a snippet from the test for
+:func:`pygama.dsp.processors.dwt.discrete_wavelet_transform`, a processor which uses
+a factory function to return a function wrapped by the
+:func:`@numba.guvectorize()<numba.guvectorize>` decorator.
+
+.. code-block:: python
+
+    import numpy as np
+    import pytest
+
+    from pygama.dsp.errors import DSPFatal
+    from pygama.dsp.processors import discrete_wavelet_transform
+
+    def test_discrete_wavelet_transform(compare_numba_vs_python):
+        """Testing function for the discrete_wavelet_transform processor."""
+
+        # set up values to use for each test case
+        len_wf_in = 16
+        wave_type = 'haar'
+        level = 2
+        len_wf_out = 4
+
+        # ensure the DSPFatal is raised for a negative level
+        with pytest.raises(DSPFatal):
+            discrete_wavelet_transform(wave_type, -1)
+
+        # ensure that a valid input gives the expected output
+        w_in = np.ones(len_wf_in)
+        w_out = np.empty(len_wf_out)
+        w_out_expected = np.ones(len_wf_out) * 2**(level / 2)
+
+        dwt_func = discrete_wavelet_transform(wave_type, level)
+        assert np.allclose(
+            compare_numba_vs_python(dwt_func, w_in, w_out),
+            w_out_expected,
+        )
+        ## rest of test function is truncated in this example
+
+In this case, the error is raised outside of the Numba-wrapped function, and
+we only need to test for the error once. For the comparison of the calculated
+values to expectation, we must initialize the output array and pass it to the
+list of inputs that should be used in the comparison. This is different than
+the previous section, where we are instead now updating the outputted values
+in place.
 
 Documentation
 -------------


Test patch:
diff --git a/tests/conftest.py b/tests/conftest.py
index 99c7e57b9..7c245a889 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -1,3 +1,4 @@
+import copy
 import inspect
 import os
 import re
@@ -103,25 +104,39 @@ def numba_vs_python(func, *inputs):
             The output of the function to be used in a unit test.
 
         """
-        # parse outputs from function signature
-        all_params = list(inspect.signature(func).parameters)
-        output_sizes = re.findall(r"(\(n*\))", func.signature.split("->")[-1])
-        noutputs = len(output_sizes)
-        output_names = all_params[-noutputs:]
-
-        # numba outputs
-        outputs_numba = func(*inputs)
-        if noutputs == 1:
-            outputs_numba = [outputs_numba]
-
-        # unwrapped python outputs
-        func_unwrapped = inspect.unwrap(func)
-        output_dict = {key: np.empty(len(inputs[0])) for key in output_names}
-        func_unwrapped(*inputs, **output_dict)
-        for spec, key in zip(output_sizes, output_dict):
-            if spec == "()":
-                output_dict[key] = output_dict[key][0]
-        outputs_python = [output_dict[key] for key in output_names]
+
+        if "->" in func.signature:
+            # parse outputs from function signature
+            all_params = list(inspect.signature(func).parameters)
+            output_sizes = re.findall(r"(\(n*\))", func.signature.split("->")[-1])
+            noutputs = len(output_sizes)
+            output_names = all_params[-noutputs:]
+
+            # numba outputs
+            outputs_numba = func(*inputs)
+            if noutputs == 1:
+                outputs_numba = [outputs_numba]
+
+            # unwrapped python outputs
+            func_unwrapped = inspect.unwrap(func)
+            output_dict = {key: np.empty(len(inputs[0])) for key in output_names}
+            func_unwrapped(*inputs, **output_dict)
+            for spec, key in zip(output_sizes, output_dict):
+                if spec == "()":
+                    output_dict[key] = output_dict[key][0]
+            outputs_python = [output_dict[key] for key in output_names]
+        else:
+            # we are testing a factory function output, which updates
+            # a single output in-place
+            noutputs = 1
+            # numba outputs
+            func(*inputs)
+            outputs_numba = copy.deepcopy(inputs[-noutputs:])
+
+            # unwrapped python outputs
+            func_unwrapped = inspect.unwrap(func)
+            func_unwrapped(*inputs)
+            outputs_python = copy.deepcopy(inputs[-noutputs:])
 
         # assert that numba and python are the same up to floating point
         # precision, setting nans to be equal
diff --git a/tests/dsp/processors/test_dwt.py b/tests/dsp/processors/test_dwt.py
new file mode 100644
index 000000000..0bec85f60
--- /dev/null
+++ b/tests/dsp/processors/test_dwt.py
@@ -0,0 +1,38 @@
+import numpy as np
+import pytest
+
+from pygama.dsp.errors import DSPFatal
+from pygama.dsp.processors import discrete_wavelet_transform
+
+
+def test_discrete_wavelet_transform(compare_numba_vs_python):
+    """Testing function for the discrete_wavelet_transform processor."""
+
+    # set up values to use for each test case
+    len_wf_in = 16
+    wave_type = "haar"
+    level = 2
+    len_wf_out = 4
+
+    # ensure the DSPFatal is raised for a negative level
+    with pytest.raises(DSPFatal):
+        discrete_wavelet_transform(wave_type, -1)
+
+    # ensure that a valid input gives the expected output
+    w_in = np.ones(len_wf_in)
+    w_out = np.empty(len_wf_out)
+    w_out_expected = np.ones(len_wf_out) * 2 ** (level / 2)
+
+    dwt_func = discrete_wavelet_transform(wave_type, level)
+    assert np.allclose(
+        compare_numba_vs_python(dwt_func, w_in, w_out),
+        w_out_expected,
+    )
+
+    # ensure that if there is a nan in w_in, all nans are outputted
+    w_in = np.ones(len_wf_in)
+    w_in[4] = np.nan
+    w_out = np.empty(len_wf_out)
+
+    dwt_func = discrete_wavelet_transform(wave_type, level)
+    assert np.all(np.isnan(compare_numba_vs_python(dwt_func, w_in, w_out)))