Core module tests completed

Author: fronzbot

adc_eval/eval/spectrum.py

@@ -38,7 +38,12 @@ def get_spectrum(data, fs=1, nfft=2**12, single_sided=True):
 
 def window_data(data, window="rectangular"):
     """Applies a window to the time-domain data."""
-    wsize = data.size
+    try:
+        wsize = data.size
+    except AttributeError:
+        data = np.array(data)
+        wsize = data.size
+
     windows = {
         "rectangular": (np.ones(wsize), 1.0),
         "hanning": (np.hanning(wsize), 1.633)
@@ -64,26 +69,10 @@ def plot_spectrum(
     no_plot=False,
     yaxis="power",
     single_sided=True,
-    fscale="MHz",
+    fscale=("MHz", 1e6),
 ):
     """Plot Power Spectrum for input signal."""
-
-    fscalar = {
-        "uHz": 1e-6,
-        "mHz": 1e-3,
-        "Hz": 1,
-        "kHz": 1e3,
-        "MHz": 1e6,
-        "GHz": 1e9,
-        "THz": 1e12,
-    }
-    if fscale not in fscalar:
-        print(f"WARNING: {fscale} not implemented. Defaulting to 'MHz'.")
-        fscale = "MHz"
-
-    # Window the data and get the single or dual-sided spectrum
-    wdata = window_data(data, window=window)
-    (freq, pwr) = get_spectrum(wdata, fs=fs, nfft=nfft, single_sided=single_sided)
+    (freq, pwr) = get_spectrum(data, fs=fs, nfft=nfft, single_sided=single_sided)
 
     # Calculate the fullscale range of the spectrum in Watts
     full_scale = calc.dBW(dr**2 / 8)
@@ -99,13 +88,7 @@ def plot_spectrum(
     lut_key = yaxis.lower()
     scalar = yaxis_lut[lut_key][0]
     yunits = yaxis_lut[lut_key][1]
-    try:
-        xscale = fscalar[fscale]
-    except KeyError:
-        print(
-            f"WARNING: {fscale} not a valid option for fscale. Valid inputs are {fscalar.keys()}."
-        )
-        print("         Defaulting to Hz.")
+    xscale = fscale[1]
 
     # Convert to dBW and perform scalar based on y-axis scaling input
     psd_out = calc.dBW(pwr, places=3) - scalar
@@ -119,9 +102,7 @@ def plot_spectrum(
     psd_ss = pwr
     if not single_sided:
         # Get single-sided spectrum for SNDR and Harmonic stats
-        (f_ss, psd_ss) = get_spectrum(
-            data * windows[window] * wscale, fs=fs, nfft=nfft, single_sided=True
-        )
+        (f_ss, psd_ss) = get_spectrum(data, fs=fs, nfft=nfft, single_sided=True)
 
     sndr_stats = calc.sndr_sfdr(psd_ss, f_ss, fs, nfft, leak=leak, full_scale=full_scale)
     harm_stats = calc.find_harmonics(
@@ -143,7 +124,7 @@ def plot_spectrum(
 
     # If plotting, prep plot and generate all required axis strings
     if not no_plot:
-        plt_str = calc.get_plot_string(stats, full_scale, fs, nfft, window, xscale, fscale)
+        plt_str = calc.get_plot_string(stats, full_scale, fs, nfft, window, xscale, fscale[0])
         fig, ax = plt.subplots(figsize=(15, 8))
         ax.plot(freq / xscale, psd_out)
         ax.set_ylabel(f"Power Spectrum ({yunits})", fontsize=18)
@@ -226,8 +207,24 @@ def analyze(
     fscale="MHz",
 ):
     """Perform spectral analysis on input waveform."""
+    fscalar = {
+        "uHz": 1e-6,
+        "mHz": 1e-3,
+        "Hz": 1,
+        "kHz": 1e3,
+        "MHz": 1e6,
+        "GHz": 1e9,
+        "THz": 1e12,
+    }
+    if fscale not in fscalar:
+        print(f"WARNING: {fscale} not implemented. Defaulting to 'MHz'.")
+        fscale = "MHz"
+    
+    # Window the data
+    wdata = window_data(data, window=window)
+    
     (freq, spectrum, stats) = plot_spectrum(
-        data,
+        wdata,
         fs=fs,
         nfft=nfft,
         dr=dr,
@@ -237,7 +234,7 @@ def analyze(
         no_plot=no_plot,
         yaxis=yaxis,
         single_sided=single_sided,
-        fscale=fscale,
+        fscale=(fscale, fscalar[fscale]),
     )
 
     return (freq, spectrum, stats)

tests/eval/test_spectrum.py

@@ -8,7 +8,7 @@
 
 RTOL = 0.05
 NLEN = 2**18
-NFFT = 2**8
+NFFT = 2**10
 DATA_SINE = [
     {
         "f1": np.random.randint(1, NFFT / 4 - 1),
@@ -175,3 +175,155 @@ def test_calc_psd_two_sine_single(data):
 
     assert np.allclose(peak2, exp_peaks[1], rtol=RTOL), assertmsg
     assert np.allclose(fpeak, exp_f2), assertmsg
+
+
+def test_window_data_as_list():
+    """Tests the window_data function when given a list instead of numpy array."""
+    data = np.random.rand(NLEN).tolist()
+    wdata = spectrum.window_data(data, window="rectangular")
+    
+    assert type(data) == type(list())
+    assert type(wdata) == type(np.ndarray([]))
+
+
+def test_window_data_bad_window_type(capfd):
+    """Tests the window_data function with an incorrect window selection."""
+    data = np.random.rand(NLEN)
+    wdata = spectrum.window_data(data, window="foobar")
+    captured = capfd.readouterr()
+    
+    assert data.size == wdata.size
+    assert data.all() == wdata.all()
+    assert "WARNING" in captured.out
+
+
+@mock.patch("adc_eval.eval.spectrum.plot_spectrum")
+def test_analyze_bad_input_scalar(mock_plot_spectrum, capfd):
+    """Tests bad input scalar keys."""
+    mock_plot_spectrum.return_value = (None, None, None)
+    mock_plot_spectrum.side_effect = lambda *args, **kwargs: (kwargs, None, None)
+    (kwargs, _, _) = spectrum.analyze([0], 1, fscale="foobar")
+    captured = capfd.readouterr()
+
+    assert "WARNING" in captured.out
+    assert kwargs.get("fscale") == ("MHz", 1e6)
+
+
+@mock.patch("adc_eval.eval.spectrum.plot_spectrum")
+def test_analyze_valid_input_scalar(mock_plot_spectrum):
+    """Tests the valid input scalar keys."""
+    mock_plot_spectrum.return_value = (None, None, None)
+    mock_plot_spectrum.side_effect = lambda *args, **kwargs: (kwargs, None, None)
+    
+    test_vals = {
+        "Hz": 1,
+        "kHz": 1e3,
+        "MHz": 1e6,
+        "GHz": 1e9,
+    }
+    for key, val in test_vals.items():
+        (kwargs, _, _) = spectrum.analyze([0], 1, fscale=key)
+        assert kwargs.get("fscale") == (key, val)
+
+
+@mock.patch("adc_eval.eval.calc.sndr_sfdr")
+@mock.patch("adc_eval.eval.calc.find_harmonics")
+def test_analyze_no_plot(mock_sndr_sfdr, mock_find_harmonics):
+    """Tests the psd output of the analyze function with no plotting."""
+    data = np.random.rand(NLEN)
+    data_sndr = {
+        "sig": {"bin": 1, "power": 2},
+    }
+    data_harms = {"harmonics": 3}
+    exp_stats = {**data_sndr, **data_harms}
+    
+    mock_sndr_sfdr.return_value = data_sndr
+    mock_find_harmonics = data_harms
+    
+    (freq, psd, stats) = spectrum.analyze(
+        data,
+        fs=1,
+        nfft=NFFT,
+        dr=1,
+        harmonics=0,
+        leak=0,
+        window="rectangular",
+        no_plot=True,
+        yaxis="power",
+        single_sided=True,
+        fscale="Hz",
+    )
+    
+    assert freq.all() == np.linspace(0, 1, int(NFFT/2)).all()
+    assert psd.size == int(NFFT/2)
+    
+    for key, value in stats.items():
+        assert value == exp_stats[key]
+
+
+@mock.patch("adc_eval.eval.calc.sndr_sfdr")
+@mock.patch("adc_eval.eval.calc.find_harmonics")
+def test_analyze_no_plot_dual(mock_sndr_sfdr, mock_find_harmonics):
+    """Tests the psd output of the analyze function with no plotting."""
+    data = np.random.rand(NLEN)
+    data_sndr = {
+        "sig": {"bin": 1, "power": 2},
+    }
+    data_harms = {"harmonics": 3}
+    exp_stats = {**data_sndr, **data_harms}
+    
+    mock_sndr_sfdr.return_value = data_sndr
+    mock_find_harmonics = data_harms
+    
+    (freq, psd, stats) = spectrum.analyze(
+        data,
+        fs=1,
+        nfft=NFFT,
+        dr=1,
+        harmonics=0,
+        leak=0,
+        window="rectangular",
+        no_plot=True,
+        yaxis="power",
+        single_sided=False,
+        fscale="Hz",
+    )
+    
+    assert freq.all() == np.linspace(-0.5, 0.5, NFFT-1).all()
+    assert psd.size == NFFT
+    for key, value in stats.items():
+        assert value == exp_stats[key]
+
+
+@mock.patch("adc_eval.eval.calc.sndr_sfdr")
+@mock.patch("adc_eval.eval.calc.find_harmonics")
+def test_analyze_no_plot_magnitude(mock_sndr_sfdr, mock_find_harmonics):
+    """Tests the psd output of the analyze function with no plotting."""
+    data = np.random.rand(NLEN)
+    data_sndr = {
+        "sig": {"bin": 1, "power": 2},
+    }
+    data_harms = {"harmonics": 3}
+    exp_stats = {**data_sndr, **data_harms}
+    
+    mock_sndr_sfdr.return_value = data_sndr
+    mock_find_harmonics = data_harms
+    
+    (freq, psd, stats) = spectrum.analyze(
+        data,
+        fs=1,
+        nfft=NFFT,
+        dr=1,
+        harmonics=0,
+        leak=0,
+        window="rectangular",
+        no_plot=True,
+        yaxis="magnitude",
+        single_sided=True,
+        fscale="Hz",
+    )
+    
+    assert freq.all() == np.linspace(0, 1, int(NFFT/2)).all()
+    assert psd.size == int(NFFT/2)
+    for key, value in stats.items():
+        assert value == exp_stats[key]

tests/test_signals.py

@@ -0,0 +1,69 @@
+"""Test the signals module."""
+
+import pytest
+import numpy as np
+from scipy import stats
+from adc_eval import signals
+
+
+RTOL = 0.01
+
+@pytest.mark.parametrize("nlen", np.random.randint(4, 2**16, 3))
+@pytest.mark.parametrize("fs", np.random.uniform(1, 1e9, 3))
+def test_time(nlen, fs):
+    """Test time with random data."""
+    value = signals.time(nlen, fs=fs)
+    assert value.size == nlen
+    assert value[0] == 0
+    assert np.isclose(value[nlen-1], (nlen-1)/fs, rtol=RTOL)
+    
+
+@pytest.mark.parametrize("nlen", np.random.randint(2**10, 2**16, 3))
+@pytest.mark.parametrize("offset", np.random.uniform(-10, 10, 3))
+@pytest.mark.parametrize("amp", np.random.uniform(0, 10, 3))
+def test_sin(nlen, offset, amp):
+    """Test sine generation with random data."""
+    fs = np.random.uniform(1, 1e9)
+    fin = np.random.uniform(fs/10, fs/3)
+    
+    t = signals.time(nlen, fs=fs)
+    value = signals.sin(t, amp=amp, offset=offset, freq=fin)
+    
+    exp_peaks = [offset - amp, amp + offset]
+    
+    assert value.size == nlen
+    assert np.isclose(max(value), exp_peaks[1], rtol=RTOL)
+    assert np.isclose(min(value), exp_peaks[0], rtol=RTOL)
+    assert value[0] == offset
+
+
+@pytest.mark.parametrize("nlen", np.random.randint(1, 2**16, 5))
+def test_noise_length(nlen):
+    """Test noise generation with random data."""
+    t = signals.time(nlen, fs=1)
+    value = signals.noise(t, mean=0, std=1)
+    
+    # Just check correct size
+    assert value.size == nlen
+    
+
+@pytest.mark.parametrize("std", np.random.uniform(0, 1, 4))
+def test_noise_length(std):
+    """Test noise is gaussian with random data."""
+    nlen = 2**12
+    t = signals.time(nlen, fs=1)
+    noise = signals.noise(t, mean=0, std=std)
+    autocorr = np.correlate(noise, noise, mode="full")
+    autocorr /= max(autocorr)
+    asize = autocorr.size
+    
+    midlag = autocorr.size // 2
+    acorr_nopeak = np.concatenate([autocorr[0:midlag-1], autocorr[midlag+1:]])
+    
+    shapiro = stats.shapiro(acorr_nopeak)
+    
+    # Check that middle lag is 1
+    assert autocorr[midlag] == 1
+    
+    # Now check that noise is gaussian
+    assert shapiro.pvalue < 0.01