# Copyright 2016 The Chromium OS Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
"""This module provides utilities to detect some artifacts and measure the
quality of audio."""
import logging
import math
import numpy
from autotest_lib.client.cros.audio import audio_analysis
# The input signal should be one sine wave with fixed frequency which
# can have silence before and/or after sine wave.
# For example:
# silence sine wave silence
# -----------|VVVVVVVVVVVVV|-----------
# (a) (b) (c)
# This module detects these artifacts:
# 1. Detect noise in (a) and (c).
# 2. Detect delay in (b).
# 3. Detect burst in (b).
# Assume the transitions between (a)(b) and (b)(c) are smooth and
# amplitude increases/decreases linearly.
# This module will detect artifacts in the sine wave.
# This module also estimates the equivalent noise level by teager operator.
# This module also detects volume changes in the sine wave. However, volume
# changes may be affected by delay or burst.
# Some artifacts may cause each other.
# In this module, amplitude and frequency are derived from Hilbert transform.
# Both amplitude and frequency are a function of time.
# To detect each artifact, each point will be compared with
# average amplitude of its block. The block size will be 1.5 ms.
# Using average amplitude can mitigate the error caused by
# Hilbert transform and noise.
# In some case, for more accuracy, the block size may be modified
# to other values.
DEFAULT_BLOCK_SIZE_SECS = 0.0015
# If the difference between average frequency of this block and
# dominant frequency of full signal is less than 0.5 times of
# dominant frequency, this block is considered to be within the
# sine wave. In most cases, if there is no sine wave(only noise),
# average frequency will be much greater than 5 times of
# dominant frequency.
# Also, for delay during playback, the frequency will be about 0
# in perfect situation or much greater than 5 times of dominant
# frequency if it's noised.
DEFAULT_FREQUENCY_ERROR = 0.5
# If the amplitude of some sample is less than 0.6 times of the
# average amplitude of its left/right block, it will be considered
# as a delay during playing.
DEFAULT_DELAY_AMPLITUDE_THRESHOLD = 0.6
# If the average amplitude of the block before or after playing
# is more than 0.5 times to the average amplitude of the wave,
# it will be considered as a noise artifact.
DEFAULT_NOISE_AMPLITUDE_THRESHOLD = 0.5
# In the sine wave, if the amplitude is more than 1.4 times of
# its left side and its right side, it will be considered as
# a burst.
DEFAULT_BURST_AMPLITUDE_THRESHOLD = 1.4
# When detecting burst, if the amplitude is lower than 0.5 times
# average amplitude, we ignore it.
DEFAULT_BURST_TOO_SMALL = 0.5
# For a signal which is the combination of sine wave with fixed frequency f and
# amplitude 1 and standard noise with amplitude k, the average teager value is
# nearly linear to the noise level k.
# Given frequency f, we simulate a sine wave with default noise level and
# calculate its average teager value. Then, we can estimate the equivalent
# noise level of input signal by the average teager value of input signal.
DEFAULT_STANDARD_NOISE = 0.005
# For delay, burst, volume increasing/decreasing, if two delay(
# burst, volume increasing/decreasing) happen within
# DEFAULT_SAME_EVENT_SECS seconds, we consider they are the
# same event.
DEFAULT_SAME_EVENT_SECS = 0.001
# When detecting increasing/decreasing volume of signal, if the amplitude
# is lower than 0.1 times average amplitude, we ignore it.
DEFAULT_VOLUME_CHANGE_TOO_SMALL = 0.1
# If average amplitude of right block is less/more than average
# amplitude of left block times DEFAULT_VOLUME_CHANGE_AMPLITUDE, it will be
# considered as decreasing/increasing on volume.
DEFAULT_VOLUME_CHANGE_AMPLITUDE = 0.1
# If the increasing/decreasing volume event is too close to the start or the end
# of sine wave, we consider its volume change as part of rising/falling phase in
# the start/end.
NEAR_START_OR_END_SECS = 0.01
# After applying Hilbert transform, the resulting amplitude and frequency may be
# extremely large in the start and/or the end part. Thus, we will append zeros
# before and after the whole wave for 0.1 secs.
APPEND_ZEROS_SECS = 0.1
# If the noise event is too close to the start or the end of the data, we
# consider its noise as part of artifacts caused by edge effect of Hilbert
# transform.
# For example, originally, the data duration is 10 seconds.
# We append 0.1 seconds of zeros in the beginning and the end of the data, so
# the data becomes 10.2 seocnds long.
# Then, we apply Hilbert transform to 10.2 seconds of data.
# Near 0.1 seconds and 10.1 seconds, there will be edge effect of Hilbert
# transform. We do not want these be treated as noise.
# If NEAR_DATA_START_OR_END_SECS is set to 0.01, then the noise happened
# at [0, 0.11] and [10.09, 10.1] will be ignored.
NEAR_DATA_START_OR_END_SECS = 0.01
# If the noise event is too close to the start or the end of the sine wave in
# the data, we consider its noise as part of artifacts caused by edge effect of
# Hilbert transform.
# A |-------------|vvvvvvvvvvvvvvvvvvvvvvv|-------------|
# B |ooooooooo| d | | d |ooooooooo|
#
# A is full signal. It contains a sine wave and silence before and after sine
# wave.
# In B, |oooo| shows the parts that we are going to check for noise before/after
# sine wave. | d | is determined by NEAR_SINE_START_OR_END_SECS.
NEAR_SINE_START_OR_END_SECS = 0.01
class SineWaveNotFound(Exception):
"""Error when there's no sine wave found in the signal"""
pass
def hilbert(x):
"""Hilbert transform copied from scipy.
More information can be found here:
http://docs.scipy.org/doc/scipy/reference/generated/scipy.signal.hilbert.html
@param x: Real signal data to transform.
@returns: Analytic signal of x, we can further extract amplitude and
frequency from it.
"""
x = numpy.asarray(x)
if numpy.iscomplexobj(x):
raise ValueError("x must be real.")
axis = -1
N = x.shape[axis]
if N <= 0:
raise ValueError("N must be positive.")
Xf = numpy.fft.fft(x, N, axis=axis)
h = numpy.zeros(N)
if N % 2 == 0:
h[0] = h[N // 2] = 1
h[1:N // 2] = 2
else:
h[0] = 1
h[1:(N + 1) // 2] = 2
if len(x.shape) > 1:
ind = [newaxis] * x.ndim
ind[axis] = slice(None)
h = h[ind]
x = numpy.fft.ifft(Xf * h, axis=axis)
return x
def noised_sine_wave(frequency, rate, noise_level):
"""Generates a sine wave of 2 second with specified noise level.
@param frequency: Frequency of sine wave.
@param rate: Sampling rate.
@param noise_level: Required noise level.
@returns: A sine wave with specified noise level.
"""
wave = []
for index in xrange(0, rate * 2):
sample = 2.0 * math.pi * frequency * float(index) / float(rate)
sine_wave = math.sin(sample)
noise = noise_level * numpy.random.standard_normal()
wave.append(sine_wave + noise)
return wave
def average_teager_value(wave, amplitude):
"""Computes the normalized average teager value.
After averaging the teager value, we will normalize the value by
dividing square of amplitude.
@param wave: Wave to apply teager operator.
@param amplitude: Average amplitude of given wave.
@returns: Average teager value.
"""
teager_value, length = 0, len(wave)
for i in range(1, length - 1):
ith_teager_value = abs(wave[i] * wave[i] - wave[i - 1] * wave[i + 1])
ith_teager_value *= max(1, abs(wave[i]))
teager_value += ith_teager_value
teager_value = (float(teager_value) / length) / (amplitude ** 2)
return teager_value
def noise_level(amplitude, frequency, rate, teager_value_of_input):
"""Computes the noise level compared with standard_noise.
For a signal which is the combination of sine wave with fixed frequency f
and amplitude 1 and standard noise with amplitude k, the average teager
value is nearly linear to the noise level k.
Thus, we can compute the average teager value of a sine wave with
standard_noise. Then, we can estimate the noise level of given input.
@param amplitude: Amplitude of input audio.
@param frequency: Dominant frequency of input audio.
@param rate: Sampling rate.
@param teager_value_of_input: Average teager value of input audio.
@returns: A float value denotes the audio is equivalent to have
how many times of noise compared with its amplitude.
For example, 0.02 denotes that the wave has a noise which
has standard distribution with standard deviation being
0.02 times the amplitude of the wave.
"""
standard_noise = DEFAULT_STANDARD_NOISE
# Generates the standard sine wave with stdandard_noise level of noise.
standard_wave = noised_sine_wave(frequency, rate, standard_noise)
# Calculates the average teager value.
teager_value_of_std_wave = average_teager_value(standard_wave, amplitude)
return (teager_value_of_input / teager_value_of_std_wave) * standard_noise
def error(f1, f2):
"""Calculates the relative error between f1 and f2.
@param f1: Exact value.
@param f2: Test value.
@returns: Relative error between f1 and f2.
"""
return abs(float(f1) - float(f2)) / float(f1)
def hilbert_analysis(signal, rate, block_size):
"""Finds amplitude and frequency of each time of signal by Hilbert transform.
@param signal: The wave to analyze.
@param rate: Sampling rate.
@param block_size: The size of block to transform.
@returns: A tuple of list: (amplitude, frequency) composed of
amplitude and frequency of each time.
"""
# To apply Hilbert transform, the wave will be transformed
# segment by segment. For each segment, its size will be
# block_size and we will only take middle part of it.
# Thus, each segment looks like: |-----|=====|=====|-----|.
# "=...=" part will be taken while "-...-" part will be ignored.
#
# The whole size of taken part will be half of block_size
# which will be hilbert_block.
# The size of each ignored part will be half of hilbert_block
# which will be half_hilbert_block.
hilbert_block = block_size // 2
half_hilbert_block = hilbert_block // 2
# As mentioned above, for each block, we will only take middle
# part of it. Thus, the whole transformation will be completed as:
# |=====|=====|-----| |-----|=====|=====|-----|
# |-----|=====|=====|-----| |-----|=====|=====|
# |-----|=====|=====|-----|
# Specially, beginning and ending part may not have ignored part.
length = len(signal)
result = []
for left_border in xrange(0, length, hilbert_block):
right_border = min(length, left_border + hilbert_block)
temp_left_border = max(0, left_border - half_hilbert_block)
temp_right_border = min(length, right_border + half_hilbert_block)
temp = hilbert(signal[temp_left_border:temp_right_border])
for index in xrange(left_border, right_border):
result.append(temp[index - temp_left_border])
result = numpy.asarray(result)
amplitude = numpy.abs(result)
phase = numpy.unwrap(numpy.angle(result))
frequency = numpy.diff(phase) / (2.0 * numpy.pi) * rate
#frequency.append(frequency[len(frequency)-1])
frequecny = numpy.append(frequency, frequency[len(frequency) - 1])
return (amplitude, frequency)
def find_block_average_value(arr, side_block_size, block_size):
"""For each index, finds average value of its block, left block, right block.
It will find average value for each index in the range.
For each index, the range of its block is
[max(0, index - block_size / 2), min(length - 1, index + block_size / 2)]
For each index, the range of its left block is
[max(0, index - size_block_size), index]
For each index, the range of its right block is
[index, min(length - 1, index + side_block_size)]
@param arr: The array to be computed.
@param side_block_size: the size of the left_block and right_block.
@param block_size: the size of the block.
@returns: A tuple of lists: (left_block_average_array,
right_block_average_array,
block_average_array)
"""
length = len(arr)
left_border, right_border = 0, 1
left_block_sum = arr[0]
right_block_sum = arr[0]
left_average_array = numpy.zeros(length)
right_average_array = numpy.zeros(length)
block_average_array = numpy.zeros(length)
for index in xrange(0, length):
while left_border < index - side_block_size:
left_block_sum -= arr[left_border]
left_border += 1
while right_border < min(length, index + side_block_size):
right_block_sum += arr[right_border]
right_border += 1
left_average_value = float(left_block_sum) / (index - left_border + 1)
right_average_value = float(right_block_sum) / (right_border - index)
left_average_array[index] = left_average_value
right_average_array[index] = right_average_value
if index + 1 < length:
left_block_sum += arr[index + 1]
right_block_sum -= arr[index]
left_border, right_border = 0, 1
block_sum = 0
for index in xrange(0, length):
while left_border < index - block_size / 2:
block_sum -= arr[left_border]
left_border += 1
while right_border < min(length, index + block_size / 2):
block_sum += arr[right_border]
right_border += 1
average_value = float(block_sum) / (right_border - left_border)
block_average_array[index] = average_value
return (left_average_array, right_average_array, block_average_array)
def find_start_end_index(dominant_frequency,
block_frequency_delta,
block_size,
frequency_error_threshold):
"""Finds start and end index of sine wave.
For each block with size of block_size, we check that whether its frequency
is close enough to the dominant_frequency. If yes, we will consider this
block to be within the sine wave.
Then, it will return the start and end index of sine wave indicating that
sine wave is between [start_index, end_index)
It's okay if the whole signal only contains sine wave.
@param dominant_frequency: Dominant frequency of signal.
@param block_frequency_delta: Average absolute difference between dominant
frequency and frequency of each block. For
each index, its block is
[max(0, index - block_size / 2),
min(length - 1, index + block_size / 2)]
@param block_size: Block size in samples.
@returns: A tuple composed of (start_index, end_index)
"""
length = len(block_frequency_delta)
# Finds the start/end time index of playing based on dominant frequency
start_index, end_index = length - 1 , 0
for index in xrange(0, length):
left_border = max(0, index - block_size / 2)
right_border = min(length - 1, index + block_size / 2)
frequency_error = block_frequency_delta[index] / dominant_frequency
if frequency_error < frequency_error_threshold:
start_index = min(start_index, left_border)
end_index = max(end_index, right_border + 1)
return (start_index, end_index)
def noise_detection(start_index, end_index,
block_amplitude, average_amplitude,
rate, noise_amplitude_threshold):
"""Detects noise before/after sine wave.
If average amplitude of some sample's block before start of wave or after
end of wave is more than average_amplitude times noise_amplitude_threshold,
it will be considered as a noise.
@param start_index: Start index of sine wave.
@param end_index: End index of sine wave.
@param block_amplitude: An array for average amplitude of each block, where
amplitude is computed from Hilbert transform.
@param average_amplitude: Average amplitude of sine wave.
@param rate: Sampling rate
@param noise_amplitude_threshold: If the average amplitude of a block is
higher than average amplitude of the wave times
noise_amplitude_threshold, it will be considered as
noise before/after playback.
@returns: A tuple of lists indicating the time that noise happens:
(noise_before_playing, noise_after_playing).
"""
length = len(block_amplitude)
amplitude_threshold = average_amplitude * noise_amplitude_threshold
same_event_samples = rate * DEFAULT_SAME_EVENT_SECS
# Detects noise before playing.
noise_time_point = []
last_noise_end_time_point = []
previous_noise_index = None
times = 0
for index in xrange(0, length):
# Ignore noise too close to the beginning or the end of sine wave.
# Check the docstring of NEAR_SINE_START_OR_END_SECS.
if ((start_index - rate * NEAR_SINE_START_OR_END_SECS) <= index and
(index < end_index + rate * NEAR_SINE_START_OR_END_SECS)):
continue
# Ignore noise too close to the beginning or the end of original data.
# Check the docstring of NEAR_DATA_START_OR_END_SECS.
if (float(index) / rate <=
NEAR_DATA_START_OR_END_SECS + APPEND_ZEROS_SECS):
continue
if (float(length - index) / rate <=
NEAR_DATA_START_OR_END_SECS + APPEND_ZEROS_SECS):
continue
if block_amplitude[index] > amplitude_threshold:
same_event = False
if previous_noise_index:
same_event = (index - previous_noise_index) < same_event_samples
if not same_event:
index_start_sec = float(index) / rate - APPEND_ZEROS_SECS
index_end_sec = float(index + 1) / rate - APPEND_ZEROS_SECS
noise_time_point.append(index_start_sec)
last_noise_end_time_point.append(index_end_sec)
times += 1
index_end_sec = float(index + 1) / rate - APPEND_ZEROS_SECS
last_noise_end_time_point[times - 1] = index_end_sec
previous_noise_index = index
noise_before_playing, noise_after_playing = [], []
for i in xrange(times):
duration = last_noise_end_time_point[i] - noise_time_point[i]
if noise_time_point[i] < float(start_index) / rate - APPEND_ZEROS_SECS:
noise_before_playing.append((noise_time_point[i], duration))
else:
noise_after_playing.append((noise_time_point[i], duration))
return (noise_before_playing, noise_after_playing)
def delay_detection(start_index, end_index,
block_amplitude, average_amplitude,
dominant_frequency,
rate,
left_block_amplitude,
right_block_amplitude,
block_frequency_delta,
delay_amplitude_threshold,
frequency_error_threshold):
"""Detects delay during playing.
For each sample, we will check whether the average amplitude of its block
is less than average amplitude of its left block and its right block times
delay_amplitude_threshold. Also, we will check whether the frequency of
its block is far from the dominant frequency.
If at least one constraint fulfilled, it will be considered as a delay.
@param start_index: Start index of sine wave.
@param end_index: End index of sine wave.
@param block_amplitude: An array for average amplitude of each block, where
amplitude is computed from Hilbert transform.
@param average_amplitude: Average amplitude of sine wave.
@param dominant_frequency: Dominant frequency of signal.
@param rate: Sampling rate
@param left_block_amplitude: Average amplitude of left block of each index.
Ref to find_block_average_value function.
@param right_block_amplitude: Average amplitude of right block of each index.
Ref to find_block_average_value function.
@param block_frequency_delta: Average absolute difference frequency to
dominant frequency of block of each index.
Ref to find_block_average_value function.
@param delay_amplitude_threshold: If the average amplitude of a block is
lower than average amplitude of the wave times
delay_amplitude_threshold, it will be considered
as delay.
@param frequency_error_threshold: Ref to DEFAULT_FREQUENCY_ERROR
@returns: List of delay occurrence:
[(time_1, duration_1), (time_2, duration_2), ...],
where time and duration are in seconds.
"""
delay_time_points = []
last_delay_end_time_points = []
previous_delay_index = None
times = 0
same_event_samples = rate * DEFAULT_SAME_EVENT_SECS
start_time = float(start_index) / rate - APPEND_ZEROS_SECS
end_time = float(end_index) / rate - APPEND_ZEROS_SECS
for index in xrange(start_index, end_index):
if block_amplitude[index] > average_amplitude * delay_amplitude_threshold:
continue
now_time = float(index) / rate - APPEND_ZEROS_SECS
if abs(now_time - start_time) < NEAR_START_OR_END_SECS:
continue
if abs(now_time - end_time) < NEAR_START_OR_END_SECS:
continue
# If amplitude less than its left/right side and small enough,
# it will be considered as a delay.
amp_threshold = average_amplitude * delay_amplitude_threshold
left_threshold = delay_amplitude_threshold * left_block_amplitude[index]
amp_threshold = min(amp_threshold, left_threshold)
right_threshold = delay_amplitude_threshold * right_block_amplitude[index]
amp_threshold = min(amp_threshold, right_threshold)
frequency_error = block_frequency_delta[index] / dominant_frequency
amplitude_too_small = block_amplitude[index] < amp_threshold
frequency_not_match = frequency_error > frequency_error_threshold
if amplitude_too_small or frequency_not_match:
same_event = False
if previous_delay_index:
same_event = (index - previous_delay_index) < same_event_samples
if not same_event:
index_start_sec = float(index) / rate - APPEND_ZEROS_SECS
index_end_sec = float(index + 1) / rate - APPEND_ZEROS_SECS
delay_time_points.append(index_start_sec)
last_delay_end_time_points.append(index_end_sec)
times += 1
previous_delay_index = index
index_end_sec = float(index + 1) / rate - APPEND_ZEROS_SECS
last_delay_end_time_points[times - 1] = index_end_sec
delay_list = []
for i in xrange(len(delay_time_points)):
duration = last_delay_end_time_points[i] - delay_time_points[i]
delay_list.append( (delay_time_points[i], duration) )
return delay_list
def burst_detection(start_index, end_index,
block_amplitude, average_amplitude,
dominant_frequency,
rate,
left_block_amplitude,
right_block_amplitude,
block_frequency_delta,
burst_amplitude_threshold,
frequency_error_threshold):
"""Detects burst during playing.
For each sample, we will check whether the average amplitude of its block is
more than average amplitude of its left block and its right block times
burst_amplitude_threshold. Also, we will check whether the frequency of
its block is not compatible to the dominant frequency.
If at least one constraint fulfilled, it will be considered as a burst.
@param start_index: Start index of sine wave.
@param end_index: End index of sine wave.
@param block_amplitude: An array for average amplitude of each block, where
amplitude is computed from Hilbert transform.
@param average_amplitude: Average amplitude of sine wave.
@param dominant_frequency: Dominant frequency of signal.
@param rate: Sampling rate
@param left_block_amplitude: Average amplitude of left block of each index.
Ref to find_block_average_value function.
@param right_block_amplitude: Average amplitude of right block of each index.
Ref to find_block_average_value function.
@param block_frequency_delta: Average absolute difference frequency to
dominant frequency of block of each index.
@param burst_amplitude_threshold: If the amplitude is higher than average
amplitude of its left block and its right block
times burst_amplitude_threshold. It will be
considered as a burst.
@param frequency_error_threshold: Ref to DEFAULT_FREQUENCY_ERROR
@returns: List of burst occurence: [time_1, time_2, ...],
where time is in seconds.
"""
burst_time_points = []
previous_burst_index = None
same_event_samples = rate * DEFAULT_SAME_EVENT_SECS
for index in xrange(start_index, end_index):
# If amplitude higher than its left/right side and large enough,
# it will be considered as a burst.
if block_amplitude[index] <= average_amplitude * DEFAULT_BURST_TOO_SMALL:
continue
if abs(index - start_index) < rate * NEAR_START_OR_END_SECS:
continue
if abs(index - end_index) < rate * NEAR_START_OR_END_SECS:
continue
amp_threshold = average_amplitude * DEFAULT_BURST_TOO_SMALL
left_threshold = burst_amplitude_threshold * left_block_amplitude[index]
amp_threshold = max(amp_threshold, left_threshold)
right_threshold = burst_amplitude_threshold * right_block_amplitude[index]
amp_threshold = max(amp_threshold, right_threshold)
frequency_error = block_frequency_delta[index] / dominant_frequency
amplitude_too_large = block_amplitude[index] > amp_threshold
frequency_not_match = frequency_error > frequency_error_threshold
if amplitude_too_large or frequency_not_match:
same_event = False
if previous_burst_index:
same_event = index - previous_burst_index < same_event_samples
if not same_event:
burst_time_points.append(float(index) / rate - APPEND_ZEROS_SECS)
previous_burst_index = index
return burst_time_points
def changing_volume_detection(start_index, end_index,
average_amplitude,
rate,
left_block_amplitude,
right_block_amplitude,
volume_changing_amplitude_threshold):
"""Finds volume changing during playback.
For each index, we will compare average amplitude of its left block and its
right block. If average amplitude of right block is more than average
amplitude of left block times (1 + DEFAULT_VOLUME_CHANGE_AMPLITUDE), it will
be considered as an increasing volume. If the one of right block is less
than that of left block times (1 - DEFAULT_VOLUME_CHANGE_AMPLITUDE), it will
be considered as a decreasing volume.
@param start_index: Start index of sine wave.
@param end_index: End index of sine wave.
@param average_amplitude: Average amplitude of sine wave.
@param rate: Sampling rate
@param left_block_amplitude: Average amplitude of left block of each index.
Ref to find_block_average_value function.
@param right_block_amplitude: Average amplitude of right block of each index.
Ref to find_block_average_value function.
@param volume_changing_amplitude_threshold: If the average amplitude of right
block is higher or lower than
that of left one times this
value, it will be considered as
a volume change.
Also refer to
DEFAULT_VOLUME_CHANGE_AMPLITUDE
@returns: List of volume changing composed of 1 for increasing and
-1 for decreasing.
"""
length = len(left_block_amplitude)
# Detects rising and/or falling volume.
previous_rising_index, previous_falling_index = None, None
changing_time = []
changing_events = []
amplitude_threshold = average_amplitude * DEFAULT_VOLUME_CHANGE_TOO_SMALL
same_event_samples = rate * DEFAULT_SAME_EVENT_SECS
for index in xrange(start_index, end_index):
# Skips if amplitude is too small.
if left_block_amplitude[index] < amplitude_threshold:
continue
if right_block_amplitude[index] < amplitude_threshold:
continue
# Skips if changing is from start or end time
if float(abs(start_index - index)) / rate < NEAR_START_OR_END_SECS:
continue
if float(abs(end_index - index)) / rate < NEAR_START_OR_END_SECS:
continue
delta_margin = volume_changing_amplitude_threshold
if left_block_amplitude[index] > 0:
delta_margin *= left_block_amplitude[index]
increasing_threshold = left_block_amplitude[index] + delta_margin
decreasing_threshold = left_block_amplitude[index] - delta_margin
if right_block_amplitude[index] > increasing_threshold:
same_event = False
if previous_rising_index:
same_event = index - previous_rising_index < same_event_samples
if not same_event:
changing_time.append(float(index) / rate - APPEND_ZEROS_SECS)
changing_events.append(+1)
previous_rising_index = index
if right_block_amplitude[index] < decreasing_threshold:
same_event = False
if previous_falling_index:
same_event = index - previous_falling_index < same_event_samples
if not same_event:
changing_time.append(float(index) / rate - APPEND_ZEROS_SECS)
changing_events.append(-1)
previous_falling_index = index
# Combines consecutive increasing/decreasing event.
combined_changing_events, prev = [], 0
for i in xrange(len(changing_events)):
if changing_events[i] == prev:
continue
combined_changing_events.append((changing_time[i], changing_events[i]))
prev = changing_events[i]
return combined_changing_events
def quality_measurement(
signal, rate,
dominant_frequency=None,
block_size_secs=DEFAULT_BLOCK_SIZE_SECS,
frequency_error_threshold=DEFAULT_FREQUENCY_ERROR,
delay_amplitude_threshold=DEFAULT_DELAY_AMPLITUDE_THRESHOLD,
noise_amplitude_threshold=DEFAULT_NOISE_AMPLITUDE_THRESHOLD,
burst_amplitude_threshold=DEFAULT_BURST_AMPLITUDE_THRESHOLD,
volume_changing_amplitude_threshold=DEFAULT_VOLUME_CHANGE_AMPLITUDE):
"""Detects several artifacts and estimates the noise level.
This method detects artifact before playing, after playing, and delay
during playing. Also, it estimates the noise level of the signal.
To avoid the influence of noise, it calculates amplitude and frequency
block by block.
@param signal: A list of numbers for one-channel PCM data. The data should
be normalized to [-1,1].
@param rate: Sampling rate
@param dominant_frequency: Dominant frequency of signal. Set None to
recalculate the frequency in this function.
@param block_size_secs: Block size in seconds. The measurement will be done
block-by-block using average amplitude and frequency
in each block to avoid noise.
@param frequency_error_threshold: Ref to DEFAULT_FREQUENCY_ERROR.
@param delay_amplitude_threshold: If the average amplitude of a block is
lower than average amplitude of the wave
times delay_amplitude_threshold, it will
be considered as delay.
Also refer to delay_detection and
DEFAULT_DELAY_AMPLITUDE_THRESHOLD.
@param noise_amplitude_threshold: If the average amplitude of a block is
higher than average amplitude of the wave
times noise_amplitude_threshold, it will
be considered as noise before/after
playback.
Also refer to noise_detection and
DEFAULT_NOISE_AMPLITUDE_THRESHOLD.
@param burst_amplitude_threshold: If the average amplitude of a block is
higher than average amplitude of its left
block and its right block times
burst_amplitude_threshold. It will be
considered as a burst.
Also refer to burst_detection and
DEFAULT_BURST_AMPLITUDE_THRESHOLD.
@param volume_changing_amplitude_threshold: If the average amplitude of right
block is higher or lower than
that of left one times this
value, it will be considered as
a volume change.
Also refer to
changing_volume_detection and
DEFAULT_VOLUME_CHANGE_AMPLITUDE
@returns: A dictoinary of detection/estimation:
{'artifacts':
{'noise_before_playback':
[(time_1, duration_1), (time_2, duration_2), ...],
'noise_after_playback':
[(time_1, duration_1), (time_2, duration_2), ...],
'delay_during_playback':
[(time_1, duration_1), (time_2, duration_2), ...],
'burst_during_playback':
[time_1, time_2, ...]
},
'volume_changes':
[(time_1, flag_1), (time_2, flag_2), ...],
'equivalent_noise_level': level
}
where durations and time points are in seconds. And,
equivalence_noise_level is the quotient of noise and
wave which refers to DEFAULT_STANDARD_NOISE.
volume_changes is a list of tuples containing time
stamps and decreasing/increasing flags for volume
change events.
"""
# Calculates the block size, from seconds to samples.
block_size = int(block_size_secs * rate)
signal = numpy.concatenate((numpy.zeros(int(rate * APPEND_ZEROS_SECS)),
signal,
numpy.zeros(int(rate * APPEND_ZEROS_SECS))))
signal = numpy.array(signal, dtype=float)
length = len(signal)
# Calculates the amplitude and frequency.
amplitude, frequency = hilbert_analysis(signal, rate, block_size)
# Finds the dominant frequency.
if not dominant_frequency:
dominant_frequency = audio_analysis.spectral_analysis(signal, rate)[0][0]
# Finds the array which contains absolute difference between dominant
# frequency and frequency at each time point.
frequency_delta = abs(frequency - dominant_frequency)
# Computes average amplitude of each type of block
res = find_block_average_value(amplitude, block_size * 2, block_size)
left_block_amplitude, right_block_amplitude, block_amplitude = res
# Computes average absolute difference of frequency and dominant frequency
# of the block of each index
_, _, block_frequency_delta = find_block_average_value(frequency_delta,
block_size * 2,
block_size)
# Finds start and end index of sine wave.
start_index, end_index = find_start_end_index(dominant_frequency,
block_frequency_delta,
block_size,
frequency_error_threshold)
if start_index > end_index:
raise SineWaveNotFound('No sine wave found in signal')
logging.debug('Found sine wave: start: %s, end: %s',
float(start_index) / rate - APPEND_ZEROS_SECS,
float(end_index) / rate - APPEND_ZEROS_SECS)
sum_of_amplitude = float(sum(amplitude[start_index:end_index]))
# Finds average amplitude of sine wave.
average_amplitude = sum_of_amplitude / (end_index - start_index)
# Finds noise before and/or after playback.
noise_before_playing, noise_after_playing = noise_detection(
start_index, end_index,
block_amplitude, average_amplitude,
rate,
noise_amplitude_threshold)
# Finds delay during playback.
delays = delay_detection(start_index, end_index,
block_amplitude, average_amplitude,
dominant_frequency,
rate,
left_block_amplitude,
right_block_amplitude,
block_frequency_delta,
delay_amplitude_threshold,
frequency_error_threshold)
# Finds burst during playback.
burst_time_points = burst_detection(start_index, end_index,
block_amplitude, average_amplitude,
dominant_frequency,
rate,
left_block_amplitude,
right_block_amplitude,
block_frequency_delta,
burst_amplitude_threshold,
frequency_error_threshold)
# Finds volume changing during playback.
volume_changes = changing_volume_detection(
start_index, end_index,
average_amplitude,
rate,
left_block_amplitude,
right_block_amplitude,
volume_changing_amplitude_threshold)
# Calculates the average teager value.
teager_value = average_teager_value(signal[start_index:end_index],
average_amplitude)
# Finds out the noise level.
noise = noise_level(average_amplitude, dominant_frequency,
rate,
teager_value)
return {'artifacts':
{'noise_before_playback': noise_before_playing,
'noise_after_playback': noise_after_playing,
'delay_during_playback': delays,
'burst_during_playback': burst_time_points
},
'volume_changes': volume_changes,
'equivalent_noise_level': noise
}