Public functions
save
Save the audio frames to a file.
Parameters
- filename: (str) output filename
- framerate: (int) frames per second
View Source
def save(self, filename: str, framerate: int):
"""Save the audio frames to a file.
:param filename: (str) output filename
:param framerate: (int) frames per second
"""
f = wave.Wave_write(filename)
f.setnchannels(self.get_nchannels())
f.setsampwidth(self.get_sampwidth())
f.setframerate(framerate)
try:
f.writeframes(self._frames)
finally:
f.close()
get_minval
Return the min value for a given sampwidth.
Parameters
- size: (int) the sampwidth
- signed: (bool) if the values will be signed or not
Raises
- SampleWidthError: Invalid given size.
Returns
View Source
@staticmethod
def get_minval(size, signed=True):
"""Return the min value for a given sampwidth.
:param size: (int) the sampwidth
:param signed: (bool) if the values will be signed or not
:raise: SampleWidthError: Invalid given size.
:return: (int) the min value
"""
if not signed:
return 0
elif size == 1:
return -128
elif size == 2:
return -32768
elif size == 4:
return -2147483648
raise SampleWidthError(size)
get_maxval
Return the max value for a given sampwidth.
Parameters
- size: (int) the sampwidth
- signed: (bool) if the values will be signed or not
Raises
- SampleWidthError: Invalid given size.
Returns
View Source
@staticmethod
def get_maxval(size, signed=True):
"""Return the max value for a given sampwidth.
:param size: (int) the sampwidth
:param signed: (bool) if the values will be signed or not
:raise: SampleWidthError: Invalid given size.
:return: (int) the max value
"""
if signed and size == 1:
return 127
elif size == 1:
return 255
elif signed and size == 2:
return 32767
elif size == 2:
return 65535
elif signed and size == 4:
return 2147483647
elif size == 4:
return 4294967295
raise SampleWidthError(size)
get_nchannels
Return the number of channels in the frames.
View Source
def get_nchannels(self):
"""Return the number of channels in the frames."""
return self._nchannels
get_sampwidth
Return the size of the frames in (1, 2, 4).
View Source
def get_sampwidth(self):
"""Return the size of the frames in (1, 2, 4)."""
return self._sampwidth
get_sample
Return the value of given sample index.
Parameters
- i: (int) index of the sample to get value
Returns
Raises
- IndexError: Invalid sample index.
View Source
def get_sample(self, i):
"""Return the value of given sample index.
:param i: (int) index of the sample to get value
:return: (int) value
:raises: IndexError: Invalid sample index.
"""
start = i * self._sampwidth
end = start + self._sampwidth
if end > len(self._frames):
raise IndexError('Sample index out of range')
return self.__frame_to_sample(self._frames[start:end])
minmax
Return the minimum and maximum values of all samples in the frames.
Returns
- (int, int) Min and max amplitude values, or (0,0) if empty frames.
View Source
def minmax(self):
"""Return the minimum and maximum values of all samples in the frames.
:return: (int, int) Min and max amplitude values, or (0,0) if empty frames.
"""
if len(self._frames) > 0:
val_min = self.get_maxval(self._sampwidth)
val_max = self.get_minval(self._sampwidth)
for i in range(len(self._frames) // self._sampwidth):
val = self.get_sample(i)
if val > val_max:
val_max = val
if val < val_min:
val_min = val
return (val_min, val_max)
return (0, 0)
min
Return the minimum of the values of all samples in the frames.
View Source
def min(self):
"""Return the minimum of the values of all samples in the frames."""
return self.minmax()[0]
max
Return the maximum of the values of all samples in the frames.
View Source
def max(self):
"""Return the maximum of the values of all samples in the frames."""
return self.minmax()[1]
absmax
Return the maximum of the *absolute value of all samples in the frames.*
View Source
def absmax(self):
"""Return the maximum of the *absolute value* of all samples in the frames."""
val_min, val_max = self.minmax()
return max(abs(val_min), abs(val_max))
avg
Return the average over all samples in the frames.
Returns
- (float) Average value rounded to 2 digits.
View Source
def avg(self):
"""Return the average over all samples in the frames.
:return: (float) Average value rounded to 2 digits.
"""
if len(self._frames) == 0:
return 0
samples_sum = 0.0
nb_samples = len(self._frames) / self._sampwidth
for i in range(int(nb_samples)):
samples_sum += self.get_sample(i)
return round(samples_sum / nb_samples, 2)
rms
Return the root-mean-square of the frames.
Returns
- (float) sqrt(sum(S_i^2) / n) rounded to 2 digits
View Source
def rms(self):
"""Return the root-mean-square of the frames.
:return: (float) sqrt(sum(S_i^2) / n) rounded to 2 digits
"""
if len(self._frames) == 0:
return 0.0
square_sum = 0.0
nb_samples = len(self._frames) / self._sampwidth
for i in range(int(nb_samples)):
val = self.get_sample(i)
square_sum += val * val
return round(math.sqrt(square_sum / nb_samples), 2)
cross
Return the number of zero crossings in frames.
Returns
- (int) Number of zero crossing or -1 if empty frames
View Source
def cross(self):
"""Return the number of zero crossings in frames.
:return: (int) Number of zero crossing or -1 if empty frames
"""
n_cross = -1
prev_val = 17
for i in range(len(self._frames) // self._sampwidth):
val = self.get_sample(i) < 0
if val != prev_val:
n_cross += 1
prev_val = val
return n_cross
clip
Return the number of frames with a value higher than the one of the factor.
Parameters
- factor: (float) All frames outside the interval are clipped.
Returns
View Source
def clip(self, factor=0.5):
"""Return the number of frames with a value higher than the one of the factor.
:param factor: (float) All frames outside the interval are clipped.
:return: (int)
"""
max_val = int(AudioFrames.get_maxval(self._sampwidth) * float(factor))
min_val = int(AudioFrames.get_minval(self._sampwidth) * float(factor))
n_clip = 0
for i in range(len(self._frames) // self._sampwidth):
val = self.get_sample(i)
if val >= max_val or val <= min_val:
n_clip += 1
return n_clip
clipping_rate
Return the clipping rate of the frames.
Percentage of samples with a value higher than the one corresponding
to the factor. Factor has to be between 0 and 1.
Parameters
- factor: (float) All frames outside the interval are clipped.
Returns
- (float) clipping rate value. To be multiplied by 100 to get the percentage!
View Source
def clipping_rate(self, factor=0.5):
"""Return the clipping rate of the frames.
Percentage of samples with a value higher than the one corresponding
to the factor. Factor has to be between 0 and 1.
:param factor: (float) All frames outside the interval are clipped.
:return: (float) clipping rate value. To be multiplied by 100 to get the percentage!
"""
if len(self._frames) == 0:
return 0.0
return float(self.clip(factor)) / (len(self._frames) // self._sampwidth)
get_frames
Return the stored frames.
View Source
def get_frames(self):
"""Return the stored frames."""
return self._frames
byteswap
Converts big-endian samples to little-endian and vice versa.
View Source
def byteswap(self):
"""Converts big-endian samples to little-endian and vice versa."""
try:
import audioop
audioop.byteswap(self._frames, self._sampwidth)
except ModuleNotFoundError:
raise NotImplementedError
findfactor
Return a factor F such that rms(add(mul(reference, -F))) is minimal.
Estimates the factor with which you should multiply reference to make
it match as well as possible to the frames.
Parameters
View Source
def findfactor(self, reference):
"""Return a factor F such that rms(add(mul(reference, -F))) is minimal.
Estimates the factor with which you should multiply reference to make
it match as well as possible to the frames.
"""
try:
import audioop
audioop.findfactor(self._frames, reference)
except ModuleNotFoundError:
raise NotImplementedError
reverse
Reverse the samples in the frames.
View Source
def reverse(self):
"""Reverse the samples in the frames."""
try:
import audioop
audioop.reverse(self._frames, self._sampwidth)
except ModuleNotFoundError:
raise NotImplementedError
mul
Return frames for which all samples are multiplied by factor.
All samples in the original frames are multiplied by the floating-point
value factor. Samples are truncated in case of overflow.
Parameters
- factor: (float) the factor which will be applied to each sample.
Returns
View Source
def mul(self, factor):
"""Return frames for which all samples are multiplied by factor.
All samples in the original frames are multiplied by the floating-point
value factor. Samples are truncated in case of overflow.
:param factor: (float) the factor which will be applied to each sample.
:return: (bytes) converted frames
"""
if len(self._frames) == 0:
return b''
factor = float(factor)
val_min = self.get_minval(self._sampwidth)
val_max = self.get_maxval(self._sampwidth)
mul_frames = self.__zero_frames()
for i in range(len(self._frames) // self._sampwidth):
val = float(self.get_sample(i))
mul_val = max(val_min, min(val_max, int(val * factor)))
mul_frames[i] = self.__sample_to_frame(mul_val)
return b''.join(mul_frames)
bias
Return frames that is the original ones with a bias added to each sample.
All samples in the original frames are summed with the given value.
Samples are truncated in case of overflow.
Parameters
- value: (int) the bias which will be applied to each sample.
Returns
View Source
def bias(self, value):
"""Return frames that is the original ones with a bias added to each sample.
All samples in the original frames are summed with the given value.
Samples are truncated in case of overflow.
:param value: (int) the bias which will be applied to each sample.
:return: (bytes) converted frames
"""
if len(self._frames) == 0:
return b''
value = int(value)
val_min = self.get_minval(self._sampwidth)
val_max = self.get_maxval(self._sampwidth)
mul_frames = self.__zero_frames()
for i in range(len(self._frames) // self._sampwidth):
val = self.get_sample(i)
mul_val = max(val_min, min(val_max, val + value))
mul_frames[i] = self.__sample_to_frame(mul_val)
return b''.join(mul_frames)
change_sampwidth
Return frames with the given number of bytes.
Parameters
- size: (int) new sample width of the frames. (1 for 8 bits, 2 for 16 bits, 4 for 32 bits)
Returns
View Source
def change_sampwidth(self, size):
"""Return frames with the given number of bytes.
:param size: (int) new sample width of the frames.
(1 for 8 bits, 2 for 16 bits, 4 for 32 bits)
:return: (bytes) converted frames
"""
if size not in (1, 2, 4):
raise SampleWidthError(size)
zero_byte = b'\x00' * size
size_frames = [zero_byte] * (len(self._frames) // self._sampwidth)
for i in range(len(self._frames) // self._sampwidth):
val = self.get_sample(i)
val = val << (4 - self._sampwidth) * 8
if size == 2:
size_frames[i] = struct.pack('<h', val >> 16)
elif size == 4:
size_frames[i] = struct.pack('<l', val)
elif size == 1:
size_frames[i] = struct.pack('<b', val >> 24)
return b''.join(size_frames)
resample
Return re-sampled frames with the given new framerate.
Parameters
- in_rate: (int) The original sample rate of the audio frames
- out_rate: (int) The desired sample rate to resample the audio frames to
Returns
- (bytes) the resampled audio frames.
Raises
- FrameRateError: invalid given in_rate
- FrameRateError: invalid given out_rate
- NotImplementedError: can't resample from inrate to outrate
View Source
def resample(self, in_rate, out_rate=16000):
"""Return re-sampled frames with the given new framerate.
:param in_rate: (int) The original sample rate of the audio frames
:param out_rate: (int) The desired sample rate to resample the audio frames to
:return: (bytes) the resampled audio frames.
:raise: FrameRateError: invalid given in_rate
:raise: FrameRateError: invalid given out_rate
:raise: NotImplementedError: can't resample from in_rate to out_rate
"""
if len(self._frames) == 0:
return b''
in_rate = int(in_rate)
out_rate = int(out_rate)
if in_rate < 8000:
raise FrameRateError(in_rate)
if out_rate < 8000:
raise FrameRateError(out_rate)
if in_rate == out_rate:
return self._frames
try:
frames = self._re_sample(in_rate, out_rate)
return frames
except NotImplementedError as e:
logging.warning("Deprecated. Since 'audioop' is removed of the Python standard library, re-sampling is not fully available from Python 3.13. ")
try:
import audioop
return audioop.ratecv(self._frames, self._sampwidth, self._nchannels, in_rate, out_rate, None)[0]
except ModuleNotFoundError:
raise e
Private functions
_re_sample
Do re-sampling, or not if in a not implemented condition.
Parameters
- in_rate: (int) The original sample rate of the audio frames
- out_rate: (int) The desired sample rate to resample the audio frames to
Returns
- (bytes) the resampled audio frames.
View Source
def _re_sample(self, in_rate, out_rate):
"""Do re-sampling, or not if in a not implemented condition.
:param in_rate: (int) The original sample rate of the audio frames
:param out_rate: (int) The desired sample rate to resample the audio frames to
:return: (bytes) the resampled audio frames.
"""
in_nsamples = len(self._frames) // self._sampwidth
out_nsamples = math.ceil(float(len(self._frames)) / float(self._sampwidth) * float(out_rate) / float(in_rate))
if in_nsamples == 0:
return b''
if AudioFrames._re_sample_valid_16kHz_rates(in_rate, out_rate) is False:
frames_out = self._ratecv(in_rate, out_rate, in_nsamples, out_nsamples)
else:
frames_out = self._re_sample_for_down16kHz(in_rate, out_rate, in_nsamples, out_nsamples)
return b''.join(frames_out)
_ratecv
Interpolate input samples to produce output samples.
This method is a python implementation of the audioop.ratecv function.
This function adjusts the rate of audio samples by interpolating the samples
from an input number of samples (innsamples) to an output number of samples
(outnsamples). It also performs a simple digital filter using the weights
weightA
and weightB
.
Known bug: the first frame is not properly interpolated.
Parameters
- in_nsamples: (int) The total number of input samples
- out_nsamples: (int) The desired number of output samples
Returns
- (list of bytes) Resampled audio frames
View Source
def _ratecv(self, in_rate, out_rate, in_nsamples, out_nsamples, state=None, weightA=1, weightB=0):
"""Interpolate input samples to produce output samples.
This method is a python implementation of the audioop.ratecv function.
This function adjusts the rate of audio samples by interpolating the samples
from an input number of samples (in_nsamples) to an output number of samples
(out_nsamples). It also performs a simple digital filter using the weights
`weightA` and `weightB`.
Known bug: the first frame is not properly interpolated.
:param in_nsamples: (int) The total number of input samples
:param out_nsamples: (int) The desired number of output samples
:return: (list of bytes) Resampled audio frames
"""
if in_rate * out_rate * in_nsamples * out_nsamples == 0:
raise ValueError('Unexpected zero rate.')
if weightA < 1 or weightB < 0:
raise ValueError('weightA should be >= 1, weightB should be >= 0')
gcd_rates = gcd(in_rate, out_rate)
in_n = in_rate // gcd_rates
in_nsamples = min(in_n, in_nsamples)
out_n = out_rate // gcd_rates
out_nsamples = min(out_n, out_nsamples)
bytes_per_frame = self._sampwidth * self._nchannels
if len(self._frames) % bytes_per_frame != 0:
raise ValueError('Fragment is not a whole number of frames')
num_frames = len(self._frames) // bytes_per_frame
if state is None:
prev_i = [0] * self._nchannels
cur_i = [0] * self._nchannels
else:
if not isinstance(state, tuple):
raise TypeError('State must be a tuple or None')
if len(state) != self._nchannels:
raise ValueError('State tuple size must match number of channels')
prev_i, cur_i = zip(*state)
prev_i = list(prev_i)
cur_i = list(cur_i)
output_length = math.ceil(num_frames * out_nsamples / in_nsamples)
zero_byte = b'\x00' * self._sampwidth
output_frames = [zero_byte] * output_length
input_idx = 0
output_idx = 0
d_accumulator = -in_nsamples
while input_idx < num_frames:
while d_accumulator < 0:
for chan in range(self._nchannels):
prev_i[chan] = cur_i[chan]
first_frame = input_idx * bytes_per_frame
last_frame = input_idx * bytes_per_frame + (chan + 1) * self._sampwidth
frame = self._frames[first_frame:last_frame]
try:
if len(frame) > 0:
cur_i[chan] = self.__frame_to_sample(frame)
cur_i[chan] = (weightA * cur_i[chan] + weightB * prev_i[chan]) // (weightA + weightB)
except:
pass
input_idx += 1
d_accumulator += out_nsamples
while d_accumulator >= 0:
for chan in range(self._nchannels):
val = round((prev_i[chan] * d_accumulator + cur_i[chan] * (out_nsamples - d_accumulator)) / out_nsamples)
try:
output_frames[output_idx] = self.__sample_to_frame(val)
except IndexError:
pass
output_idx += 1
d_accumulator -= in_n
return output_frames
_re_sample_valid_16kHz_rates
Check if given frame rates are supported for simple re-sampling.
Returns
- (bool) False if interpolation needed
Parameters
View Source
@staticmethod
def _re_sample_valid_16kHz_rates(in_rate, out_rate):
"""Check if given frame rates are supported for simple re-sampling.
:return: (bool) False if interpolation needed
"""
if in_rate < out_rate:
return False
if out_rate != 16000:
return False
if in_rate not in (192000, 96000, 48000, 32000, 16000):
return False
return True
_re_sample_for_down16kHz
Do re-sampling to 16kHz from a multiple.
Get actual frames, turn into samples, and reduce into the new expected
number of samples. Implemented solution algorithm is:
- add "in_n" samples into a buffer from frames;
- reduce the buffer to "out_n" samples;
- Go to 1. until the end of the frames is reached.
Parameters
- in_rate: (int) The original sample rate of the audio frames
- out_rate: (int) The desired sample rate to resample the audio frames to (expected 16kHz)
- in_nsamples: (int) The number of samples required for re-sampling
- out_nsamples: (int) The number of re-sampled samples
Returns
- (bytes) the resampled audio frames.
View Source
def _re_sample_for_down16kHz(self, in_rate, out_rate, in_nsamples, out_nsamples):
"""Do re-sampling to 16kHz from a multiple.
Get actual frames, turn into samples, and reduce into the new expected
number of samples. Implemented solution algorithm is:
1. add "in_n" samples into a buffer from frames;
2. reduce the buffer to "out_n" samples;
3. Go to 1. until the end of the frames is reached.
:param in_rate: (int) The original sample rate of the audio frames
:param out_rate: (int) The desired sample rate to resample the audio frames to (expected 16kHz)
:param in_nsamples: (int) The number of samples required for re-sampling
:param out_nsamples: (int) The number of re-sampled samples
:return: (bytes) the resampled audio frames.
"""
gcd_rates = gcd(in_rate, out_rate)
in_n = in_rate // gcd_rates
in_n = min(in_n, in_nsamples)
out_n = out_rate // gcd_rates
out_n = min(out_n, out_nsamples)
zero_byte = b'\x00' * self._sampwidth
size_frames = [zero_byte] * out_nsamples
in_buffer = [0] * in_n
cur_n = 0
i = 0
while i < in_nsamples:
in_buffer[cur_n] = self.get_sample(i)
cur_n = cur_n + 1
i = i + 1
if cur_n == in_n or i == in_nsamples:
if cur_n == in_n and in_n % out_n == 0:
out_buffer = self._down_sample(in_buffer, out_n)
else:
while in_n % out_n != 0:
in_n += 1
in_buffer.append(in_buffer[cur_n - 1])
while cur_n < in_n:
in_buffer[cur_n] = in_buffer[cur_n - 1]
cur_n = cur_n + 1
i = i + 1
out_buffer = self._down_sample(in_buffer, out_n)
for x in range(len(out_buffer)):
val = out_buffer[x]
size_frames[i // in_n - 1 + x] = self.__sample_to_frame(val)
cur_n = 0
return size_frames
_down_sample
Reduce the given samples to n items.
Down-sample is not applicable if not len(samples) % n.
Parameters
- samples: (list) List of sample values
- n: number of expected sample values in the returned list
Returns
- (list) reduced list of samples
View Source
@staticmethod
def _down_sample(samples, n):
"""Reduce the given samples to n items.
Down-sample is not applicable if not len(samples) % n.
:param samples: (list) List of sample values
:param n: number of expected sample values in the returned list
:return: (list) reduced list of samples
"""
if len(samples) % n == 0:
down = [0] * n
step = len(samples) // n
buffer = list()
cur = 0
for i in range(len(samples) + 1):
if len(buffer) == step:
down[cur] = int(sum(buffer) / len(buffer))
buffer = list()
cur = cur + 1
if i < len(samples):
buffer.append(samples[i])
else:
raise NotImplementedError('Down-sample is not applicable {:d}%{:d} != 0.'.format(len(samples), n))
return down