From 6f9c55e336536bc7a03f9a041f0fdbb074229a8d Mon Sep 17 00:00:00 2001 From: Sergio Aguirre Date: Sat, 31 Oct 2026 20:32:01 +0100 Subject: [PATCH] [FIX] find_delay and corr_coef, with an example of pytta.functions find_delay read signal1.N, which SignalObj does not have, so it always raised AttributeError; it also multiplied a half spectrum by a full one. It now takes the peak of scipy.signal.correlate of the first channels and returns the delay in samples, positive when the second signal comes later, for signals of any length. corr_coef passed (samples, 1) arrays to numpy.corrcoef, which takes each row as a variable: it built a matrix of twice the number of samples on each side (73 GB for one second at 48 kHz) and returned the correlation of two samples. It now correlates the first channels. examples/functions_module.py shows each function of the module. Co-Authored-By: Claude Fable 5.1 --- examples/functions_module.py | 70 +++++++++++++++++++++++++++++ pytta/functions.py | 29 +++++------- tests/test_delay_and_correlation.py | 37 +++++++++++++++ 3 files changed, 118 insertions(+), 18 deletions(-) create mode 100644 examples/functions_module.py create mode 100644 tests/test_delay_and_correlation.py diff --git a/examples/functions_module.py b/examples/functions_module.py new file mode 100644 index 0000000..899d169 --- /dev/null +++ b/examples/functions_module.py @@ -0,0 +1,70 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +""" +Examples of the functions of pytta.functions, which are also available as +pytta.. Run it as a script; the figures open at the end. +""" +import os +import tempfile + +import matplotlib.pyplot as plt + +import pytta + +if __name__ == "__main__": + + # Signals to work with: two sines and a sweep, at 48 kHz + sine1k = pytta.generate.sin(Arms=0.5, freq=1000, timeLength=1, + samplingRate=48000) + sine2k = pytta.generate.sin(Arms=0.25, freq=2000, timeLength=1, + samplingRate=48000) + sweep = pytta.generate.sweep(freqMin=50, freqMax=16000, fftDegree=16, + samplingRate=48000) + + # Audio devices available to measure with + print(pytta.list_devices()) + + # Several signals as channels of one SignalObj, and back + twoChannels = pytta.merge(sine1k, sine2k) + first, second = pytta.split(twoChannels) + print('channels:', twoChannels.numChannels, '->', first.numChannels, + 'and', second.numChannels) + + # Convolution, here of the sweep with a 10 ms delay of half amplitude + delayed = pytta.SignalObj([0.0] * 480 + [0.5], 'time', 48000) + echo = pytta.fft_convolve(sweep, delayed) + + # Delay between two signals, in samples, and their correlation + print('delay of the echo:', pytta.find_delay(sweep, echo), 'samples') + print('correlation of a sine with itself:', + pytta.corr_coef(sine1k, sine1k)) + + # Change of sampling rate + sine44 = pytta.resample(sine1k, 44100) + print('samples at 44.1 kHz:', sine44.numSamples) + + # Time of the peak of each signal + print('peak of the echo at', pytta.peak_time(echo), 's') + + # Sound pressure level in third octave bands, as an Analysis + level = pytta.SPL(sine1k, nthOct=3, minFreq=500, maxFreq=2000) + print(level) + + # Wave files and HDF5 files (PyTTa objects keep their metadata) + with tempfile.TemporaryDirectory() as folder: + wavName = os.path.join(folder, 'sine.wav') + pytta.write_wav(wavName, sine1k) + fromWav = pytta.read_wav(wavName) + h5Name = os.path.join(folder, 'signals.hdf5') + pytta.save(h5Name, sine1k, sweep) + loaded = pytta.load(h5Name) + print('read from the wave file:', fromWav.numSamples, 'samples;', + 'loaded from the HDF5 file:', list(loaded.keys())) + + # Figures + pytta.plot_time(sine1k, sine2k) + pytta.plot_time_dB(echo) + pytta.plot_freq(sweep, smooth=True) + pytta.plot_bars(level) + pytta.plot_spectrogram(sweep) + plt.show() diff --git a/pytta/functions.py b/pytta/functions.py index 9b39239..97bf20e 100644 --- a/pytta/functions.py +++ b/pytta/functions.py @@ -11,7 +11,7 @@ >>> pytta.read_wav(fileName) >>> pytta.write_wav(fileName, signalObject) >>> pytta.merge(signalObj1, signalObj2, ..., signalObjN) - >>> pytta.slipt(signalObj) + >>> pytta.split(signalObj) >>> pytta.fft_convolve(signalObj1, signalObj2) >>> pytta.find_delay(signalObj1, signalObj2) >>> pytta.corr_coef(signalObj1, signalObj2) @@ -221,29 +221,22 @@ def fft_convolve(signal1, signal2): def find_delay(signal1, signal2): """ - Cross Correlation alternative. + Delay of signal2 relative to signal1, in samples, from the peak of the + cross correlation of their first channels. Positive when signal2 comes + later; the signals may have different lengths. - More efficient fft based method to calculate time shift between two signals. - - >>> shift = pytta.find_delay(signal1,signal2) + >>> shift = pytta.find_delay(signal1, signal2) """ - if signal1.N != signal2.N: - return print('Signal1 and Signal2 must have the same length') - else: - freqSignal1 = signal1.freqSignal - freqSignal2 = sfft.fft(np.flipud(signal2.timeSignal)) - convoluted = np.real(sfft.ifft(freqSignal1 * freqSignal2)) - convShifted = sfft.fftshift(convoluted) - zeroIndex = int(signal1.numSamples / 2) - 1 - shift = zeroIndex - np.argmax(convShifted) - return shift + x, y = signal1.timeSignal[:, 0], signal2.timeSignal[:, 0] + lags = ss.correlation_lags(y.size, x.size) + return int(lags[np.argmax(ss.correlate(y, x, method='fft'))]) def corr_coef(signal1, signal2): - """Finds the correlation coefficient between two SignalObjs using the numpy.corrcoef() function.""" - coef = np.corrcoef(signal1.timeSignal, signal2.timeSignal) - return coef[0, 1] + """Correlation coefficient between the first channels of two SignalObjs + of the same length, by numpy.corrcoef.""" + return np.corrcoef(signal1.timeSignal[:, 0], signal2.timeSignal[:, 0])[0, 1] def resample(signal, newSamplingRate): diff --git a/tests/test_delay_and_correlation.py b/tests/test_delay_and_correlation.py new file mode 100644 index 0000000..5806db8 --- /dev/null +++ b/tests/test_delay_and_correlation.py @@ -0,0 +1,37 @@ +""" +Tests for pytta.find_delay and pytta.corr_coef. Each test pins one rule. +""" +import os +import unittest + +os.environ.setdefault('MPLBACKEND', 'Agg') + +import numpy as np + +import pytta + +FS = 48000 + + +class TestDelayAndCorrelation(unittest.TestCase): + + def test_delay_in_samples_of_the_second_signal(self): + # Positive when the second signal comes later; the signals may have + # different lengths. + x = np.random.default_rng(0).standard_normal(FS) + early = pytta.SignalObj(x, 'time', FS) + late = pytta.SignalObj(np.concatenate((np.zeros(100), x)), 'time', FS) + self.assertEqual(pytta.find_delay(early, late), 100) + self.assertEqual(pytta.find_delay(late, early), -100) + + def test_correlation_coefficient_of_two_signals(self): + # One coefficient for the first channels, not one per sample. + x = np.random.default_rng(1).standard_normal(FS) + a = pytta.SignalObj(x, 'time', FS) + b = pytta.SignalObj(-2 * x, 'time', FS) + self.assertAlmostEqual(pytta.corr_coef(a, a), 1, places=6) + self.assertAlmostEqual(pytta.corr_coef(a, b), -1, places=6) + + +if __name__ == '__main__': + unittest.main()