"""
Read and write objects to disk, read and write audio files, read SOFA files.
The functions :py:func:`read` and :py:func:`write` allow to save or load
several pyfar objects and other variables. So, e.g., workspaces in notebooks
can be stored. :py:class:`Signal <pyfar.signal.Signal>` objects can be
imported and exported as WAV files using :py:func:`read_wav` and
:py:func:`write_wav`. :py:func:`read_sofa` provides functionality to read the
data stored in a SOFA file.
"""
import os.path
import pathlib
import warnings
import sofar as sf
import zipfile
import io
import tempfile
import numpy as np
try:
import soundfile
soundfile_imported = True
except ModuleNotFoundError or OSError:
soundfile_imported = False
soundfile_warning = (
"python-soundfile could not be imported. Try to install it using `pip "
"install soundfile`. If this not works search the documentation for "
"help: https://python-soundfile.readthedocs.io")
import pyfar
from pyfar import Signal, FrequencyData, Coordinates
from . import _codec as codec
import pyfar.classes.filter as fo
[docs]def read_sofa(filename, verify=True):
"""
Import a SOFA file as pyfar object.
Parameters
----------
filename : string, Path
Input SOFA file (cf. [#]_, [#]_).
verify : bool, optional
Verify if the data contained in the SOFA file agrees with the AES69
standard (see references). If the verification fails, the SOFA file
can be loaded by setting ``verify=False``. The default is ``True``
Returns
-------
audio : pyfar audio object
The audio object that is returned depends on the DataType of the SOFA
object:
- :py:class:`~pyfar.classes.audio.Signal`
A Signal object is returned is the DataType is ``'FIR'``,
``'FIR-E'``, or ``'FIRE'``.
- :py:class:`~pyfar.classes.audio.FrequencyData`
A FrequencyData object is returned is the DataType is ``'TF'``,
``'TF-E'``, or ``'TFE'``.
The `cshape` of the object is is ``(M, R)`` with `M` being the number
of measurements and `R` being the number of receivers from the SOFA
file.
source_coordinates : Coordinates
Coordinates object containing the data stored in
`SOFA_object.SourcePosition`. The domain, convention and unit are
automatically matched.
receiver_coordinates : Coordinates
Coordinates object containing the data stored in
`SOFA_object.RecevierPosition`. The domain, convention and unit are
automatically matched.
Notes
-----
* This function uses the sofar package to read SOFA files [#]_.
References
----------
.. [#] https://www.sofaconventions.org
.. [#] “AES69-2020: AES Standard for File Exchange-Spatial Acoustic Data
File Format.”, 2020.
.. [#] https://pyfar.org
"""
sofafile = sf.read_sofa(filename, verify)
# Check for DataType
if sofafile.GLOBAL_DataType in ['FIR', 'FIR-E', 'FIRE']:
# make a Signal
signal = Signal(sofafile.Data_IR, sofafile.Data_SamplingRate)
elif sofafile.GLOBAL_DataType in ['TF', 'TF-E', 'TFE']:
# make FrequencyData
signal = FrequencyData(
sofafile.Data_Real + 1j * sofafile.Data_Imag, sofafile.N)
else:
raise ValueError(
"DataType {sofafile.GLOBAL_DataType} is not supported.")
# Source
s_values = sofafile.SourcePosition
s_domain, s_convention, s_unit = _sofa_pos(sofafile.SourcePosition_Type)
source_coordinates = Coordinates(
s_values[:, 0],
s_values[:, 1],
s_values[:, 2],
domain=s_domain,
convention=s_convention,
unit=s_unit)
# Receiver
r_values = sofafile.ReceiverPosition
r_domain, r_convention, r_unit = _sofa_pos(sofafile.ReceiverPosition_Type)
receiver_coordinates = Coordinates(
r_values[:, 0],
r_values[:, 1],
r_values[:, 2],
domain=r_domain,
convention=r_convention,
unit=r_unit)
return signal, source_coordinates, receiver_coordinates
def _sofa_pos(pos_type):
if pos_type == 'spherical':
domain = 'sph'
convention = 'top_elev'
unit = 'deg'
elif pos_type == 'cartesian':
domain = 'cart'
convention = 'right'
unit = 'met'
else:
raise ValueError("Position:Type {pos_type} is not supported.")
return domain, convention, unit
[docs]def read(filename):
"""
Read any compatible pyfar object or numpy array (.far file) from disk.
Parameters
----------
filename : string, Path
Input file. If no extension is provided, .far-suffix is added.
Returns
-------
collection: dict
Contains pyfar objects like
``{ 'name1': 'obj1', 'name2': 'obj2' ... }``.
Examples
--------
Read signal and orientations objects stored in a .far file.
>>> collection = pyfar.read('my_objs.far')
>>> my_signal = collection['my_signal']
>>> my_orientations = collection['my_orientations']
"""
# Check for .far file extension
filename = pathlib.Path(filename).with_suffix('.far')
collection = {}
with open(filename, 'rb') as f:
zip_buffer = io.BytesIO()
zip_buffer.write(f.read())
with zipfile.ZipFile(zip_buffer) as zip_file:
zip_paths = zip_file.namelist()
obj_names_hints = [
path.split('/')[:2] for path in zip_paths if '/$' in path]
for name, hint in obj_names_hints:
if codec._is_pyfar_type(hint[1:]):
obj = codec._decode_object_json_aided(name, hint, zip_file)
elif hint == '$ndarray':
obj = codec._decode_ndarray(f'{name}/{hint}', zip_file)
else:
raise TypeError(
'.far-file contains unknown types.'
'This might occur when writing and reading files with'
'different versions of Pyfar.')
collection[name] = obj
if 'builtin_wrapper' in collection:
for key, value in collection['builtin_wrapper'].items():
collection[key] = value
collection.pop('builtin_wrapper')
return collection
[docs]def write(filename, compress=False, **objs):
"""
Write any compatible pyfar object or numpy array and often used builtin
types as .far file to disk.
Parameters
----------
filename : string
Full path or filename. If now extension is provided, .far-suffix
will be add to filename.
compress : bool
Default is ``False`` (uncompressed).
Compressed files take less disk space but need more time for writing
and reading.
**objs:
Objects to be saved as key-value arguments, e.g.,
``name1=object1, name2=object2``.
Examples
--------
Save Signal object, Orientations objects and numpy array to disk.
>>> s = pyfar.Signal([1, 2, 3], 44100)
>>> o = pyfar.Orientations.from_view_up([1, 0, 0], [0, 1, 0])
>>> a = np.array([1,2,3])
>>> pyfar.io.write('my_objs.far', signal=s, orientations=o, array=a)
Notes
-----
* Supported builtin types are:
bool, bytes, complex, float, frozenset, int, list, set, str and tuple
"""
# Check for .far file extension
filename = pathlib.Path(filename).with_suffix('.far')
compression = zipfile.ZIP_STORED if compress else zipfile.ZIP_DEFLATED
zip_buffer = io.BytesIO()
builtin_wrapper = codec.BuiltinsWrapper()
with zipfile.ZipFile(zip_buffer, "a", compression) as zip_file:
for name, obj in objs.items():
if codec._is_pyfar_type(obj):
codec._encode_object_json_aided(obj, name, zip_file)
elif codec._is_numpy_type(obj):
codec._encode({f'${type(obj).__name__}': obj}, name, zip_file)
elif type(obj) in codec._supported_builtin_types():
builtin_wrapper[name] = obj
else:
error = (
f'Objects of type {type(obj)} cannot be written to disk.')
if isinstance(obj, fo.Filter):
error = f'{error}. Consider casting to {fo.Filter}'
raise TypeError(error)
if len(builtin_wrapper) > 0:
codec._encode_object_json_aided(
builtin_wrapper, 'builtin_wrapper', zip_file)
with open(filename, 'wb') as f:
f.write(zip_buffer.getvalue())
[docs]def read_audio(filename, dtype='float64', **kwargs):
"""
Import an audio file as :py:class:`~pyfar.classes.audio.Signal` object.
Reads 'wav', 'aiff', 'ogg', and 'flac' files among others. For a complete
list see :py:func:`audio_formats`.
Parameters
----------
filename : string, Path
Input file.
dtype : {'float64', 'float32', 'int32', 'int16'}, optional
Data type of the returned signal, by default ``'float64'``.
Floating point audio data is typically in the range from
``-1.0`` to ``1.0``. Note that ``'int16'`` and ``'int32'`` should only
be used if the data was written in the same format. Integer data is in
the range from ``-2**15`` to ``2**15-1`` for ``'int16'`` and from
``-2**31`` to ``2**31-1`` for ``'int32'``.
**kwargs
Other keyword arguments to be passed to :py:func:`soundfile.read`. This
is needed, e.g, to read RAW audio files.
Returns
-------
signal : Signal
:py:class:`~pyfar.classes.audio.Signal` object containing the audio
data.
Notes
-----
* This function is based on :py:func:`soundfile.read`.
* Reading int values from a float file will *not* scale the data to
[-1.0, 1.0). If the file contains ``np.array([42.6], dtype='float32')``,
you will read ``np.array([43], dtype='int32')`` for ``dtype='int32'``.
"""
if not soundfile_imported:
warnings.warn(soundfile_warning)
return
data, sampling_rate = soundfile.read(
file=filename, dtype=dtype, always_2d=True, **kwargs)
signal = Signal(data.T, sampling_rate, domain='time', dtype=dtype)
return signal
[docs]def write_audio(signal, filename, subtype=None, overwrite=True, **kwargs):
"""
Write a :py:class:`~pyfar.classes.audio.Signal` object as a audio file to
disk.
Writes 'wav', 'aiff', 'ogg', and 'flac' files among others. For a complete
list see :py:func:`audio_formats`.
Parameters
----------
signal : Signal
Object to be written.
filename : string, Path
Output file. The format is determined from the file extension.
See :py:func:`audio_formats` for all possible formats.
subtype : str, optional
The subtype of the sound file, the default value depends on the
selected `format` (see :py:func:`default_audio_subtype`).
See :py:func:`audio_subtypes` for all possible subtypes for
a given ``format``.
overwrite : bool
Select wether to overwrite the audio file, if it already exists.
The default is ``True``.
**kwargs
Other keyword arguments to be passed to :py:func:`soundfile.write`.
Notes
-----
* Signals are flattened before writing to disk (e.g. a signal with
``cshape = (3, 2)`` will be written to disk as a six channel audio file).
* This function is based on :py:func:`soundfile.write`.
* Except for the subtypes ``'FLOAT'``, ``'DOUBLE'`` and ``'VORBIS'`` ´
amplitudes larger than +/- 1 are clipped.
"""
if not soundfile_imported:
warnings.warn(soundfile_warning)
return
sampling_rate = signal.sampling_rate
data = signal.time
# Reshape to 2D
data = data.reshape(-1, data.shape[-1])
if len(signal.cshape) != 1:
warnings.warn(f"Signal flattened to {data.shape[0]} channels.")
# Check if file exists and for overwrite
if overwrite is False and os.path.isfile(filename):
raise FileExistsError(
"File already exists,"
"use overwrite option to disable error.")
else:
# Only the subtypes FLOAT, DOUBLE, VORBIS are not clipped,
# see _clipped_audio_subtypes()
format = pathlib.Path(filename).suffix[1:]
if subtype is None:
subtype = default_audio_subtype(format)
if (np.any(data > 1.) and
subtype.upper() not in ['FLOAT', 'DOUBLE', 'VORBIS']):
warnings.warn(
f'{format}-files of subtype {subtype} are clipped to +/- 1.')
soundfile.write(
file=filename, data=data.T, samplerate=sampling_rate,
subtype=subtype, **kwargs)
[docs]def read_wav(filename):
"""
Import a WAV file as :py:class:`~pyfar.classes.audio.Signal` object.
Parameters
----------
filename : string, Path
Input file.
Returns
-------
signal : Signal
:py:class:`~pyfar.classes.audio.Signal` object containing the audio
data from the WAV file.
Notes
-----
* This function is based on :py:func:`read_audio`.
"""
warnings.warn(("This function will be deprecated in pyfar 0.5.0 in favor "
"of pyfar.io.read_audio."),
PendingDeprecationWarning)
signal = read_audio(filename)
return signal
[docs]def write_wav(signal, filename, subtype=None, overwrite=True):
"""
Write a :py:class:`~pyfar.classes.audio.Signal` object as a WAV file to
disk.
Parameters
----------
signal : Signal
Object to be written.
filename : string, Path
Output file.
overwrite : bool
Select wether to overwrite the WAV file, if it already exists.
The default is ``True``.
Notes
-----
* Signals are flattened before writing to disk (e.g. a signal with
``cshape = (3, 2)`` will be written to disk as a six channel wav file).
* This function is based on :py:func:`write_audio`.
* Except for the subtypes ``'FLOAT'`` and ``'DOUBLE'``,
amplitudes larger than +/- 1 are clipped.
"""
warnings.warn(("This function will be deprecated in pyfar 0.5.0 in favor "
"of pyfar.io.read_audio."),
PendingDeprecationWarning)
# .wav file extension
filename = pathlib.Path(filename).with_suffix('.wav')
write_audio(signal, filename, subtype=subtype, overwrite=overwrite)
[docs]def audio_subtypes(format=None):
"""Return a dictionary of available audio subtypes.
Parameters
----------
format : str
If given, only compatible subtypes are returned.
Notes
-----
This function is a wrapper of :py:func:`soundfile.available_subtypes()`.
Examples
--------
>>> import pyfar as pf
>>> pf.io.audio_subtypes('FLAC')
{'PCM_24': 'Signed 24 bit PCM',
'PCM_16': 'Signed 16 bit PCM',
'PCM_S8': 'Signed 8 bit PCM'}
"""
if not soundfile_imported:
warnings.warn(soundfile_warning)
return
return soundfile.available_subtypes(format=format)
[docs]def default_audio_subtype(format):
"""Return the default subtype for a given format.
Notes
-----
This function is a wrapper of :py:func:`soundfile.default_audio_subtype()`.
Examples
--------
>>> import pyfar as pf
>>> pf.io.default_audio_subtype('WAV')
'PCM_16'
>>> pf.io.default_audio_subtype('MAT5')
'DOUBLE'
"""
if not soundfile_imported:
warnings.warn(soundfile_warning)
return
return soundfile.default_subtype(format)
def _clipped_audio_subtypes():
"""Creates a dictionary of format/subtype combinations which are clipped by
:py:func:´write_audio`.
This function is not called directly due to the need of writing all files
to disk. It needs to be called manually:
pyfar.io.io._clipped_audio_subtypes().
"""
if not soundfile_imported:
warnings.warn(soundfile_warning)
return
collection = {}
signal = pyfar.Signal([-1.5, -1, -.5, 0, .5, 1, 1.5]*100, 44100)
with tempfile.TemporaryDirectory() as tmpdir:
formats = pyfar.io.audio_formats()
for format in formats:
filename = os.path.join(tmpdir, 'test_file.'+format)
for subtype in pyfar.io.audio_subtypes(format):
write_valid = not _soundfile_write_errors(format, subtype)
read_valid = not _soundfile_read_errors(format, subtype)
format_valid = soundfile.check_format(format, subtype)
if write_valid and read_valid and format_valid:
if format == 'RAW':
write_audio(signal, filename, subtype=subtype)
signal_read = read_audio(
filename, samplerate=44100, channels=1,
subtype=subtype)
else:
write_audio(signal, filename, subtype=subtype)
signal_read = read_audio(filename)
if (np.any(signal_read.time > 1.1) and
np.any(signal_read.time < -1.1)):
behavior = 'not clipping (' + format + ')'
elif (np.any(signal_read.time > .1) and
np.any(signal_read.time < -.1)):
behavior = 'clipping to +/- 1 (' + format + ')'
else:
raise ValueError(f"{format}/{subtype}")
if subtype not in collection:
collection[subtype] = [behavior]
else:
collection[subtype] = collection[subtype] + [behavior]
return collection
def _soundfile_write_errors(format, subtype):
"""Checks if a write error due to soundfile/libsnfile can be expected.
Written according to test_write_audio_read_audio.
"""
if format == 'AIFF' and subtype == 'DWVW_12':
error_expected = True
else:
error_expected = False
return error_expected
def _soundfile_read_errors(format, subtype):
"""Checks if a read error due to soundfile/libsnfile can be expected.
Written according to test_write_audio_read_audio.
"""
if 'DWVW' in subtype and (format == 'AIFF' or format == 'RAW'):
error_expected = True
else:
error_expected = False
return error_expected