Skip to content

Commit

Permalink
Fix tests on compute_delays
Browse files Browse the repository at this point in the history
  • Loading branch information
jranalli committed Nov 14, 2023
1 parent 20997ce commit 9c46180
Show file tree
Hide file tree
Showing 2 changed files with 11 additions and 50 deletions.
2 changes: 0 additions & 2 deletions src/solartoolbox/signalproc.py
Original file line number Diff line number Diff line change
Expand Up @@ -412,8 +412,6 @@ def compute_delays(ts_in, ts_out, mode='loop', scaling='coeff'):
Time lag between the two timeseries at the maximum value of the cross
correlation. Values are always integer multiples of the sampling period
as the max correlation values are limited to the discrete time steps.
corr : array(float)
The peak value of the cross correlation at the identified delay.
extra_data : dict{} or None
if compute_extras was true, additional info will be returned. Fields
are:
Expand Down
59 changes: 11 additions & 48 deletions tests/test_signalproc.py
Original file line number Diff line number Diff line change
Expand Up @@ -52,17 +52,15 @@ def test_correlation_illegal(corr_data):
with raises(ValueError):
c, lag = correlation(x1, x2, scaling="illegal")


@pytest.mark.parametrize("scaling", ['coeff', 'none'])
def test_fftcorrelate_identity(corr_data, scaling):
dt, t, x1, x2, dly = corr_data
c = _fftcorrelate(x1, x1, scaling)
cr, lag = correlation(x1, x1, scaling=scaling)
# import matplotlib.pyplot as plt
# plt.plot(c)
# plt.plot(cr)
# plt.show()
assert np.allclose(c, cr)


@pytest.mark.parametrize("scaling", ['coeff', 'none'])
def test_fftcorrelate_shift(corr_data, scaling):
d, t, x1, x2, dly = corr_data
Expand All @@ -71,7 +69,6 @@ def test_fftcorrelate_shift(corr_data, scaling):
assert np.allclose(c, cr)



def test_averaged_psd():
# Create a simple sinusoidal signal
fs = 10 # sample rate
Expand Down Expand Up @@ -216,8 +213,6 @@ def test_apply_delay(delay):
assert np.allclose(tf_delayed['tf'], tf_expected['tf'], atol=1e-5)




def test_averaged_psd_multi():
np.random.seed(2023)
# Create a simple sinusoidal signal
Expand All @@ -239,8 +234,6 @@ def test_averaged_psd_multi():
scaling = 'density'
psds = [averaged_psd(tsig, navgs, overlap, window, detrend, scaling) for tsig in tsigs]



# Calculate the PSD directly using scipy.signal.welch
freqs, psd_direct = signal.welch(xs, fs, window, nperseg=len(x1)//navgs,
noverlap=int(overlap*len(x1)//navgs),
Expand All @@ -258,7 +251,6 @@ def test_averaged_tf_multi():
t = np.linspace(0, T, int(T*fs), endpoint=False) # time variable
delay = 3


x = 0.5 * np.sin(2 * np.pi * 2 * t) + np.random.random(len(t))

y1 = np.roll(x, int(delay*fs))
Expand Down Expand Up @@ -288,11 +280,11 @@ def test_averaged_tf_multi():
noverlap=int(overlap*len(x)//navgs),
detrend=detrend)
sometf_direct = somePxy / somePxx
import matplotlib.pyplot as plt
plt.plot(freqs, np.abs(sometf_direct).T)
plt.figure()
plt.plot(freqs, np.angle(sometf_direct).T)
plt.show()
# import matplotlib.pyplot as plt
# plt.plot(freqs, np.abs(sometf_direct).T)
# plt.figure()
# plt.plot(freqs, np.angle(sometf_direct).T)
# plt.show()


# Calculate the PSD directly using scipy.signal.welch
Expand All @@ -308,42 +300,17 @@ def test_averaged_tf_multi():
# Check that the PSDs match
assert np.allclose(tfs, tf_direct, atol=1e-5)

def test_correlation_multi():
np.random.seed(2023)
# Create a simple sinusoidal signal
fs = 10 # sample rate
T = 5.0 # seconds
t = np.linspace(0, T, int(T * fs), endpoint=False) # time variable
delay = 0.25

x = 0.5 * np.sin(2 * np.pi * 2 * t) + np.random.random(len(t))

y1 = np.roll(x, int(delay * fs))
y2 = np.roll(x, int(2 * delay * fs))
y3 = np.roll(x, int(3 * delay * fs))
y4 = np.roll(x, int(4 * delay * fs))

ys = [y1, y2, y3, y4]
x_tsig = pd.Series(x, index=pd.TimedeltaIndex(t, 's'))
ysigs = [pd.Series(y, index=pd.TimedeltaIndex(t, 's')) for y in ys]

corr = [correlation(x_tsig, y_tsig, scaling='none')[0] for y_tsig in ysigs]
corr = np.array(corr)

c = np.flip(signal.correlate([x_tsig], ysigs, method='fft'), axis=0)

assert np.allclose(corr, c)

@pytest.fixture(params=['loop', 'fft'])
def compute_delays_modes(request):
return request.param


@pytest.mark.parametrize("delay", [-200,-50.0, -25, -5, 0.0, 5, 25, 50,200])
@pytest.mark.parametrize("delay", [-200, -50.0, -25, -5, 0.0, 5, 25, 50, 200])
def test_compute_delays(delay, compute_delays_modes):
np.random.seed(2023)
# Create a simple sinusoidal signal
fs = 250 # sample rate
fs = 100 # sample rate
T = 500.0 # seconds
t = np.linspace(0, T, int(T * fs), endpoint=False) # time variable
# delay = 0.25
Expand All @@ -358,16 +325,12 @@ def test_compute_delays(delay, compute_delays_modes):
yi = np.roll(x, int(deli * fs))
ys.append(yi)
delay_ins.append(deli)
# y2 = np.roll(x, int(2 * delay * fs))
# y3 = np.roll(x, int(3 * delay * fs))
# y4 = np.roll(x, int(4 * delay * fs))
#
# ys = [x, y1, y2, y3, y4]

x_tsig = pd.Series(x, index=pd.TimedeltaIndex(t, 's'))
ysigs = [pd.Series(y, index=pd.TimedeltaIndex(t, 's')) for y in ys]
xsigs = [x_tsig for y in ys]
xsigs = pd.DataFrame(np.array(xsigs).T, index=x_tsig.index)

delays, corrs, _, _ = compute_delays(xsigs, ysigs, compute_delays_modes)
delays, _ = compute_delays(xsigs, ysigs, compute_delays_modes)

assert np.allclose(delays, delay_ins)

0 comments on commit 9c46180

Please sign in to comment.