diff --git a/docs/pyttadoc/functions.rst b/docs/pyttadoc/functions.rst index add57c0..4c84377 100644 --- a/docs/pyttadoc/functions.rst +++ b/docs/pyttadoc/functions.rst @@ -26,4 +26,8 @@ Functions .. autofunction:: pytta.plot_bars .. autofunction:: pytta.plot_spectrogram .. autofunction:: pytta.plot_waterfall - +.. autofunction:: pytta.read_audio +.. autofunction:: pytta.write_audio +.. autofunction:: pytta.read_blockwise +.. autofunction:: pytta.write_txt +.. autofunction:: pytta.read_txt diff --git a/pytta/__init__.py b/pytta/__init__.py index d8740aa..535b662 100644 --- a/pytta/__init__.py +++ b/pytta/__init__.py @@ -113,7 +113,8 @@ list_devices, print_devices, get_device_from_user,\ fft_convolve, find_delay, corr_coef, resample,\ peak_time, save, load, plot_time, plot_time_dB, plot_freq,\ - plot_bars, plot_spectrogram, plot_waterfall, SPL + plot_bars, plot_spectrogram, plot_waterfall, SPL,\ + read_audio, write_audio, read_blockwise, write_txt, read_txt # Must go on v0.1.0 (DEPRECATED) from .functions import fft_degree @@ -146,6 +147,11 @@ 'resample', 'corr_coef', 'peak_time', + 'read_audio', + 'write_audio', + 'read_blockwise', + 'write_txt', + 'read_txt', 'save', 'load', 'weighting', diff --git a/pytta/functions.py b/pytta/functions.py index 9b39239..16d6e83 100644 --- a/pytta/functions.py +++ b/pytta/functions.py @@ -23,6 +23,11 @@ >>> pytta.plot_bars(signalObj1, signalObj2, ..., signalObjN) >>> pytta.save(fileName, obj1, ..., objN) >>> pytta.load(fileName) + >>> pytta.read_audio(fileName) + >>> pytta.write_audio(fileName, signalObj) + >>> pytta.read_blockwise(fileName, blockSize) + >>> pytta.write_txt(fileName, signalObj) + >>> pytta.read_txt(fileName) For further information, check the function specific documentation. @@ -274,6 +279,98 @@ def peak_time(signal): else: return peaks_time[0] +def _soundfile(): + try: + import soundfile + except ImportError as err: # pragma: no cover + raise ImportError('read_audio, write_audio and read_blockwise need the ' + 'soundfile package (pip install soundfile).') from err + return soundfile + + +def read_audio(fileName, dtype='float64'): + """ + SignalObj from an audio file in any format that libsndfile reads (WAV, + FLAC, CAF, AU, AIFF, OGG, ...), as ita_read_flac, ita_read_caf and + ita_read_au; needs the soundfile package. + + >>> sig = pytta.read_audio('take.flac') + """ + sf = _soundfile() + data, fs = sf.read(fileName, dtype=dtype, always_2d=True) + return SignalObj(data, 'time', fs) + + +def write_audio(fileName, signal, subtype=None): + """ + signal to an audio file whose format follows the extension (WAV, FLAC, + CAF, AU, AIFF, ...), with the libsndfile subtype (e.g. 'PCM_24', + 'FLOAT'; the default of the format if None); needs soundfile. + + >>> pytta.write_audio('take.flac', sig, 'PCM_24') + """ + sf = _soundfile() + sf.write(fileName, np.asarray(signal.timeSignal, dtype=float), + int(signal.samplingRate), subtype=subtype) + + +def read_blockwise(fileName, block_size, overlap=0): + """ + Generator of SignalObj blocks of block_size samples from an audio file, + consecutive blocks sharing overlap samples, as ita_read_blockwise; the + last block is shorter when the file ends. Needs soundfile. + + >>> for block in pytta.read_blockwise('long.wav', 48000): ... + """ + sf = _soundfile() + fs = sf.info(fileName).samplerate + for block in sf.blocks(fileName, blocksize=block_size, overlap=overlap, + always_2d=True, dtype='float64'): + yield SignalObj(block, 'time', fs) + + +def write_txt(fileName, signal, domain='time', delimiter=' '): + """ + signal to a text file as ita_write_txt: the time (s) or frequency (Hz) + in the first column, then each channel, in the time domain its samples + and in the frequency domain the real and imaginary parts of its DFT + (numpy.fft.rfft of timeSignal). + + >>> pytta.write_txt('response.txt', ir, domain='freq') + """ + x = np.asarray(signal.timeSignal, dtype=float) + if domain == 'time': + cols = [np.arange(x.shape[0]) / signal.samplingRate] + [x[:, c] for c in range(x.shape[1])] + elif domain == 'freq': + X = np.fft.rfft(x, axis=0) + cols = [np.fft.rfftfreq(x.shape[0], 1 / signal.samplingRate)] + for c in range(x.shape[1]): + cols += [X[:, c].real, X[:, c].imag] + else: + raise ValueError("domain must be 'time' or 'freq'.") + np.savetxt(fileName, np.column_stack(cols), delimiter=delimiter, fmt='%.17g') + + +def read_txt(fileName, domain='time', samplingRate=None, numSamples=None, delimiter=None): + """ + SignalObj from a text file of write_txt or ita_write_txt. In the time + domain the sampling rate comes from the time column; in the frequency + domain it is the last frequency times 2 (numSamples even, 2 (bins - 1), + unless given). + """ + data = np.loadtxt(fileName, delimiter=delimiter, ndmin=2) + if domain == 'time': + fs = samplingRate or round(1 / (data[1, 0] - data[0, 0])) + return SignalObj(data[:, 1:], 'time', fs) + if domain != 'freq': + raise ValueError("domain must be 'time' or 'freq'.") + f = data[:, 0] + n = numSamples or 2 * (f.size - 1) + fs = samplingRate or round(f[1] * n) + X = data[:, 1::2] + 1j * data[:, 2::2] + return SignalObj(np.fft.irfft(X, n, axis=0), 'time', fs) + + def fft_degree(*args,**kwargs): """ DEPRECATED diff --git a/tests/test_audio_io.py b/tests/test_audio_io.py new file mode 100644 index 0000000..ef7e153 --- /dev/null +++ b/tests/test_audio_io.py @@ -0,0 +1,74 @@ +""" +Tests for the file functions of pytta.functions that follow the ITA-Toolbox +(ita_read_flac, ita_read_caf, ita_read_au, ita_read_blockwise, ita_write_txt). +Each test pins a round trip through a temporary file. +""" +import importlib.util +import os +import tempfile +import unittest + +import numpy as np + +os.environ.setdefault('MPLBACKEND', 'Agg') + +import pytta + +FS = 48000 +NEEDS_SOUNDFILE = unittest.skipUnless(importlib.util.find_spec('soundfile'), + 'the audio file functions need soundfile') + + +class TestAudioIO(unittest.TestCase): + + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + label = 'x' + t = np.arange(4800) / FS + self.x = np.column_stack([0.5 * np.sin(2 * np.pi * 440 * t), 0.25 * np.cos(2 * np.pi * 1000 * t)]) + self.sig = pytta.SignalObj(self.x, 'time', FS) + self.x32 = self.sig.timeSignal.astype(float) + + def tearDown(self): + self.tmp.cleanup() + + def path(self, name): + return os.path.join(self.tmp.name, name) + + @NEEDS_SOUNDFILE + def test_audio_formats_round_trip(self): + # FLAC at 24 bits keeps the samples within half a step; CAF and AU + # with float samples keep the float32 samples of SignalObj exactly. + for name, subtype, tol in (('a.flac', 'PCM_24', 2 ** -23), ('a.caf', 'FLOAT', 0), + ('a.au', 'FLOAT', 0), ('a.wav', 'DOUBLE', 0)): + pytta.write_audio(self.path(name), self.sig, subtype) + back = pytta.read_audio(self.path(name)) + e = np.abs(back.timeSignal - self.x32).max() + print(f'\n {name} {subtype}: {e:.1e}', end='') + self.assertEqual(back.samplingRate, FS) + self.assertLessEqual(e, tol) + + @NEEDS_SOUNDFILE + def test_blockwise_reading_rebuilds_the_file(self): + pytta.write_audio(self.path('b.wav'), self.sig, 'FLOAT') + blocks = [b.timeSignal for b in pytta.read_blockwise(self.path('b.wav'), 1000)] + sizes = [b.shape[0] for b in blocks] + self.assertEqual(sizes, [1000] * 4 + [800]) + np.testing.assert_array_equal(np.vstack(blocks), self.x32) + + def test_text_round_trip(self): + # Time samples and the real and imaginary parts of the DFT come + # back; the sampling rate is read from the first column. + for domain in ('time', 'freq'): + pytta.write_txt(self.path(f'{domain}.txt'), self.sig, domain) + back = pytta.read_txt(self.path(f'{domain}.txt'), domain) + e = np.abs(back.timeSignal - self.x32).max() + print(f'\n text, {domain} domain: {e:.1e}, fs {back.samplingRate}', end='') + self.assertEqual(back.samplingRate, FS) + self.assertLess(e, 1e-7) + with open(self.path('freq.txt')) as fh: + self.assertEqual(len(fh.readline().split()), 1 + 2 * 2) + + +if __name__ == '__main__': + unittest.main()