[FIX] find_delay and corr_coef, with an example of pytta.functions

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.<function>. 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()

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):

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()