Added filter tests

Author: fronzbot

adc_eval/eval/spectrum.py

@@ -132,7 +132,7 @@ def plot_spectrum(
         fig, ax = plt.subplots(figsize=(15, 8))
         ax.plot(freq / xscale, psd_out)
         ax.set_ylabel(f"Power Spectrum ({yunits})", fontsize=18)
-        ax.set_xlabel(f"Frequency ({fscale})", fontsize=16)
+        ax.set_xlabel(f"Frequency ({fscale[0]})", fontsize=16)
         ax.set_title("Output Power Spectrum", fontsize=16)
         ax.set_xlim([xmin, fs / (2 * xscale)])
         ax.set_ylim([1.1 * min(psd_out), 1])

adc_eval/filter.py renamed to adc_eval/filt.py

@@ -3,6 +3,7 @@
 import numpy as np
 from scipy.signal import remez, freqz
 import matplotlib.pyplot as plt
+from adc_eval import signals
 from adc_eval.eval import spectrum
 from adc_eval.eval import calc
 
@@ -38,13 +39,35 @@ class CICDecimate:
 
     def __init__(self, dec=2, order=1, fs=1):
         """Initialize the CIC filter."""
-        self.dec = dec
-        self.order = order
+        self._dec = dec
+        self._order = order
         self.fs = fs
-        self.gain = self.dec**self.order
+        self.gain = dec**order
         self._xout = None
         self._xfilt = None
 
+    @property
+    def dec(self):
+        """Returns the decimation factor."""
+        return self._dec
+
+    @dec.setter
+    def dec(self, value):
+        """Sets the decimation factor."""
+        self._dec = value
+        self.gain = value**self._order
+
+    @property
+    def order(self):
+        """Returns the order of the filter."""
+        return self._order
+
+    @order.setter
+    def order(self, value):
+        """Sets the filter order."""
+        self._order = value
+        self.gain = self.dec**value
+
     @property
     def out(self):
         """Filtered and decimated output data."""
@@ -70,9 +93,11 @@ def filt(self, xarray):
 
         self._xfilt = ycomb / self.gain
 
-    def decimate(self):
+    def decimate(self, xarray=None):
         """decimation routine."""
-        self._xout = self._xfilt[:: self.dec]
+        if xarray is None:
+            xarray = self._xfilt
+        self._xout = xarray[:: self.dec]
 
     def run(self, xarray):
         """Runs filtering and decimation on input list."""
@@ -81,8 +106,7 @@ def run(self, xarray):
 
     def response(self, fft, no_plot=False):
         """Plots the frequency response of the pre-decimated filter."""
-        xin = np.zeros(fft)
-        xin[0] = 1
+        xin = signals.impulse(fft)
         self.filt(xin)
         (freq, psd, stats) = spectrum.analyze(
             self._xfilt * fft / np.sqrt(2),
@@ -147,7 +171,7 @@ def __init__(self, dec=1, fs=1, bit_depth=16, coeffs=None):
         self.dec = dec
         self.fs = fs
         self.bit_depth = bit_depth
-        self.ntaps = np.size(coeffs) if coeffs is None else 0
+        self.ntaps = np.size(coeffs) if coeffs is not None else 0
         self.yfilt = None
         self._out = None
 

adc_eval/signals.py

@@ -16,3 +16,10 @@ def sin(t, amp=0.5, offset=0.5, freq=1e3, ph0=0):
 def noise(t, mean=0, std=0.1):
     """Generate random noise."""
     return np.random.normal(mean, std, size=len(t))
+
+
+def impulse(nsamp, mag=1):
+    """Generate an impulse input."""
+    data = np.zeros(nsamp)
+    data[0] = mag
+    return data

tests/test_filt.py

@@ -0,0 +1,177 @@
+"""Test the filter module."""
+
+import pytest
+import numpy as np
+from unittest import mock
+from adc_eval import filt
+from adc_eval import signals
+
+
+@pytest.mark.parametrize("dec", np.random.randint(1, 20, 4))
+def test_cic_decimate_set_dec_updates_gain(dec):
+    """Tests that changing decimation factor updates gain."""
+    cicfilt = filt.CICDecimate(dec=1, order=2)
+    assert cicfilt.gain == 1
+
+    cicfilt.dec = dec
+    assert cicfilt.gain == (dec**2)
+
+
+@pytest.mark.parametrize("order", np.random.randint(1, 20, 4))
+def test_cic_decimate_set_order_updates_gain(order):
+    """Tests that changing filter order updates gain."""
+    cicfilt = filt.CICDecimate(dec=2, order=1)
+    assert cicfilt.gain == 2
+
+    cicfilt.order = order
+    assert cicfilt.gain == (2**order)
+
+
+def test_cic_decimate_returns_ndarray():
+    """Tests the CICDecimate output data conversion."""
+    cicfilt = filt.CICDecimate()
+    data = np.random.randn(100).tolist()
+    cicfilt._xout = data
+
+    assert type(cicfilt.out) == type(np.array(list()))
+    assert cicfilt.out.all() == np.array(data).all()
+
+
+@pytest.mark.parametrize("dec", np.random.randint(1, 20, 4))
+def test_cic_decimate_function(dec):
+    """Tests the CICDecimate decimate function."""
+    cicfilt = filt.CICDecimate(dec=dec)
+    data = np.random.randn(100)
+    cicfilt.decimate(data)
+
+    exp_result = data[::dec]
+
+    assert cicfilt.out.size == exp_result.size
+    assert cicfilt.out.all() == exp_result.all()
+
+
+def test_cic_decimate_function_none_input():
+    """Tests the CICDecimate decimate function with no input arg."""
+    cicfilt = filt.CICDecimate(dec=1)
+    data = np.random.randn(100)
+    cicfilt._xfilt = data
+    cicfilt.decimate()
+
+    exp_result = data
+
+    assert cicfilt.out.size == exp_result.size
+    assert cicfilt.out.all() == exp_result.all()
+
+
+@pytest.mark.parametrize("nlen", np.random.randint(8, 2**10, 4))
+def test_cic_decimate_all_ones(nlen):
+    """Test the CICDecimate filtering with all ones."""
+    cicfilt = filt.CICDecimate(dec=1, order=1)
+    data = np.ones(nlen)
+    cicfilt.run(data)
+
+    exp_data = data.copy()
+    exp_data[0] = 0
+
+    assert cicfilt.out.all() == exp_data.all()
+
+
+@pytest.mark.parametrize("nlen", np.random.randint(8, 2**10, 4))
+def test_cic_decimate_all_zeros(nlen):
+    """Test the CICDecimate filtering with all zeros."""
+    cicfilt = filt.CICDecimate(dec=1, order=1)
+    data = np.zeros(nlen)
+    cicfilt.run(data)
+
+    exp_data = data.copy()
+
+    assert cicfilt.out.all() == exp_data.all()
+
+
+@pytest.mark.parametrize("nlen", np.random.randint(8, 2**10, 4))
+def test_cic_decimate_impulse(nlen):
+    """Test the CICDecimate filtering with impulse."""
+    cicfilt = filt.CICDecimate(dec=1, order=1)
+    data = signals.impulse(nlen)
+    cicfilt.run(data)
+
+    exp_data = np.concatenate([[0], data[0:-1]])
+
+    assert cicfilt.out.all() == exp_data.all()
+
+
+def test_fir_lowpass_returns_ndarray():
+    """Tests the FIRLowPass output data conversion."""
+    fir = filt.FIRLowPass()
+    data = np.random.randn(100).tolist()
+    fir._out = data
+
+    assert type(fir.out) == type(np.array(list()))
+    assert fir.out.all() == np.array(data).all()
+
+
+@pytest.mark.parametrize("dec", np.random.randint(1, 20, 4))
+def test_fir_decimate_function(dec):
+    """Tests the FIRLowPass decimate function."""
+    fir = filt.FIRLowPass(dec=dec)
+    data = np.random.randn(100)
+    fir.decimate(data)
+
+    exp_result = data[::dec]
+
+    assert fir.out.size == exp_result.size
+    assert fir.out.all() == exp_result.all()
+
+
+def test_fir_decimate_function_none_input():
+    """Tests the FIRLowPass decimate function with no input arg."""
+    fir = filt.FIRLowPass(dec=1)
+    data = np.random.randn(100)
+    fir.yfilt = data
+    fir.decimate()
+
+    exp_result = data
+
+    assert fir.out.size == exp_result.size
+    assert fir.out.all() == exp_result.all()
+
+
+@mock.patch("adc_eval.filt.remez")
+def test_fir_lowpass_tap_generation(mock_remez, capfd):
+    """Tests the FIRLowPass decimate function."""
+    fir = filt.FIRLowPass()
+    fir.ntaps = 3
+    fir.bit_depth = 12
+    mock_remez.return_value = np.ones(3)
+
+    (taps, coeffs) = fir.generate_taps(0.1)
+
+    captured = capfd.readouterr()
+    exp_coeffs = [2**12, 2**12, 2**12]
+
+    assert "WARNING" in captured.out
+    assert taps == 3
+    assert coeffs == exp_coeffs
+
+
+@pytest.mark.parametrize("ntaps", np.random.randint(3, 511, 5))
+def test_fir_lowpass_run(ntaps):
+    """Tests the FIRLowPass run function."""
+    fir = filt.FIRLowPass()
+    fir.ntaps = ntaps
+    fir.bit_depth = 10
+    fir.coeffs = 2**10 * np.ones(ntaps)
+    data = signals.impulse(2**12)
+    fir.run(data)
+
+    exp_sum = np.ceil((ntaps + 1) / 2)
+    out_sum = sum(fir.out)
+
+    tap_val = int(exp_sum)
+
+    assert fir.out.size == data.size
+    assert max(fir.out) == 1
+    assert min(fir.out) == 0
+    assert fir.out[0:tap_val].all() == 1
+    assert fir.out[tap_val + 1 :].all() == 0
+    assert out_sum == exp_sum

tests/test_signals.py

@@ -68,3 +68,14 @@ def test_noise_length(std):
 
     # Now check that noise is gaussian
     assert shapiro.pvalue < 0.01
+
+
+@pytest.mark.parametrize("nlen", np.random.randint(2, 2**12, 3))
+@pytest.mark.parametrize("mag", np.random.uniform(0.1, 100, 3))
+def test_impulse(nlen, mag):
+    """Test impulse generation with random length and amplitude."""
+    data = signals.impulse(nlen, mag)
+
+    assert data.size == nlen
+    assert data[0] == mag
+    assert data[1:].all() == 0