Add audio and text file functions (read_audio, write_audio, read_blockwise, write_txt, read_txt)

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

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',

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

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