Estimate stats of the volume of an audio channel.
It extends the BaseVolume class to estimate and analyze the volume (rms) of an audio channel over specified time windows. It calculates rms values for segments of audio data and provides statistical measures for these values.
Main functionalities:
Create a volume estimator for the given channel.
Fields:
def __init__(self, channel: Channel, win_len: float=0.01):
"""Create a volume estimator for the given channel.
Fields:
- _channel: (Channel) The audio channel being analyzed.
- _win_len: (float) The window length used for estimating rms values.
- _volumes: (list) List of calculated rms values.
- _rms: (float) The global rms value of the channel.
:param channel: (Channel) The channel to work on.
:param win_len: (float) Window length to estimate the volume.
"""
super(ChannelVolume, self).__init__(win_len)
pos = channel.tell()
channel.rewind()
self._channel = channel
self._win_len = win_len
self._volumes = ChannelVolume.calculate_volumes(channel, win_len)
channel.seek(pos)
self._rms = channel.rms()
Set manually the rms at a given position.
def set_volume_value(self, index: int, value: float) -> None:
"""Set manually the rms at a given position.
:param index: (int) The index of the value to set.
:param value: (float) The value of the channel.
"""
self._volumes[index] = value
Force to re-estimate the global rms value and re-assign all members.
def evaluate(self, win_len: float=0.01) -> None:
"""Force to re-estimate the global rms value and re-assign all members.
:param win_len: (float) Window length to estimate the volume.
"""
pos = self._channel.tell()
self._channel.rewind()
'\n nb_frames = int(win_len * self._channel.get_framerate())\n nb_vols = int(self._channel.get_duration()/win_len) + 1\n previous_rms = 0\n for i in range(nb_vols):\n frames = self._channel.get_frames(nb_frames)\n a = AudioFrames(frames, self._channel.get_sampwidth(), 1)\n rms = a.rms()\n if rms > 0: # provide negative values of corrupted audio files\n self._volumes[i] = rms\n elif rms < 0:\n self._volumes[i] = previous_rms\n logging.warning("Corrupted audio? The RMS shouldn\'t be a negative value. Got {}.".format(rms))\n previous_rms = rms\n '
volumes = ChannelVolume.calculate_volumes(self._channel, self._win_len)
self._channel.seek(pos)
self._volumes = volumes
self._rms = self._channel.rms()
self._win_len = win_len
Estimate the volume values of an audio channel.
Computes the Root Mean Square (RMS) values of audio frames in a given channel over specified time windows. It handles corrupted audio by logging warnings and using the previous RMS value for negative RMS readings.
Flow:
>>> channel = Channel(framerate=16000, sampwidth=2, frames=b' ')
>>> volumes = ChannelVolume.calculate_volumes(channel, win_len=0.01)
>>> print(volumes)
> [2.74]
@staticmethod
def calculate_volumes(channel: Channel, win_len: float=0.01) -> list:
"""Estimate the volume values of an audio channel.
Computes the Root Mean Square (RMS) values of audio frames in a given
channel over specified time windows. It handles corrupted audio by
logging warnings and using the previous RMS value for negative RMS
readings.
Flow:
1. Calculate the number of frames per window and the total number of windows.
2. Initialize a list to store RMS values and set the initial previous_rms to 0.
3. Loop through each window, extract frames, and compute the RMS value.
4. If the RMS value is positive, store it; if negative, log a warning and use the previous RMS value.
5. Return the list of RMS values.
:example:
>>> channel = Channel(framerate=16000, sampwidth=2, frames=b'\x01\x00\x02\x00\x03\x00\x04\x00')
>>> volumes = ChannelVolume.calculate_volumes(channel, win_len=0.01)
>>> print(volumes)
[2.74]
:param channel: (Channel) The channel to work on.
:param win_len: (float) Window length to estimate the volume.
:return: (list) The volume values.
"""
nb_frames = int(win_len * channel.get_framerate())
nb_vols = int(channel.get_duration() / win_len) + 1
volumes = [0.0] * nb_vols
previous_rms = 0
for i in range(nb_vols):
frames = channel.get_frames(nb_frames)
a = AudioFrames(frames, channel.get_sampwidth(), 1)
rms = a.rms()
if rms > 0:
volumes[i] = rms
elif rms < 0:
volumes[i] = previous_rms
logging.warning("Corrupted audio? The RMS shouldn't be a negative value but got {}".format(rms))
previous_rms = rms
if len(volumes) > 0 and volumes[-1] == 0:
volumes.pop()
return volumes
Remove outliers from the volume statistics of the channel.
It sorts the volume values, determines a threshold at the 85th percentile, and caps any volume values above this threshold to the threshold value.
def normalize_volumes(self) -> int:
"""Remove outliers from the volume statistics of the channel.
It sorts the volume values, determines a threshold at the 85th percentile,
and caps any volume values above this threshold to the threshold value.
:return: (int) Number of modified outliers.
"""
if len(self._volumes) == 0:
return 0
nb = 0
sorted_volumes = sorted(self._volumes)
rms_threshold = sorted_volumes[int(0.85 * len(sorted_volumes))]
for i, v in enumerate(self._volumes):
if v > rms_threshold:
self._volumes[i] = rms_threshold
nb += 1
return nb
def __repr__(self):
return str(self._volumes)