"""
Class summary
=============
.. autosummary::
DataSource
StringDataSource
ADSFactory
ADSFactory.AudioDataSource
ADSFactory.ADSDecorator
ADSFactory.OverlapADS
ADSFactory.LimiterADS
ADSFactory.RecorderADS
DataValidator
AudioEnergyValidator
"""
from abc import ABCMeta, abstractmethod
import math
from array import array
from .io import Rewindable, from_file, BufferAudioSource, PyAudioSource
from .exceptions import DuplicateArgument
import sys
try:
import numpy
_WITH_NUMPY = True
except ImportError as e:
_WITH_NUMPY = False
try:
from builtins import str
basestring = str
except ImportError as e:
if sys.version_info >= (3, 0):
basestring = str
__all__ = ["DataSource", "DataValidator", "StringDataSource", "ADSFactory", "AudioEnergyValidator"]
[docs]class DataSource():
"""
Base class for objects passed to :func:`auditok.core.StreamTokenizer.tokenize`.
Subclasses should implement a :func:`DataSource.read` method.
"""
__metaclass__ = ABCMeta
[docs] @abstractmethod
def read(self):
"""
Read a piece of data read from this source.
If no more data is available, return None.
"""
[docs]class DataValidator():
"""
Base class for a validator object used by :class:`.core.StreamTokenizer` to check
if read data is valid.
Subclasses should implement :func:`is_valid` method.
"""
__metaclass__ = ABCMeta
[docs] @abstractmethod
def is_valid(self, data):
"""
Check whether `data` is valid
"""
[docs]class StringDataSource(DataSource):
"""
A class that represent a :class:`DataSource` as a string buffer.
Each call to :func:`DataSource.read` returns on character and moves one step forward.
If the end of the buffer is reached, :func:`read` returns None.
:Parameters:
`data` :
a basestring object.
"""
def __init__(self, data):
self._data = None
self._current = 0
self.set_data(data)
[docs] def read(self):
"""
Read one character from buffer.
:Returns:
Current character or None if end of buffer is reached
"""
if self._current >= len(self._data):
return None
self._current += 1
return self._data[self._current - 1]
[docs] def set_data(self, data):
"""
Set a new data buffer.
:Parameters:
`data` : a basestring object
New data buffer.
"""
if not isinstance(data, basestring):
raise ValueError("data must an instance of basestring")
self._data = data
self._current = 0
[docs]class ADSFactory:
"""
Factory class that makes it easy to create an :class:`ADSFactory.AudioDataSource` object that implements
:class:`DataSource` and can therefore be passed to :func:`auditok.core.StreamTokenizer.tokenize`.
Whether you read audio data from a file, the microphone or a memory buffer, this factory
instantiates and returns the right :class:`ADSFactory.AudioDataSource` object.
There are many other features you want your :class:`ADSFactory.AudioDataSource` object to have, such as:
memorize all read audio data so that you can rewind and reuse it (especially useful when
reading data from the microphone), read a fixed amount of data (also useful when reading
from the microphone), read overlapping audio frames (often needed when dosing a spectral
analysis of data).
:func:`ADSFactory.ads` automatically creates and return object with the desired behavior according
to the supplied keyword arguments.
"""
@staticmethod
def _check_normalize_args(kwargs):
for k in kwargs:
if not k in ["block_dur", "hop_dur", "block_size", "hop_size", "max_time", "record",
"audio_source", "filename", "data_buffer", "frames_per_buffer", "sampling_rate",
"sample_width", "channels", "sr", "sw", "ch", "asrc", "fn", "fpb", "db", "mt",
"rec", "bd", "hd", "bs", "hs"]:
raise ValueError("Invalid argument: {0}".format(k))
if "block_dur" in kwargs and "bd" in kwargs:
raise DuplicateArgument("Either 'block_dur' or 'bd' must be specified, not both")
if "hop_dur" in kwargs and "hd" in kwargs:
raise DuplicateArgument("Either 'hop_dur' or 'hd' must be specified, not both")
if "block_size" in kwargs and "bs" in kwargs:
raise DuplicateArgument("Either 'block_size' or 'bs' must be specified, not both")
if "hop_size" in kwargs and "hs" in kwargs:
raise DuplicateArgument("Either 'hop_size' or 'hs' must be specified, not both")
if "max_time" in kwargs and "mt" in kwargs:
raise DuplicateArgument("Either 'max_time' or 'mt' must be specified, not both")
if "audio_source" in kwargs and "asrc" in kwargs:
raise DuplicateArgument("Either 'audio_source' or 'asrc' must be specified, not both")
if "filename" in kwargs and "fn" in kwargs:
raise DuplicateArgument("Either 'filename' or 'fn' must be specified, not both")
if "data_buffer" in kwargs and "db" in kwargs:
raise DuplicateArgument("Either 'filename' or 'db' must be specified, not both")
if "frames_per_buffer" in kwargs and "fbb" in kwargs:
raise DuplicateArgument("Either 'frames_per_buffer' or 'fpb' must be specified, not both")
if "sampling_rate" in kwargs and "sr" in kwargs:
raise DuplicateArgument("Either 'sampling_rate' or 'sr' must be specified, not both")
if "sample_width" in kwargs and "sw" in kwargs:
raise DuplicateArgument("Either 'sample_width' or 'sw' must be specified, not both")
if "channels" in kwargs and "ch" in kwargs:
raise DuplicateArgument("Either 'channels' or 'ch' must be specified, not both")
if "record" in kwargs and "rec" in kwargs:
raise DuplicateArgument("Either 'record' or 'rec' must be specified, not both")
kwargs["bd"] = kwargs.pop("block_dur", None) or kwargs.pop("bd", None)
kwargs["hd"] = kwargs.pop("hop_dur", None) or kwargs.pop("hd", None)
kwargs["bs"] = kwargs.pop("block_size", None) or kwargs.pop("bs", None)
kwargs["hs"] = kwargs.pop("hop_size", None) or kwargs.pop("hs", None)
kwargs["mt"] = kwargs.pop("max_time", None) or kwargs.pop("mt", None)
kwargs["asrc"] = kwargs.pop("audio_source", None) or kwargs.pop("asrc", None)
kwargs["fn"] = kwargs.pop("filename", None) or kwargs.pop("fn", None)
kwargs["db"] = kwargs.pop("data_buffer", None) or kwargs.pop("db", None)
record = kwargs.pop("record", False)
if not record:
record = kwargs.pop("rec", False)
if not isinstance(record, bool):
raise TypeError("'record' must be a boolean")
kwargs["rec"] = record
# keep long names for arguments meant for BufferAudioSource and PyAudioSource
if "frames_per_buffer" in kwargs or "fpb" in kwargs:
kwargs["frames_per_buffer"] = kwargs.pop("frames_per_buffer", None) or kwargs.pop("fpb", None)
if "sampling_rate" in kwargs or "sr" in kwargs:
kwargs["sampling_rate"] = kwargs.pop("sampling_rate", None) or kwargs.pop("sr", None)
if "sample_width" in kwargs or "sw" in kwargs:
kwargs["sample_width"] = kwargs.pop("sample_width", None) or kwargs.pop("sw", None)
if "channels" in kwargs or "ch" in kwargs:
kwargs["channels"] = kwargs.pop("channels", None) or kwargs.pop("ch", None)
[docs] @staticmethod
def ads(**kwargs):
"""
Create an return an :class:`ADSFactory.AudioDataSource`. The type and behavior of the object is the result
of the supplied parameters.
:Parameters:
*No parameters* :
read audio data from the available built-in microphone with the default parameters.
The returned :class:`ADSFactory.AudioDataSource` encapsulate an :class:`io.PyAudioSource` object and hence
it accepts the next four parameters are passed to use instead of their default values.
`sampling_rate`, `sr` : *(int)*
number of samples per second. Default = 16000.
`sample_width`, `sw` : *(int)*
number of bytes per sample (must be in (1, 2, 4)). Default = 2
`channels`, `ch` : *(int)*
number of audio channels. Default = 1 (only this value is currently accepted)
`frames_per_buffer`, `fpb` : *(int)*
number of samples of PyAudio buffer. Default = 1024.
`audio_source`, `asrc` : an `AudioSource` object
read data from this audio source
`filename`, `fn` : *(string)*
build an `io.AudioSource` object using this file (currently only wave format is supported)
`data_buffer`, `db` : *(string)*
build an `io.BufferAudioSource` using data in `data_buffer`. If this keyword is used,
`sampling_rate`, `sample_width` and `channels` are passed to `io.BufferAudioSource`
constructor and used instead of default values.
`max_time`, `mt` : *(float)*
maximum time (in seconds) to read. Default behavior: read until there is no more data
available.
`record`, `rec` : *(bool)*
save all read data in cache. Provide a navigable object which boasts a `rewind` method.
Default = False.
`block_dur`, `bd` : *(float)*
processing block duration in seconds. This represents the quantity of audio data to return
each time the :func:`read` method is invoked. If `block_dur` is 0.025 (i.e. 25 ms) and the sampling
rate is 8000 and the sample width is 2 bytes, :func:`read` returns a buffer of 0.025 * 8000 * 2 = 400
bytes at most. This parameter will be looked for (and used if available) before `block_size`.
If neither parameter is given, `block_dur` will be set to 0.01 second (i.e. 10 ms)
`hop_dur`, `hd` : *(float)*
quantity of data to skip from current processing window. if `hop_dur` is supplied then there
will be an overlap of `block_dur` - `hop_dur` between two adjacent blocks. This
parameter will be looked for (and used if available) before `hop_size`. If neither parameter
is given, `hop_dur` will be set to `block_dur` which means that there will be no overlap
between two consecutively read blocks.
`block_size`, `bs` : *(int)*
number of samples to read each time the `read` method is called. Default: a block size
that represents a window of 10ms, so for a sampling rate of 16000, the default `block_size`
is 160 samples, for a rate of 44100, `block_size` = 441 samples, etc.
`hop_size`, `hs` : *(int)*
determines the number of overlapping samples between two adjacent read windows. For a
`hop_size` of value *N*, the overlap is `block_size` - *N*. Default : `hop_size` = `block_size`,
means that there is no overlap.
:Returns:
An AudioDataSource object that has the desired features.
:Exampels:
1. **Create an AudioDataSource that reads data from the microphone (requires Pyaudio) with default audio parameters:**
.. code:: python
from auditok import ADSFactory
ads = ADSFactory.ads()
ads.get_sampling_rate()
16000
ads.get_sample_width()
2
ads.get_channels()
1
2. **Create an AudioDataSource that reads data from the microphone with a sampling rate of 48KHz:**
.. code:: python
from auditok import ADSFactory
ads = ADSFactory.ads(sr=48000)
ads.get_sampling_rate()
48000
3. **Create an AudioDataSource that reads data from a wave file:**
.. code:: python
import auditok
from auditok import ADSFactory
ads = ADSFactory.ads(fn=auditok.dataset.was_der_mensch_saet_mono_44100_lead_trail_silence)
ads.get_sampling_rate()
44100
ads.get_sample_width()
2
ads.get_channels()
1
4. **Define size of read blocks as 20 ms**
.. code:: python
import auditok
from auditok import ADSFactory
'''
we know samling rate for previous file is 44100 samples/second
so 10 ms are equivalent to 441 samples and 20 ms to 882
'''
block_size = 882
ads = ADSFactory.ads(bs = 882, fn=auditok.dataset.was_der_mensch_saet_mono_44100_lead_trail_silence)
ads.open()
# read one block
data = ads.read()
ads.close()
len(data)
1764
assert len(data) == ads.get_sample_width() * block_size
5. **Define block size as a duration (use block_dur or bd):**
.. code:: python
import auditok
from auditok import ADSFactory
dur = 0.25 # second
ads = ADSFactory.ads(bd = dur, fn=auditok.dataset.was_der_mensch_saet_mono_44100_lead_trail_silence)
'''
we know samling rate for previous file is 44100 samples/second
for a block duration of 250 ms, block size should be 0.25 * 44100 = 11025
'''
ads.get_block_size()
11025
assert ads.get_block_size() == int(0.25 * 44100)
ads.open()
# read one block
data = ads.read()
ads.close()
len(data)
22050
assert len(data) == ads.get_sample_width() * ads.get_block_size()
6. **Read overlapping blocks (one of hope_size, hs, hop_dur or hd > 0):**
For better readability we'd better use :class:`auditok.io.BufferAudioSource` with a string buffer:
.. code:: python
import auditok
from auditok import ADSFactory
'''
we supply a data beffer instead of a file (keyword 'bata_buffer' or 'db')
sr : sampling rate = 16 samples/sec
sw : sample width = 1 byte
ch : channels = 1
'''
buffer = "abcdefghijklmnop" # 16 bytes = 1 second of data
bd = 0.250 # block duration = 250 ms = 4 bytes
hd = 0.125 # hop duration = 125 ms = 2 bytes
ads = ADSFactory.ads(db = "abcdefghijklmnop", bd = bd, hd = hd, sr = 16, sw = 1, ch = 1)
ads.open()
ads.read()
'abcd'
ads.read()
'cdef'
ads.read()
'efgh'
ads.read()
'ghij'
data = ads.read()
assert data == 'ijkl'
7. **Limit amount of read data (use max_time or mt):**
.. code:: python
'''
We know audio file is larger than 2.25 seconds
We want to read up to 2.25 seconds of audio data
'''
ads = ADSFactory.ads(mt = 2.25, fn=auditok.dataset.was_der_mensch_saet_mono_44100_lead_trail_silence)
ads.open()
data = []
while True:
d = ads.read()
if d is None:
break
data.append(d)
ads.close()
data = b''.join(data)
assert len(data) == int(ads.get_sampling_rate() * 2.25 * ads.get_sample_width() * ads.get_channels())
"""
# copy user's dicionary (shallow copy)
kwargs = kwargs.copy()
# check and normalize keyword arguments
ADSFactory._check_normalize_args(kwargs)
block_dur = kwargs.pop("bd")
hop_dur = kwargs.pop("hd")
block_size = kwargs.pop("bs")
hop_size = kwargs.pop("hs")
max_time = kwargs.pop("mt")
audio_source = kwargs.pop("asrc")
filename = kwargs.pop("fn")
data_buffer = kwargs.pop("db")
record = kwargs.pop("rec")
# Case 1: an audio source is supplied
if audio_source is not None:
if (filename, data_buffer) != (None, None):
raise Warning("You should provide one of 'audio_source', 'filename' or 'data_buffer'\
keyword parameters. 'audio_source' will be used")
# Case 2: a file name is supplied
elif filename is not None:
if data_buffer is not None:
raise Warning("You should provide one of 'filename' or 'data_buffer'\
keyword parameters. 'filename' will be used")
audio_source = from_file(filename)
# Case 3: a data_buffer is supplied
elif data_buffer is not None:
audio_source = BufferAudioSource(data_buffer = data_buffer, **kwargs)
# Case 4: try to access native audio input
else:
audio_source = PyAudioSource(**kwargs)
if block_dur is not None:
if block_size is not None:
raise DuplicateArgument("Either 'block_dur' or 'block_size' can be specified, not both")
else:
block_size = int(audio_source.get_sampling_rate() * block_dur)
elif block_size is None:
# Set default block_size to 10 ms
block_size = int(audio_source.get_sampling_rate() / 100)
# Instantiate base AudioDataSource
ads = ADSFactory.AudioDataSource(audio_source=audio_source, block_size=block_size)
# Limit data to be read
if max_time is not None:
ads = ADSFactory.LimiterADS(ads=ads, max_time=max_time)
# Record, rewind and reuse data
if record:
ads = ADSFactory.RecorderADS(ads=ads)
# Read overlapping blocks of data
if hop_dur is not None:
if hop_size is not None:
raise DuplicateArgument("Either 'hop_dur' or 'hop_size' can be specified, not both")
else:
hop_size = int(audio_source.get_sampling_rate() * hop_dur)
if hop_size is not None:
if hop_size <= 0 or hop_size > block_size:
raise ValueError("hop_size must be > 0 and <= block_size")
if hop_size < block_size:
ads = ADSFactory.OverlapADS(ads=ads, hop_size=hop_size)
return ads
[docs] class AudioDataSource(DataSource):
"""
Base class for AudioDataSource objects.
It inherits from DataSource and encapsulates an AudioSource object.
"""
def __init__(self, audio_source, block_size):
self.audio_source = audio_source
self.block_size = block_size
def get_block_size(self):
return self.block_size
def set_block_size(self, size):
self.block_size = size
def get_audio_source(self):
return self.audio_source
def set_audio_source(self, audio_source):
self.audio_source = audio_source
def open(self):
self.audio_source.open()
def close(self):
self.audio_source.close()
def is_open(self):
return self.audio_source.is_open()
def get_sampling_rate(self):
return self.audio_source.get_sampling_rate()
def get_sample_width(self):
return self.audio_source.get_sample_width()
def get_channels(self):
return self.audio_source.get_channels()
def rewind(self):
if isinstance(self.audio_source, Rewindable):
self.audio_source.rewind()
else:
raise Exception("Audio source is not rewindable")
def is_rewindable(self):
return isinstance(self.audio_source, Rewindable)
[docs] def read(self):
return self.audio_source.read(self.block_size)
[docs] class ADSDecorator(AudioDataSource):
"""
Base decorator class for AudioDataSource objects.
"""
__metaclass__ = ABCMeta
def __init__(self, ads):
self.ads = ads
self.get_block_size = self.ads.get_block_size
self.set_block_size = self.ads.set_block_size
self.get_audio_source = self.ads.get_audio_source
self.open = self.ads.open
self.close = self.ads.close
self.is_open = self.ads.is_open
self.get_sampling_rate = self.ads.get_sampling_rate
self.get_sample_width = self.ads.get_sample_width
self.get_channels = self.ads.get_channels
def is_rewindable(self):
return self.ads.is_rewindable
def rewind(self):
self.ads.rewind()
self._reinit()
def set_audio_source(self, audio_source):
self.ads.set_audio_source(audio_source)
self._reinit()
def open(self):
if not self.ads.is_open():
self.ads.open()
self._reinit()
@abstractmethod
def _reinit(self):
pass
[docs] class OverlapADS(ADSDecorator):
"""
A class for AudioDataSource objects that can read and return overlapping audio frames
"""
def __init__(self, ads, hop_size):
ADSFactory.ADSDecorator.__init__(self, ads)
if hop_size <= 0 or hop_size > self.get_block_size():
raise ValueError("hop_size must be either 'None' or \
between 1 and block_size (both inclusive)")
self.hop_size = hop_size
self._actual_block_size = self.get_block_size()
self._reinit()
def _get_block_size():
return self._actual_block_size
def _read_first_block(self):
# For the first call, we need an entire block of size 'block_size'
block = self.ads.read()
if block is None:
return None
# Keep a slice of data in cache and append it in the next call
if len(block) > self._hop_size_bytes:
self._cache = block[self._hop_size_bytes:]
# Up from the next call, we will use '_read_next_blocks'
# and we only read 'hop_size'
self.ads.set_block_size(self.hop_size)
self.read = self._read_next_blocks
return block
def _read_next_blocks(self):
block = self.ads.read()
if block is None:
return None
# Append block to cache data to ensure overlap
block = self._cache + block
# Keep a slice of data in cache only if we have a full length block
# if we don't that means that this is the last block
if len(block) == self._block_size_bytes:
self._cache = block[self._hop_size_bytes:]
else:
self._cache = None
return block
def _reinit(self):
self._cache = None
self.ads.set_block_size(self._actual_block_size)
self._hop_size_bytes = self.hop_size * \
self.get_sample_width() * \
self.get_channels()
self._block_size_bytes = self.get_block_size() * \
self.get_sample_width() * \
self.get_channels()
self.read = self._read_first_block
[docs] class LimiterADS(ADSDecorator):
"""
A class for AudioDataSource objects that can read a fixed amount of data.
This can be useful when reading data from the microphone or from large audio files.
"""
def __init__(self, ads, max_time):
ADSFactory.ADSDecorator.__init__(self, ads)
self.max_time = max_time
self._reinit()
[docs] def read(self):
if self._total_read_bytes >= self._max_read_bytes:
return None
block = self.ads.read()
if block is None:
return None
self._total_read_bytes += len(block)
if self._total_read_bytes >= self._max_read_bytes:
self.close()
return block
def _reinit(self):
self._max_read_bytes = int(self.max_time * self.get_sampling_rate()) * \
self.get_sample_width() * \
self.get_channels()
self._total_read_bytes = 0
[docs] class RecorderADS(ADSDecorator):
"""
A class for AudioDataSource objects that can record all audio data they read,
with a rewind facility.
"""
def __init__(self, ads):
ADSFactory.ADSDecorator.__init__(self, ads)
self._reinit()
def _read_and_rec(self):
# Read and save read data
block = self.ads.read()
if block is not None:
self._cache.append(block)
return block
def _read_simple(self):
# Read without recording
return self.ads.read()
def rewind(self):
if self._record:
# If has been recording, create a new BufferAudioSource
# from recorded data
dbuffer = self._concatenate(self._cache)
asource = BufferAudioSource(dbuffer, self.get_sampling_rate(),
self.get_sample_width(),
self.get_channels())
self.set_audio_source(asource)
self.open()
self._cache = []
self._record = False
self.read = self._read_simple
else:
self.ads.rewind()
if not self.is_open():
self.open()
def is_rewindable(self):
return True
def _reinit(self):
# when audio_source is replaced, start recording again
self._record = True
self._cache = []
self.read = self._read_and_rec
def _concatenate(self, data):
try:
# should always work for python 2
# work for python 3 ONLY if data is a list (or an iterator)
# whose each element is a 'bytes' objects
return b''.join(data)
except TypeError:
# work for 'str' in python 2 and python 3
return ''.join(data)
[docs]class AudioEnergyValidator(DataValidator):
"""
The most basic auditok audio frame validator.
This validator computes the log energy of an input audio frame
and return True if the result is >= a given threshold, False
otherwise.
:Parameters:
`sample_width` : *(int)*
Number of bytes of one audio sample. This is used to convert data from `basestring` or `Bytes` to
an array of floats.
`energy_threshold` : *(float)*
A threshold used to check whether an input data buffer is valid.
"""
if _WITH_NUMPY:
_formats = {1: numpy.int8 , 2: numpy.int16, 4: numpy.int32}
@staticmethod
def _convert(signal, sample_width):
return numpy.array(numpy.frombuffer(signal, dtype=AudioEnergyValidator._formats[sample_width]), dtype=numpy.float64)
@staticmethod
def _signal_energy(signal):
return float(numpy.dot(signal, signal)) / len(signal)
@staticmethod
def _signal_log_energy(signal):
energy = AudioEnergyValidator._signal_energy(signal)
if energy <= 0:
return -200
return 10. * numpy.log10(energy)
else:
_formats = {1: 'b' , 2: 'h', 4: 'i'}
@staticmethod
def _convert(signal, sample_width):
return array("d", array(AudioEnergyValidator._formats[sample_width], signal))
@staticmethod
def _signal_energy(signal):
energy = 0.
for a in signal:
energy += a * a
return energy / len(signal)
@staticmethod
def _signal_log_energy(signal):
energy = AudioEnergyValidator._signal_energy(signal)
if energy <= 0:
return -200
return 10. * math.log10(energy)
def __init__(self, sample_width, energy_threshold=45):
self.sample_width = sample_width
self._energy_threshold = energy_threshold
[docs] def is_valid(self, data):
"""
Check if data is valid. Audio data will be converted into an array (of
signed values) of which the log energy is computed. Log energy is computed
as follows:
.. code:: python
arr = AudioEnergyValidator._convert(signal, sample_width)
energy = float(numpy.dot(arr, arr)) / len(arr)
log_energy = 10. * numpy.log10(energy)
:Parameters:
`data` : either a *string* or a *Bytes* buffer
`data` is converted into a numerical array using the `sample_width`
given in the constructor.
:Retruns:
True if `log_energy` >= `energy_threshold`, False otherwise.
"""
signal = AudioEnergyValidator._convert(data, self.sample_width)
return AudioEnergyValidator._signal_log_energy(signal) >= self._energy_threshold
def get_energy_threshold(self):
return self._energy_threshold
def set_energy_threshold(self, threshold):
self._energy_threshold = threshold