admin/webrtc_m130
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

This CL fixes the following:

Browse Source 
- snake_case -> CapWords
- compulsory docstring added
- style

A followup CL will fix remaining issues as raised by the next version of the WebRTC Python linter (update in progress).

BUG=webrtc:7218
NOTRY=True

Review-Url: https://codereview.webrtc.org/2793903006
Cr-Commit-Position: refs/heads/master@{#17543}
This commit is contained in:
alessiob 2017-04-05 07:25:33 -07:00 committed by Commit bot
parent 129fc9ce94
commit dea682d25b
19 changed files with 532 additions and 268 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

11
webrtc/modules/audio_processing/test/py_quality_assessment/README.md
View File

@ -12,8 +12,15 @@ reference one used for evaluation.
- OS: Linux
- Python 2.7
- Python libraries: numpy, scipy, pydub (0.17.0+)
- `$ sudo apt install python-numpy python-scipy`
- `$ sudo pip install pydub`
- It is recommended that a dedicated Python environment is used
- install `virtualenv`
- `$ sudo apt-get install python-virtualenv`
- setup a new Python environment (e.g., `my_env`)
- `$ cd ~ && virtualenv my_env`
- activate the new Python environment
- `$ source ~/my_env/bin/activate`
- add dependcies via `pip`
- `(my_env)$ pip install numpy pydub scipy`
- PolqaOem64 (see http://www.polqa.info/)
- Tested with POLQA Library v1.180 / P863 v2.400
- Aachen Impulse Response (AIR) Database

9
webrtc/modules/audio_processing/test/py_quality_assessment/apm_quality_assessment.py
View File

@ -32,7 +32,10 @@ _EVAL_SCORE_WORKER_NAMES = _EVAL_SCORE_WORKER_CLASSES.keys()
_DEFAULT_CONFIG_FILE = 'apm_configs/default.json'
def _instance_arguments_parser():
def _InstanceArgumentsParser():
"""Arguments parser factory.
"""
parser = argparse.ArgumentParser(description=(
'Perform APM module quality assessment on one or more input files using '
'one or more audioproc_f configuration files and one or more noise '
@ -76,13 +79,13 @@ def main():
# TODO(alessiob): level = logging.INFO once debugged.
logging.basicConfig(level=logging.DEBUG)
parser = _instance_arguments_parser()
parser = _InstanceArgumentsParser()
args = parser.parse_args()
simulator = simulation.ApmModuleSimulator(
aechen_ir_database_path=args.air_db_path,
polqa_tool_path=args.polqa_path)
simulator.run(
simulator.Run(
config_filepaths=args.config_files,
input_filepaths=args.input_files,
noise_generator_names=args.noise_generators,

46
webrtc/modules/audio_processing/test/py_quality_assessment/apm_quality_assessment_export.py
View File

@ -29,7 +29,10 @@ RE_INPUT_NAME = re.compile(r'input-(.+)')
RE_NOISE_NAME = re.compile(r'noise-(.+)')
RE_SCORE_NAME = re.compile(r'score-(.+)\.txt')
def _InstanceArgumentsParser():
"""Arguments parser factory.
"""
parser = argparse.ArgumentParser(description=(
'Exports pre-computed APM module quality assessment results into HTML '
'tables.'))
@ -61,8 +64,15 @@ def _InstanceArgumentsParser():
def _GetScoreDescriptors(score_filepath):
"""
Extract a score descriptors from the score file path.
"""Extracts a score descriptor from the given score file path.
Args:
score_filepath: path to the score file.
Returns:
A tuple of strings (APM configuration name, input audio track name,
noise generator name, noise generator parameters name, evaluation score
name).
"""
config_name, input_name, noise_name, noise_params, score_name = (
score_filepath.split(os.sep)[-5:])
@ -74,10 +84,21 @@ def _GetScoreDescriptors(score_filepath):
def _ExcludeScore(config_name, input_name, noise_name, score_name, args):
"""
"""Decides whether excluding a score.
Given a score descriptor, encoded in config_name, input_name, noise_name, and
score_name, use the corresponding regular expressions to determine if the
score should be excluded.
Args:
config_name: APM configuration name.
input_name: input audio track name.
noise_name: noise generator name.
score_name: evaluation score name.
args: parsed arguments.
Returns:
A boolean.
"""
value_regexpr_pairs = [
(config_name, args.config_names),
@ -96,9 +117,14 @@ def _ExcludeScore(config_name, input_name, noise_name, score_name, args):
return False
def _GetOutputFilename(filename_suffix):
"""
Build the filename for the exported file.
def _BuildOutputFilename(filename_suffix):
"""Builds the filename for the exported file.
Args:
filename_suffix: suffix for the output file name.
Returns:
A string.
"""
if filename_suffix is None:
return 'results.html'
@ -133,23 +159,23 @@ def main():
# Get metadata.
score_path, _ = os.path.split(score_filepath)
audio_in_filepath, audio_ref_filepath = (
data_access.Metadata.load_audio_in_ref_paths(score_path))
data_access.Metadata.LoadAudioInRefPaths(score_path))
audio_out_filepath = os.path.abspath(os.path.join(
score_path, audioproc_wrapper.AudioProcWrapper.OUTPUT_FILENAME))
# Add the score to the nested dictionary.
scores[score_name][config_name][input_name][noise_name][noise_params] = {
'score': data_access.ScoreFile.load(score_filepath),
'score': data_access.ScoreFile.Load(score_filepath),
'audio_in_filepath': audio_in_filepath,
'audio_out_filepath': audio_out_filepath,
'audio_ref_filepath': audio_ref_filepath,
}
# Export.
output_filepath = os.path.join(args.output_dir, _GetOutputFilename(
output_filepath = os.path.join(args.output_dir, _BuildOutputFilename(
args.filename_suffix))
exporter = export.HtmlExport(output_filepath)
exporter.export(scores)
exporter.Export(scores)
logging.info('output file successfully written in %s', output_filepath)
sys.exit(0)

35
webrtc/modules/audio_processing/test/py_quality_assessment/apm_quality_assessment_gencfgs.py
View File

@ -18,8 +18,10 @@ import quality_assessment.data_access as data_access
OUTPUT_PATH = os.path.abspath('apm_configs')
def _generate_default_overridden(config_override):
"""
def _GenerateDefaultOverridden(config_override):
"""Generates one or more APM overriden configurations.
For each item in config_override, it overrides the default configuration and
writes a new APM configuration file.
@ -39,8 +41,11 @@ def _generate_default_overridden(config_override):
settings.use_ns = rtc::Optional<bool>(true);
settings.use_ts = rtc::Optional<bool>(true);
settings.use_vad = rtc::Optional<bool>(true);
"""
Args:
config_override: dict of APM configuration file names as keys; the values
are dict instances encoding the audioproc_f flags.
"""
for config_filename in config_override:
config = config_override[config_filename]
config['-all_default'] = None
@ -49,14 +54,12 @@ def _generate_default_overridden(config_override):
config_filename))
logging.debug('config file <%s> | %s', config_filepath, config)
data_access.AudioProcConfigFile.save(config_filepath, config)
data_access.AudioProcConfigFile.Save(config_filepath, config)
logging.info('config file created: <%s>', config_filepath)
def generate_all_default_but_one():
"""
Generate configurations in which all the default flags are used but one (one
flag at a time is excluded).
def _GenerateAllDefaultButOne():
"""Disables the flags enabled by default one-by-one.
"""
CONFIG_SETS = {
'no_AEC': {'-aec': 0,},
@ -67,14 +70,11 @@ def generate_all_default_but_one():
'no_transient_suppressor': {'-ts': 0,},
'no_vad': {'-vad': 0,},
}
return _generate_default_overridden(CONFIG_SETS)
_GenerateDefaultOverridden(CONFIG_SETS)
def generate_all_default_plus_one():
"""
Generate configuratoins in which all the default flags are used and each
unused flag is added one at a time.
def _GenerateAllDefaultPlusOne():
"""Enables the flags disabled by default one-by-one.
"""
CONFIG_SETS = {
'with_AECM': {'-aec': 0, '-aecm': 1,}, # AEC and AECM are exclusive.
@ -87,14 +87,13 @@ def generate_all_default_plus_one():
'with_LC': {'-lc': 1,},
'with_refined_adaptive_filter': {'-refined_adaptive_filter': 1,},
}
return _generate_default_overridden(CONFIG_SETS)
_GenerateDefaultOverridden(CONFIG_SETS)
def main():
logging.basicConfig(level=logging.INFO)
generate_all_default_plus_one()
generate_all_default_but_one()
_GenerateAllDefaultPlusOne()
_GenerateAllDefaultButOne()
if __name__ == '__main__':

5
webrtc/modules/audio_processing/test/py_quality_assessment/apm_quality_assessment_unittest.py
View File

@ -6,11 +6,16 @@
# in the file PATENTS. All contributing project authors may
# be found in the AUTHORS file in the root of the source tree.
"""Unit tests for the apm_quality_assessment module.
"""
import unittest
import apm_quality_assessment
class TestSimulationScript(unittest.TestCase):
"""Unit tests for the apm_quality_assessment module.
"""
def test_main(self):
# Exit with error code if no arguments are passed.

4
webrtc/modules/audio_processing/test/py_quality_assessment/quality_assessment/audioproc_wrapper.py
View File

@ -36,7 +36,7 @@ class AudioProcWrapper(object):
def output_filepath(self):
return self._output_signal_filepath
def run(self, config_filepath, input_filepath, output_path):
def Run(self, config_filepath, input_filepath, output_path):
"""Run audioproc_f.
Args:
@ -57,7 +57,7 @@ class AudioProcWrapper(object):
return
# Load configuration.
self._config = data_access.AudioProcConfigFile.load(config_filepath)
self._config = data_access.AudioProcConfigFile.Load(config_filepath)
# Set remaining parametrs.
self._config['-i'] = self._input_signal_filepath

56
webrtc/modules/audio_processing/test/py_quality_assessment/quality_assessment/data_access.py
View File

@ -13,8 +13,8 @@ import json
import os
def make_directory(path):
"""Recursively make a directory without rising exceptions if already existing.
def MakeDirectory(path):
"""Makes a directory recursively without rising exceptions if existing.
Args:
path: path to the directory to be created.
@ -34,13 +34,14 @@ class Metadata(object):
_AUDIO_IN_REF_FILENAME = 'audio_in_ref.txt'
@classmethod
def load_audio_in_ref_paths(cls, metadata_path):
"""Metadata loader for input and reference audio track paths.
def LoadAudioInRefPaths(cls, metadata_path):
"""Loads the input and the reference audio track paths.
Args:
metadata_path: path to the directory containing the metadata file.
Returns: pair of metadata file paths for the input and output audio tracks.
Returns:
Pair of metadata file paths for the input and output audio tracks.
"""
metadata_filepath = os.path.join(metadata_path, cls._AUDIO_IN_REF_FILENAME)
with open(metadata_filepath) as f:
@ -49,9 +50,14 @@ class Metadata(object):
return audio_in_filepath, audio_ref_filepath
@classmethod
def save_audio_in_ref_paths(cls, output_path, audio_in_filepath,
def SaveAudioInRefPaths(cls, output_path, audio_in_filepath,
audio_ref_filepath):
"""Metadata saver for input and reference audio track paths.
"""Saves the input and the reference audio track paths.
Args:
output_path: path to the directory containing the metadata file.
audio_in_filepath: path to the input audio track file.
audio_ref_filepath: path to the reference audio track file.
"""
output_filepath = os.path.join(output_path, cls._AUDIO_IN_REF_FILENAME)
with open(output_filepath, 'w') as f:
@ -68,12 +74,26 @@ class AudioProcConfigFile(object):
pass
@classmethod
def load(cls, filepath):
def Load(cls, filepath):
"""Loads a configuration file for audioproc_f.
Args:
filepath: path to the configuration file.
Returns:
A dict containing the configuration.
"""
with open(filepath) as f:
return json.load(f)
@classmethod
def save(cls, filepath, config):
def Save(cls, filepath, config):
"""Saves a configuration file for audioproc_f.
Args:
filepath: path to the configuration file.
config: a dict containing the configuration.
"""
with open(filepath, 'w') as f:
json.dump(config, f)
@ -86,11 +106,25 @@ class ScoreFile(object):
pass
@classmethod
def load(cls, filepath):
def Load(cls, filepath):
"""Loads a score from file.
Args:
filepath: path to the score file.
Returns:
A float encoding the score.
"""
with open(filepath) as f:
return float(f.readline().strip())
@classmethod
def save(cls, filepath, score):
def Save(cls, filepath, score):
"""Saves a score into a file.
Args:
filepath: path to the score file.
score: float encoding the score.
"""
with open(filepath, 'w') as f:
f.write('{0:f}\n'.format(score))

99
webrtc/modules/audio_processing/test/py_quality_assessment/quality_assessment/eval_scores.py
View File

@ -33,10 +33,15 @@ class EvaluationScore(object):
self._score = None
@classmethod
def register_class(cls, class_to_register):
"""Register an EvaluationScore implementation.
def RegisterClass(cls, class_to_register):
"""Registers an EvaluationScore implementation.
Decorator to automatically register the classes that extend EvaluationScore.
Example usage:
@EvaluationScore.RegisterClass
class AudioLevelScore(EvaluationScore):
pass
"""
cls.REGISTERED_CLASSES[class_to_register.NAME] = class_to_register
return class_to_register
@ -49,54 +54,66 @@ class EvaluationScore(object):
def score(self):
return self._score
def set_reference_signal_filepath(self, filepath):
""" Set the path to the audio track used as reference signal.
def SetReferenceSignalFilepath(self, filepath):
""" Sets the path to the audio track used as reference signal.
Args:
filepath: path to the reference audio track.
"""
self._reference_signal_filepath = filepath
def set_tested_signal_filepath(self, filepath):
""" Set the path to the audio track used as test signal.
def SetTestedSignalFilepath(self, filepath):
""" Sets the path to the audio track used as test signal.
Args:
filepath: path to the test audio track.
"""
self._tested_signal_filepath = filepath
def _load_reference_signal(self):
assert self._reference_signal_filepath is not None
self._reference_signal = signal_processing.SignalProcessingUtils.load_wav(
self._reference_signal_filepath)
def _load_tested_signal(self):
assert self._tested_signal_filepath is not None
self._tested_signal = signal_processing.SignalProcessingUtils.load_wav(
self._tested_signal_filepath)
def run(self, output_path):
def Run(self, output_path):
"""Extracts the score for the set input-reference pair.
Args:
output_path: path to the directory where the output is written.
"""
self._output_filepath = os.path.join(output_path, 'score-{}.txt'.format(
self.NAME))
try:
# If the score has already been computed, load.
self._load_score()
self._LoadScore()
logging.debug('score found and loaded')
except IOError:
# Compute the score.
logging.debug('score not found, compute')
self._run(output_path)
self._Run(output_path)
def _run(self, output_path):
def _Run(self, output_path):
# Abstract method.
raise NotImplementedError()
def _load_score(self):
return data_access.ScoreFile.load(self._output_filepath)
def _LoadReferenceSignal(self):
assert self._reference_signal_filepath is not None
self._reference_signal = signal_processing.SignalProcessingUtils.LoadWav(
self._reference_signal_filepath)
def _save_score(self):
return data_access.ScoreFile.save(self._output_filepath, self._score)
def _LoadTestedSignal(self):
assert self._tested_signal_filepath is not None
self._tested_signal = signal_processing.SignalProcessingUtils.LoadWav(
self._tested_signal_filepath)
@EvaluationScore.register_class
def _LoadScore(self):
return data_access.ScoreFile.Load(self._output_filepath)
def _SaveScore(self):
return data_access.ScoreFile.Save(self._output_filepath, self._score)
@EvaluationScore.RegisterClass
class AudioLevelScore(EvaluationScore):
"""Compute the difference between the average audio level of the tested and
"""Audio level score.
Defined as the difference between the average audio level of the tested and
the reference signals.
Unit: dB
@ -109,16 +126,18 @@ class AudioLevelScore(EvaluationScore):
def __init__(self):
EvaluationScore.__init__(self)
def _run(self, output_path):
self._load_reference_signal()
self._load_tested_signal()
def _Run(self, output_path):
self._LoadReferenceSignal()
self._LoadTestedSignal()
self._score = self._tested_signal.dBFS - self._reference_signal.dBFS
self._save_score()
self._SaveScore()
@EvaluationScore.register_class
@EvaluationScore.RegisterClass
class PolqaScore(EvaluationScore):
"""Compute the POLQA score.
"""POLQA score.
See http://www.polqa.info/.
Unit: MOS
Ideal: 4.5
@ -141,7 +160,7 @@ class PolqaScore(EvaluationScore):
logging.error('cannot find POLQA tool binary file')
raise exceptions.FileNotFoundError()
def _run(self, output_path):
def _Run(self, output_path):
polqa_out_filepath = os.path.join(output_path, 'polqa.out')
if os.path.exists(polqa_out_filepath):
os.unlink(polqa_out_filepath)
@ -157,15 +176,21 @@ class PolqaScore(EvaluationScore):
subprocess.call(args, cwd=self._polqa_tool_path)
# Parse POLQA tool output and extract the score.
polqa_output = self._parse_output_file(polqa_out_filepath)
polqa_output = self._ParseOutputFile(polqa_out_filepath)
self._score = float(polqa_output['PolqaScore'])
self._save_score()
self._SaveScore()
@classmethod
def _parse_output_file(cls, polqa_out_filepath):
def _ParseOutputFile(cls, polqa_out_filepath):
"""
Parse the POLQA tool output formatted as a table ('-t' option).
Parses the POLQA tool output formatted as a table ('-t' option).
Args:
polqa_out_filepath: path to the POLQA tool output file.
Returns:
A dict.
"""
data = []
with open(polqa_out_filepath) as f:

6
webrtc/modules/audio_processing/test/py_quality_assessment/quality_assessment/eval_scores_factory.py
View File

@ -17,8 +17,7 @@ from . import eval_scores
class EvaluationScoreWorkerFactory(object):
"""Factory class used to instantiate evaluation score workers.
It can be used by instanciating a factory, passing parameters to the
constructor. These parameters are used to instantiate evaluation score
The ctor gets the parametrs that are used to instatiate the evaluation score
workers.
"""
@ -27,6 +26,9 @@ class EvaluationScoreWorkerFactory(object):
def GetInstance(self, evaluation_score_class):
"""Creates an EvaluationScore instance given a class object.
Args:
evaluation_score_class: EvaluationScore class object (not an instance).
"""
logging.debug(
'factory producing a %s evaluation score', evaluation_score_class)

4
webrtc/modules/audio_processing/test/py_quality_assessment/quality_assessment/eval_scores_unittest.py
View File

@ -6,7 +6,7 @@
# in the file PATENTS. All contributing project authors may
# be found in the AUTHORS file in the root of the source tree.
"""Unit tests for the evaluation scores.
"""Unit tests for the eval_scores module.
"""
import unittest
@ -15,6 +15,8 @@ from . import eval_scores
class TestEvalScores(unittest.TestCase):
"""Unit tests for the eval_scores module.
"""
def test_registered_classes(self):
# Check that there is at least one registered evaluation score worker.

19
webrtc/modules/audio_processing/test/py_quality_assessment/quality_assessment/evaluation.py
View File

@ -13,28 +13,39 @@ import logging
class ApmModuleEvaluator(object):
"""APM evaluator class.
"""
def __init__(self):
pass
@classmethod
def run(cls, evaluation_score_workers, apm_output_filepath,
def Run(cls, evaluation_score_workers, apm_output_filepath,
reference_input_filepath, output_path):
"""Runs the evaluation.
Iterates over the given evaluation score workers.
Args:
evaluation_score_workers: list of EvaluationScore instances.
apm_output_filepath: path to the audio track file with the APM output.
reference_input_filepath: path to the reference audio track file.
output_path: output path.
Returns:
A dict of evaluation score name and score pairs.
"""
# Init.
scores = {}
for evaluation_score_worker in evaluation_score_workers:
logging.info(' computing <%s> score', evaluation_score_worker.NAME)
evaluation_score_worker.set_reference_signal_filepath(
evaluation_score_worker.SetReferenceSignalFilepath(
reference_input_filepath)
evaluation_score_worker.set_tested_signal_filepath(
evaluation_score_worker.SetTestedSignalFilepath(
apm_output_filepath)
evaluation_score_worker.run(output_path)
evaluation_score_worker.Run(output_path)
scores[evaluation_score_worker.NAME] = evaluation_score_worker.score
return scores

4
webrtc/modules/audio_processing/test/py_quality_assessment/quality_assessment/exceptions.py
View File

@ -11,8 +11,12 @@
class FileNotFoundError(Exception):
"""File not found exeception.
"""
pass
class SignalProcessingException(Exception):
"""Signal processing exeception.
"""
pass

135
webrtc/modules/audio_processing/test/py_quality_assessment/quality_assessment/export.py
View File

@ -12,6 +12,8 @@ import re
class HtmlExport(object):
"""HTML exporter class for APM quality scores.
"""
# Path to CSS and JS files.
_PATH = os.path.dirname(os.path.realpath(__file__))
@ -31,9 +33,8 @@ class HtmlExport(object):
self._noise_params = None
self._output_filepath = output_filepath
def export(self, scores):
"""
Export the scores into an HTML file.
def Export(self, scores):
"""Exports the scores into an HTML file.
Args:
scores: nested dictionary containing the scores.
@ -41,29 +42,33 @@ class HtmlExport(object):
# Generate one table for each evaluation score.
tables = []
for score_name in sorted(scores.keys()):
tables.append(self._build_score_table(score_name, scores[score_name]))
tables.append(self._BuildScoreTable(score_name, scores[score_name]))
# Create the html file.
html = (
'<html>' +
self._build_header() +
self._BuildHeader() +
'<body onload="initialize()">' +
'<h1>Results from {}</h1>'.format(self._output_filepath) +
self._NEW_LINE.join(tables) +
'</body>' +
'</html>')
self._save(self._output_filepath, html)
self._Save(self._output_filepath, html)
def _build_header(self):
"""
HTML file header with page title and either embedded or linked CSS and JS
def _BuildHeader(self):
"""Builds the <head> section of the HTML file.
The header contains the page title and either embedded or linked CSS and JS
files.
Returns:
A string with <head>...</head> HTML.
"""
html = ['<head>', '<title>Results</title>']
# Function to append the lines of a text file to html.
def _embed_file(filepath):
def EmbedFile(filepath):
with open(filepath) as f:
for l in f:
html.append(l.strip())
@ -72,7 +77,7 @@ class HtmlExport(object):
if self._INLINE_CSS:
# Embed.
html.append('<style>')
_embed_file(self._CSS_FILEPATH)
EmbedFile(self._CSS_FILEPATH)
html.append('</style>')
else:
# Link.
@ -83,7 +88,7 @@ class HtmlExport(object):
if self._INLINE_JS:
# Embed.
html.append('<script>')
_embed_file(self._JS_FILEPATH)
EmbedFile(self._JS_FILEPATH)
html.append('</script>')
else:
# Link.
@ -94,54 +99,88 @@ class HtmlExport(object):
return self._NEW_LINE.join(html)
def _build_score_table(self, score_name, scores):
"""
Generate a table for a specific evaluation score (e.g., POLQA).
def _BuildScoreTable(self, score_name, scores):
"""Builds a table for a specific evaluation score (e.g., POLQA).
Args:
score_name: name of the score.
scores: nested dictionary of scores.
Returns:
A string with <table>...</table> HTML.
"""
config_names = sorted(scores.keys())
input_names = sorted(scores[config_names[0]].keys())
rows = [self._table_row(
rows = [self._BuildTableRow(
score_name, config_name, scores[config_name], input_names) for (
config_name) in config_names]
html = (
'<table celpadding="0" cellspacing="0">' +
'<thead><tr>{}</tr></thead>'.format(
self._table_header(score_name, input_names)) +
self._BuildTableHeader(score_name, input_names)) +
'<tbody>' +
'<tr>' + '</tr><tr>'.join(rows) + '</tr>' +
'</tbody>' +
'</table>' + self._legend())
'</table>' + self._BuildLegend())
return html
def _table_header(self, score_name, input_names):
"""
Generate a table header with the name of the evaluation score in the first
column and then one column for each probing signal.
def _BuildTableHeader(self, score_name, input_names):
"""Builds the cells of a table header.
A table header starts with a cell containing the name of the evaluation
score, and then it includes one column for each probing signal.
Args:
score_name: name of the score.
input_names: list of probing signal names.
Returns:
A string with a list of <th>...</th> HTML elements.
"""
html = (
'<th>{}</th>'.format(self._format_name(score_name)) +
'<th>{}</th>'.format(self._FormatName(score_name)) +
'<th>' + '</th><th>'.join(
[self._format_name(name) for name in input_names]) + '</th>')
[self._FormatName(name) for name in input_names]) + '</th>')
return html
def _table_row(self, score_name, config_name, scores, input_names):
def _BuildTableRow(self, score_name, config_name, scores, input_names):
"""Builds the cells of a table row.
A table row starts with the name of the APM configuration file, and then it
includes one column for each probing singal.
Args:
score_name: name of the score.
config_name: name of the APM configuration.
scores: nested dictionary of scores.
input_names: list of probing signal names.
Returns:
A string with a list of <td>...</td> HTML elements.
"""
Generate a table body row with the name of the APM configuration file in the
first column and then one column for each probing singal.
"""
cells = [self._table_cell(
cells = [self._BuildTableCell(
scores[input_name], score_name, config_name, input_name) for (
input_name) in input_names]
html = ('<td>{}</td>'.format(self._format_name(config_name)) +
html = ('<td>{}</td>'.format(self._FormatName(config_name)) +
'<td>' + '</td><td>'.join(cells) + '</td>')
return html
def _table_cell(self, scores, score_name, config_name, input_name):
"""
Generate a table cell content with all the scores for the current evaluation
score, APM configuration, and probing signal.
def _BuildTableCell(self, scores, score_name, config_name, input_name):
"""Builds the inner content of a table cell.
A table cell includes all the scores computed for a specific evaluation
score (e.g., POLQA), APM configuration (e.g., default), and probing signal.
Args:
scores: dictionary of score data.
score_name: name of the score.
config_name: name of the APM configuration.
input_name: name of the probing signal.
Returns:
A string with the HTML of a table body cell.
"""
# Init noise generator names and noise parameters cache (if not done).
if self._noise_names is None:
@ -195,9 +234,13 @@ class HtmlExport(object):
return html
def _legend(self):
"""
Generate the legend for each noise generator name and parameters pair.
def _BuildLegend(self):
"""Builds the legend.
The legend details noise generator name and parameter pairs.
Returns:
A string with a <div class="legend">...</div> HTML element.
"""
items = []
for name_index, noise_name in enumerate(self._noise_names):
@ -213,10 +256,24 @@ class HtmlExport(object):
return html
@classmethod
def _save(cls, output_filepath, html):
def _Save(cls, output_filepath, html):
"""Writes the HTML file.
Args:
output_filepath: output file path.
html: string with the HTML content.
"""
with open(output_filepath, 'w') as f:
f.write(html)
@classmethod
def _format_name(cls, name):
def _FormatName(cls, name):
"""Formats a name.
Args:
name: a string.
Returns:
A copy of name in which underscores and dashes are replaced with a space.
"""
return re.sub(r'[_\-]', ' ', name)

171
webrtc/modules/audio_processing/test/py_quality_assessment/quality_assessment/noise_generation.py
View File

@ -47,23 +47,34 @@ class NoiseGenerator(object):
signal.
A noise generator generates one or more input-reference pairs.
TODO(alessiob): Rename from NoiseGenerator to InputReferencePairGenerator.
"""
NAME = None
REGISTERED_CLASSES = {}
def __init__(self):
# Input
# Init dictionaries with one entry for each noise generator configuration
# (e.g., different SNRs).
# Noisy audio track files (stored separately in a cache folder).
self._noisy_signal_filepaths = None
self._output_paths = None
# Path to be used for the APM simulation output files.
self._apm_output_paths = None
# Reference audio track files (stored separately in a cache folder).
self._reference_signal_filepaths = None
self.clear()
self.Clear()
@classmethod
def register_class(cls, class_to_register):
"""Register an NoiseGenerator implementation.
def RegisterClass(cls, class_to_register):
"""Registers an NoiseGenerator implementation.
Decorator to automatically register the classes that extend NoiseGenerator.
Example usage:
@NoiseGenerator.RegisterClass
class IdentityGenerator(NoiseGenerator):
pass
"""
cls.REGISTERED_CLASSES[class_to_register.NAME] = class_to_register
return class_to_register
@ -77,37 +88,44 @@ class NoiseGenerator(object):
return self._noisy_signal_filepaths
@property
def output_paths(self):
return self._output_paths
def apm_output_paths(self):
return self._apm_output_paths
@property
def reference_signal_filepaths(self):
return self._reference_signal_filepaths
def generate(
def Generate(
self, input_signal_filepath, input_noise_cache_path, base_output_path):
"""Generate a set of noisy input and reference audiotrack file pairs.
"""Generates a set of noisy input and reference audiotrack file pairs.
This method initializes an empty set of pairs and calls the _generate()
This method initializes an empty set of pairs and calls the _Generate()
method implemented in a concrete class.
Args:
input_signal_filepath: path to the clean input audio track file.
input_noise_cache_path: path to the cache of noisy audio track files.
base_output_path: base path where output is written.
"""
self.clear()
return self._generate(
self.Clear()
self._Generate(
input_signal_filepath, input_noise_cache_path, base_output_path)
def clear(self):
def Clear(self):
"""Clears the generated output path dictionaries.
"""
self._noisy_signal_filepaths = {}
self._output_paths = {}
self._apm_output_paths = {}
self._reference_signal_filepaths = {}
def _generate(
def _Generate(
self, input_signal_filepath, input_noise_cache_path, base_output_path):
"""This is an abstract method to be implemented in each concrete class.
"""Abstract method to be implemented in each concrete class.
"""
raise NotImplementedError()
def _add_noise_snr_pairs(self, base_output_path, noisy_mix_filepaths,
snr_value_pairs):
def _AddNoiseSnrPairs(self, base_output_path, noisy_mix_filepaths,
snr_value_pairs):
"""Adds noisy-reference signal pairs.
Args:
@ -120,8 +138,8 @@ class NoiseGenerator(object):
for snr_noisy, snr_refence in snr_value_pairs:
config_name = '{0}_{1:d}_{2:d}_SNR'.format(
noise_track_name, snr_noisy, snr_refence)
output_path = self._make_dir(base_output_path, config_name)
self._add_noise_reference_files_pair(
output_path = self._MakeDir(base_output_path, config_name)
self._AddNoiseReferenceFilesPair(
config_name=config_name,
noisy_signal_filepath=noisy_mix_filepaths[
noise_track_name][snr_noisy],
@ -129,30 +147,38 @@ class NoiseGenerator(object):
noise_track_name][snr_refence],
output_path=output_path)
def _add_noise_reference_files_pair(self, config_name, noisy_signal_filepath,
reference_signal_filepath, output_path):
def _AddNoiseReferenceFilesPair(self, config_name, noisy_signal_filepath,
reference_signal_filepath, output_path):
"""Adds one noisy-reference signal pair.
Args:
config_name: name of the APM configuration.
noisy_signal_filepath: path to noisy audio track file.
reference_signal_filepath: path to reference audio track file.
output_path: APM output path.
"""
assert config_name not in self._noisy_signal_filepaths
self._noisy_signal_filepaths[config_name] = os.path.abspath(
noisy_signal_filepath)
self._output_paths[config_name] = os.path.abspath(output_path)
self._apm_output_paths[config_name] = os.path.abspath(output_path)
self._reference_signal_filepaths[config_name] = os.path.abspath(
reference_signal_filepath)
# Save noisy and reference file paths.
data_access.Metadata.save_audio_in_ref_paths(
data_access.Metadata.SaveAudioInRefPaths(
output_path=output_path,
audio_in_filepath=self._noisy_signal_filepaths[config_name],
audio_ref_filepath=self._reference_signal_filepaths[config_name])
@classmethod
def _make_dir(cls, base_output_path, noise_generator_config_name):
def _MakeDir(cls, base_output_path, noise_generator_config_name):
output_path = os.path.join(base_output_path, noise_generator_config_name)
data_access.make_directory(output_path)
data_access.MakeDirectory(output_path)
return output_path
# Identity generator.
@NoiseGenerator.register_class
@NoiseGenerator.RegisterClass
class IdentityGenerator(NoiseGenerator):
"""Generator that adds no noise.
@ -164,18 +190,18 @@ class IdentityGenerator(NoiseGenerator):
def __init__(self):
NoiseGenerator.__init__(self)
def _generate(
def _Generate(
self, input_signal_filepath, input_noise_cache_path, base_output_path):
CONFIG_NAME = 'default'
output_path = self._make_dir(base_output_path, CONFIG_NAME)
self._add_noise_reference_files_pair(
output_path = self._MakeDir(base_output_path, CONFIG_NAME)
self._AddNoiseReferenceFilesPair(
config_name=CONFIG_NAME,
noisy_signal_filepath=input_signal_filepath,
reference_signal_filepath=input_signal_filepath,
output_path=output_path)
@NoiseGenerator.register_class
@NoiseGenerator.RegisterClass
class WhiteNoiseGenerator(NoiseGenerator):
"""Additive white noise generator.
"""
@ -197,18 +223,18 @@ class WhiteNoiseGenerator(NoiseGenerator):
def __init__(self):
NoiseGenerator.__init__(self)
def _generate(
def _Generate(
self, input_signal_filepath, input_noise_cache_path, base_output_path):
# Load the input signal.
input_signal = signal_processing.SignalProcessingUtils.load_wav(
input_signal = signal_processing.SignalProcessingUtils.LoadWav(
input_signal_filepath)
input_signal = signal_processing.SignalProcessingUtils.normalize(
input_signal = signal_processing.SignalProcessingUtils.Normalize(
input_signal)
# Create the noise track.
noise_signal = signal_processing.SignalProcessingUtils.generate_white_noise(
noise_signal = signal_processing.SignalProcessingUtils.GenerateWhiteNoise(
input_signal)
noise_signal = signal_processing.SignalProcessingUtils.normalize(
noise_signal = signal_processing.SignalProcessingUtils.Normalize(
noise_signal)
# Create the noisy mixes (once for each unique SNR value).
@ -222,11 +248,11 @@ class WhiteNoiseGenerator(NoiseGenerator):
# Create and save if not done.
if not os.path.exists(noisy_signal_filepath):
# Create noisy signal.
noisy_signal = signal_processing.SignalProcessingUtils.mix_signals(
noisy_signal = signal_processing.SignalProcessingUtils.MixSignals(
input_signal, noise_signal, snr)
# Save.
signal_processing.SignalProcessingUtils.save_wav(
signal_processing.SignalProcessingUtils.SaveWav(
noisy_signal_filepath, noisy_signal)
# Add file to the collection of mixes.
@ -235,8 +261,8 @@ class WhiteNoiseGenerator(NoiseGenerator):
# Add all the noisy-reference signal pairs.
for snr_noisy, snr_refence in self._SNR_VALUE_PAIRS:
config_name = '{0:d}_{1:d}_SNR'.format(snr_noisy, snr_refence)
output_path = self._make_dir(base_output_path, config_name)
self._add_noise_reference_files_pair(
output_path = self._MakeDir(base_output_path, config_name)
self._AddNoiseReferenceFilesPair(
config_name=config_name,
noisy_signal_filepath=noisy_mix_filepaths[snr_noisy],
reference_signal_filepath=noisy_mix_filepaths[snr_refence],
@ -244,7 +270,7 @@ class WhiteNoiseGenerator(NoiseGenerator):
# TODO(alessiob): remove comment when class implemented.
# @NoiseGenerator.register_class
# @NoiseGenerator.RegisterClass
class NarrowBandNoiseGenerator(NoiseGenerator):
"""Additive narrow-band noise generator.
"""
@ -254,13 +280,13 @@ class NarrowBandNoiseGenerator(NoiseGenerator):
def __init__(self):
NoiseGenerator.__init__(self)
def _generate(
def _Generate(
self, input_signal_filepath, input_noise_cache_path, base_output_path):
# TODO(alessiob): implement.
pass
@NoiseGenerator.register_class
@NoiseGenerator.RegisterClass
class EnvironmentalNoiseGenerator(NoiseGenerator):
"""Additive environmental noise generator.
"""
@ -290,27 +316,22 @@ class EnvironmentalNoiseGenerator(NoiseGenerator):
def __init__(self):
NoiseGenerator.__init__(self)
def _generate(
def _Generate(
self, input_signal_filepath, input_noise_cache_path, base_output_path):
"""Generate environmental noise.
"""Generates environmental noise.
For each noise track and pair of SNR values, the following 2 audio tracks
For each noise track and pair of SNR values, the following two audio tracks
are created: the noisy signal and the reference signal. The former is
obtained by mixing the (clean) input signal to the corresponding noise
track enforcing the target SNR.
Args:
input_signal_filepath: (clean) input signal file path.
input_noise_cache_path: path for the cached noise track files.
base_output_path: base output path.
"""
# Init.
snr_values = set([snr for pair in self._SNR_VALUE_PAIRS for snr in pair])
# Load the input signal.
input_signal = signal_processing.SignalProcessingUtils.load_wav(
input_signal = signal_processing.SignalProcessingUtils.LoadWav(
input_signal_filepath)
input_signal = signal_processing.SignalProcessingUtils.normalize(
input_signal = signal_processing.SignalProcessingUtils.Normalize(
input_signal)
noisy_mix_filepaths = {}
@ -323,9 +344,9 @@ class EnvironmentalNoiseGenerator(NoiseGenerator):
logging.error('cannot find the <%s> noise track', noise_track_filename)
raise exceptions.FileNotFoundError()
noise_signal = signal_processing.SignalProcessingUtils.load_wav(
noise_signal = signal_processing.SignalProcessingUtils.LoadWav(
noise_track_filepath)
noise_signal = signal_processing.SignalProcessingUtils.normalize(
noise_signal = signal_processing.SignalProcessingUtils.Normalize(
noise_signal)
# Create the noisy mixes (once for each unique SNR value).
@ -338,24 +359,26 @@ class EnvironmentalNoiseGenerator(NoiseGenerator):
# Create and save if not done.
if not os.path.exists(noisy_signal_filepath):
# Create noisy signal.
noisy_signal = signal_processing.SignalProcessingUtils.mix_signals(
noisy_signal = signal_processing.SignalProcessingUtils.MixSignals(
input_signal, noise_signal, snr)
# Save.
signal_processing.SignalProcessingUtils.save_wav(
signal_processing.SignalProcessingUtils.SaveWav(
noisy_signal_filepath, noisy_signal)
# Add file to the collection of mixes.
noisy_mix_filepaths[noise_track_name][snr] = noisy_signal_filepath
# Add all the noise-SNR pairs.
self._add_noise_snr_pairs(
self._AddNoiseSnrPairs(
base_output_path, noisy_mix_filepaths, self._SNR_VALUE_PAIRS)
@NoiseGenerator.register_class
@NoiseGenerator.RegisterClass
class EchoNoiseGenerator(NoiseGenerator):
"""Echo noise generator.
TODO(alessiob): Rename from echo to reverberation.
"""
NAME = 'echo'
@ -381,7 +404,7 @@ class EchoNoiseGenerator(NoiseGenerator):
NoiseGenerator.__init__(self)
self._aechen_ir_database_path = aechen_ir_database_path
def _generate(
def _Generate(
self, input_signal_filepath, input_noise_cache_path, base_output_path):
"""Generates echo noise.
@ -390,17 +413,12 @@ class EchoNoiseGenerator(NoiseGenerator):
created: the noisy signal and the reference signal. The former is
obtained by mixing the (clean) input signal to the corresponding noise
track enforcing the target SNR.
Args:
input_signal_filepath: (clean) input signal file path.
input_noise_cache_path: path for the cached noise track files.
base_output_path: base output path.
"""
# Init.
snr_values = set([snr for pair in self._SNR_VALUE_PAIRS for snr in pair])
# Load the input signal.
input_signal = signal_processing.SignalProcessingUtils.load_wav(
input_signal = signal_processing.SignalProcessingUtils.LoadWav(
input_signal_filepath)
noisy_mix_filepaths = {}
@ -412,14 +430,14 @@ class EchoNoiseGenerator(NoiseGenerator):
noise_signal = None
try:
# Load noise track.
noise_signal = signal_processing.SignalProcessingUtils.load_wav(
noise_signal = signal_processing.SignalProcessingUtils.LoadWav(
noise_track_filepath)
except IOError: # File not found.
# Generate noise track by applying the impulse response.
impulse_response_filepath = os.path.join(
self._aechen_ir_database_path,
self._IMPULSE_RESPONSES[impulse_response_name])
noise_signal = self._generate_noise_track(
noise_signal = self._GenerateNoiseTrack(
noise_track_filepath, input_signal, impulse_response_filepath)
assert noise_signal is not None
@ -434,21 +452,21 @@ class EchoNoiseGenerator(NoiseGenerator):
# Create and save if not done.
if not os.path.exists(noisy_signal_filepath):
# Create noisy signal.
noisy_signal = signal_processing.SignalProcessingUtils.mix_signals(
noisy_signal = signal_processing.SignalProcessingUtils.MixSignals(
input_signal, noise_signal, snr, bln_pad_shortest=True)
# Save.
signal_processing.SignalProcessingUtils.save_wav(
signal_processing.SignalProcessingUtils.SaveWav(
noisy_signal_filepath, noisy_signal)
# Add file to the collection of mixes.
noisy_mix_filepaths[impulse_response_name][snr] = noisy_signal_filepath
# Add all the noise-SNR pairs.
self._add_noise_snr_pairs(base_output_path, noisy_mix_filepaths,
self._SNR_VALUE_PAIRS)
self._AddNoiseSnrPairs(base_output_path, noisy_mix_filepaths,
self._SNR_VALUE_PAIRS)
def _generate_noise_track(self, noise_track_filepath, input_signal,
def _GenerateNoiseTrack(self, noise_track_filepath, input_signal,
impulse_response_filepath):
"""Generates noise track.
@ -459,6 +477,9 @@ class EchoNoiseGenerator(NoiseGenerator):
noise_track_filepath: output file path for the noise track.
input_signal: (clean) input signal samples.
impulse_response_filepath: impulse response file path.
Returns:
AudioSegment instance.
"""
# Load impulse response.
data = scipy.io.loadmat(impulse_response_filepath)
@ -470,11 +491,11 @@ class EchoNoiseGenerator(NoiseGenerator):
# Apply impulse response.
processed_signal = (
signal_processing.SignalProcessingUtils.apply_impulse_response(
signal_processing.SignalProcessingUtils.ApplyImpulseResponse(
input_signal, impulse_response))
# Save.
signal_processing.SignalProcessingUtils.save_wav(
signal_processing.SignalProcessingUtils.SaveWav(
noise_track_filepath, processed_signal)
return processed_signal

9
webrtc/modules/audio_processing/test/py_quality_assessment/quality_assessment/noise_generation_factory.py
View File

@ -17,9 +17,9 @@ from . import noise_generation
class NoiseGeneratorFactory(object):
"""Factory class used to instantiate noise generator workers.
It can be used by instanciating a factory, passing parameters to the
constructor. These parameters are used to instantiate noise generator
workers.
It can be used by instanciating a factory, passing parameters to the
constructor. These parameters are used to instantiate noise generator
workers.
"""
def __init__(self, aechen_ir_database_path):
@ -27,6 +27,9 @@ class NoiseGeneratorFactory(object):
def GetInstance(self, noise_generator_class):
"""Creates an NoiseGenerator instance given a class object.
Args:
noise_generator_class: NoiseGenerator class object (not an instance).
"""
logging.debug(
'factory producing a %s noise generator', noise_generator_class)

20
webrtc/modules/audio_processing/test/py_quality_assessment/quality_assessment/noise_generation_unittest.py
View File

@ -20,6 +20,8 @@ from . import signal_processing
class TestNoiseGen(unittest.TestCase):
"""Unit tests for the noise_generation module.
"""
def setUp(self):
"""Create temporary folders."""
@ -54,7 +56,7 @@ class TestNoiseGen(unittest.TestCase):
self.assertTrue(os.path.exists(input_signal_filepath))
# Load input signal.
input_signal = signal_processing.SignalProcessingUtils.load_wav(
input_signal = signal_processing.SignalProcessingUtils.LoadWav(
input_signal_filepath)
# Try each registered noise generator.
@ -72,7 +74,7 @@ class TestNoiseGen(unittest.TestCase):
registered_classes[noise_generator_name])
# Generate the noisy input - reference pairs.
noise_generator.generate(
noise_generator.Generate(
input_signal_filepath=input_signal_filepath,
input_noise_cache_path=self._input_noise_cache_path,
base_output_path=self._base_output_path)
@ -92,7 +94,7 @@ class TestNoiseGen(unittest.TestCase):
self.assertEqual(number_of_pairs,
len(noise_generator.noisy_signal_filepaths))
self.assertEqual(number_of_pairs,
len(noise_generator.output_paths))
len(noise_generator.apm_output_paths))
self.assertEqual(number_of_pairs,
len(noise_generator.reference_signal_filepaths))
@ -108,30 +110,30 @@ class TestNoiseGen(unittest.TestCase):
input_signal: AudioSegment instance.
"""
input_signal_length = (
signal_processing.SignalProcessingUtils.count_samples(input_signal))
signal_processing.SignalProcessingUtils.CountSamples(input_signal))
# Iterate over the noisy signal - reference pairs.
for noise_config_name in noise_generator.config_names:
# Load the noisy input file.
noisy_signal_filepath = noise_generator.noisy_signal_filepaths[
noise_config_name]
noisy_signal = signal_processing.SignalProcessingUtils.load_wav(
noisy_signal = signal_processing.SignalProcessingUtils.LoadWav(
noisy_signal_filepath)
# Check noisy input signal length.
noisy_signal_length = (
signal_processing.SignalProcessingUtils.count_samples(noisy_signal))
signal_processing.SignalProcessingUtils.CountSamples(noisy_signal))
self.assertGreaterEqual(noisy_signal_length, input_signal_length)
# Load the reference file.
reference_signal_filepath = (
noise_generator.reference_signal_filepaths[noise_config_name])
reference_signal = signal_processing.SignalProcessingUtils.load_wav(
reference_signal = signal_processing.SignalProcessingUtils.LoadWav(
reference_signal_filepath)
# Check noisy input signal length.
reference_signal_length = (
signal_processing.SignalProcessingUtils.count_samples(
signal_processing.SignalProcessingUtils.CountSamples(
reference_signal))
self.assertGreaterEqual(reference_signal_length, input_signal_length)
@ -143,5 +145,5 @@ class TestNoiseGen(unittest.TestCase):
"""
# Iterate over the noisy signal - reference pairs.
for noise_config_name in noise_generator.config_names:
output_path = noise_generator.output_paths[noise_config_name]
output_path = noise_generator.apm_output_paths[noise_config_name]
self.assertTrue(os.path.exists(output_path))

79
webrtc/modules/audio_processing/test/py_quality_assessment/quality_assessment/signal_processing.py
View File

@ -6,6 +6,9 @@
# in the file PATENTS. All contributing project authors may
# be found in the AUTHORS file in the root of the source tree.
"""Signal processing utility module.
"""
import array
import logging
import os
@ -34,15 +37,21 @@ from . import exceptions
class SignalProcessingUtils(object):
"""Collection of signal processing utilities.
"""
def __init__(self):
pass
@classmethod
def load_wav(cls, filepath, channels=1):
"""Load wav file.
def LoadWav(cls, filepath, channels=1):
"""Loads wav file.
Return:
Args:
filepath: path to the wav audio track file to load.
channels: number of channels (downmixing to mono by default).
Returns:
AudioSegment instance.
"""
if not os.path.exists(filepath):
@ -52,21 +61,24 @@ class SignalProcessingUtils(object):
filepath, format='wav', channels=channels)
@classmethod
def save_wav(cls, output_filepath, signal):
"""Save wav file.
def SaveWav(cls, output_filepath, signal):
"""Saves wav file.
Args:
output_filepath: string, output file path.
output_filepath: path to the wav audio track file to save.
signal: AudioSegment instance.
"""
return signal.export(output_filepath, format='wav')
@classmethod
def count_samples(cls, signal):
def CountSamples(cls, signal):
"""Number of samples per channel.
Args:
signal: AudioSegment instance.
Returns:
An integer.
"""
number_of_samples = len(signal.get_array_of_samples())
assert signal.channels > 0
@ -74,10 +86,10 @@ class SignalProcessingUtils(object):
return number_of_samples / signal.channels
@classmethod
def generate_white_noise(cls, signal):
"""Generate white noise.
def GenerateWhiteNoise(cls, signal):
"""Generates white noise.
Generate white noise with the same duration and in the same format as a
White noise is generated with the same duration and in the same format as a
given signal.
Args:
@ -94,8 +106,15 @@ class SignalProcessingUtils(object):
volume=0.0)
@classmethod
def apply_impulse_response(cls, signal, impulse_response):
"""Apply an impulse response to a signal.
def ApplyImpulseResponse(cls, signal, impulse_response):
"""Applies an impulse response to a signal.
Args:
signal: AudioSegment instance.
impulse_response: list or numpy vector of float values.
Returns:
AudioSegment instance.
"""
# Get samples.
assert signal.channels == 1, (
@ -133,11 +152,27 @@ class SignalProcessingUtils(object):
return convolved_signal
@classmethod
def normalize(cls, signal):
def Normalize(cls, signal):
"""Normalizes a signal.
Args:
signal: AudioSegment instance.
Returns:
An AudioSegment instance.
"""
return signal.apply_gain(-signal.max_dBFS)
@classmethod
def copy(cls, signal):
def Copy(cls, signal):
"""Makes a copy os a signal.
Args:
signal: AudioSegment instance.
Returns:
An AudioSegment instance.
"""
return pydub.AudioSegment(
data=signal.get_array_of_samples(),
metadata={
@ -148,11 +183,10 @@ class SignalProcessingUtils(object):
})
@classmethod
def mix_signals(cls, signal, noise, target_snr=0.0,
bln_pad_shortest=False):
"""Mix two signals with a target SNR.
def MixSignals(cls, signal, noise, target_snr=0.0, bln_pad_shortest=False):
"""Mixes two signals with a target SNR.
Mix two signals up to a desired SNR by scaling noise (noise).
Mix two signals with a desired SNR by scaling noise (noise).
If the target SNR is +/- infinite, a copy of signal/noise is returned.
Args:
@ -161,16 +195,19 @@ class SignalProcessingUtils(object):
target_snr: float, numpy.Inf or -numpy.Inf (dB).
bln_pad_shortest: if True, it pads the shortest signal with silence at the
end.
Returns:
An AudioSegment instance.
"""
# Handle infinite target SNR.
if target_snr == -np.Inf:
# Return a copy of noise.
logging.warning('SNR = -Inf, returning noise')
return cls.copy(noise)
return cls.Copy(noise)
elif target_snr == np.Inf:
# Return a copy of signal.
logging.warning('SNR = +Inf, returning signal')
return cls.copy(signal)
return cls.Copy(signal)
# Check signal and noise power.
signal_power = float(signal.dBFS)
@ -208,4 +245,4 @@ class SignalProcessingUtils(object):
# Mix signals using the target SNR.
gain_db = signal_power - noise_power - target_snr
return cls.normalize(signal.overlay(noise.apply_gain(gain_db)))
return cls.Normalize(signal.overlay(noise.apply_gain(gain_db)))

18
webrtc/modules/audio_processing/test/py_quality_assessment/quality_assessment/signal_processing_unittest.py
View File

@ -19,15 +19,17 @@ from . import signal_processing
class TestSignalProcessing(unittest.TestCase):
"""Unit tests for the signal_processing module.
"""
def testMixSignals(self):
# Generate a template signal with which white noise can be generated.
silence = pydub.AudioSegment.silent(duration=1000, frame_rate=48000)
# Generate two distinct AudioSegment instances with 1 second of white noise.
signal = signal_processing.SignalProcessingUtils.generate_white_noise(
signal = signal_processing.SignalProcessingUtils.GenerateWhiteNoise(
silence)
noise = signal_processing.SignalProcessingUtils.generate_white_noise(
noise = signal_processing.SignalProcessingUtils.GenerateWhiteNoise(
silence)
# Extract samples.
@ -35,7 +37,7 @@ class TestSignalProcessing(unittest.TestCase):
noise_samples = noise.get_array_of_samples()
# Test target SNR -Inf (noise expected).
mix_neg_inf = signal_processing.SignalProcessingUtils.mix_signals(
mix_neg_inf = signal_processing.SignalProcessingUtils.MixSignals(
signal, noise, -np.Inf)
self.assertTrue(len(noise), len(mix_neg_inf)) # Check duration.
mix_neg_inf_samples = mix_neg_inf.get_array_of_samples()
@ -43,7 +45,7 @@ class TestSignalProcessing(unittest.TestCase):
all([x == y for x, y in zip(noise_samples, mix_neg_inf_samples)]))
# Test target SNR 0.0 (different data expected).
mix_0 = signal_processing.SignalProcessingUtils.mix_signals(
mix_0 = signal_processing.SignalProcessingUtils.MixSignals(
signal, noise, 0.0)
self.assertTrue(len(signal), len(mix_0)) # Check duration.
self.assertTrue(len(noise), len(mix_0))
@ -54,7 +56,7 @@ class TestSignalProcessing(unittest.TestCase):
any([x != y for x, y in zip(noise_samples, mix_0_samples)]))
# Test target SNR +Inf (signal expected).
mix_pos_inf = signal_processing.SignalProcessingUtils.mix_signals(
mix_pos_inf = signal_processing.SignalProcessingUtils.MixSignals(
signal, noise, np.Inf)
self.assertTrue(len(signal), len(mix_pos_inf)) # Check duration.
mix_pos_inf_samples = mix_pos_inf.get_array_of_samples()
@ -63,13 +65,13 @@ class TestSignalProcessing(unittest.TestCase):
def testMixSignalsMinInfPower(self):
silence = pydub.AudioSegment.silent(duration=1000, frame_rate=48000)
signal = signal_processing.SignalProcessingUtils.generate_white_noise(
signal = signal_processing.SignalProcessingUtils.GenerateWhiteNoise(
silence)
with self.assertRaises(exceptions.SignalProcessingException):
_ = signal_processing.SignalProcessingUtils.mix_signals(
_ = signal_processing.SignalProcessingUtils.MixSignals(
signal, silence, 0.0)
with self.assertRaises(exceptions.SignalProcessingException):
_ = signal_processing.SignalProcessingUtils.mix_signals(
_ = signal_processing.SignalProcessingUtils.MixSignals(
silence, signal, 0.0)

70
webrtc/modules/audio_processing/test/py_quality_assessment/quality_assessment/simulation.py
View File

@ -48,10 +48,18 @@ class ApmModuleSimulator(object):
self._config_filepaths = None
self._input_filepaths = None
def run(self, config_filepaths, input_filepaths, noise_generator_names,
def Run(self, config_filepaths, input_filepaths, noise_generator_names,
eval_score_names, output_dir):
"""
"""Runs the APM simulation.
Initializes paths and required instances, then runs all the simulations.
Args:
config_filepaths: set of APM configuration files to test.
input_filepaths: set of input audio track files to test.
noise_generator_names: set of noise generator names to test.
eval_score_names: set of evaluation score names to test.
output_dir: base path to the output directory for wav files and outcomes.
"""
self._base_output_path = os.path.abspath(output_dir)
@ -67,15 +75,16 @@ class ApmModuleSimulator(object):
name) in eval_score_names]
# Set APM configuration file paths.
self._config_filepaths = self._get_paths_collection(config_filepaths)
self._config_filepaths = self._CreatePathsCollection(config_filepaths)
# Set probing signal file paths.
self._input_filepaths = self._get_paths_collection(input_filepaths)
self._input_filepaths = self._CreatePathsCollection(input_filepaths)
self._simulate_all()
self._SimulateAll()
def _SimulateAll(self):
"""Runs all the simulations.
def _simulate_all(self):
"""
Iterates over the combinations of APM configurations, probing signals, and
noise generators.
"""
@ -98,7 +107,7 @@ class ApmModuleSimulator(object):
self._base_output_path,
'_cache',
'input_{}-noise_{}'.format(input_name, noise_generator.NAME))
data_access.make_directory(input_noise_cache_path)
data_access.MakeDirectory(input_noise_cache_path)
logging.debug('input-noise cache path: <%s>', input_noise_cache_path)
# Full output path.
@ -107,21 +116,29 @@ class ApmModuleSimulator(object):
'cfg-{}'.format(config_name),
'input-{}'.format(input_name),
'noise-{}'.format(noise_generator.NAME))
data_access.make_directory(output_path)
data_access.MakeDirectory(output_path)
logging.debug('output path: <%s>', output_path)
self._simulate(noise_generator, input_filepath,
self._Simulate(noise_generator, input_filepath,
input_noise_cache_path, output_path, config_filepath)
def _simulate(self, noise_generator, input_filepath, input_noise_cache_path,
def _Simulate(self, noise_generator, input_filepath, input_noise_cache_path,
output_path, config_filepath):
"""
Simulates a given combination of APM configurations, probing signals, and
noise generators. It iterates over the noise generator internal
"""Runs a single set of simulation.
Simulates a given combination of APM configuration, probing signal, and
noise generator. It iterates over the noise generator internal
configurations.
Args:
noise_generator: NoiseGenerator instance.
input_filepath: input audio track file to test.
input_noise_cache_path: path for the noisy audio track files.
output_path: base output path for the noise generator.
config_filepath: APM configuration file to test.
"""
# Generate pairs of noisy input and reference signal files.
noise_generator.generate(
noise_generator.Generate(
input_signal_filepath=input_filepath,
input_noise_cache_path=input_noise_cache_path,
base_output_path=output_path)
@ -133,11 +150,11 @@ class ApmModuleSimulator(object):
# APM input and output signal paths.
noisy_signal_filepath = noise_generator.noisy_signal_filepaths[
noise_generator_config_name]
evaluation_output_path = noise_generator.output_paths[
evaluation_output_path = noise_generator.apm_output_paths[
noise_generator_config_name]
# Simulate a call using the audio processing module.
self._audioproc_wrapper.run(
self._audioproc_wrapper.Run(
config_filepath=config_filepath,
input_filepath=noisy_signal_filepath,
output_path=evaluation_output_path)
@ -147,18 +164,25 @@ class ApmModuleSimulator(object):
noise_generator_config_name]
# Evaluate.
self._evaluator.run(
self._evaluator.Run(
evaluation_score_workers=self._evaluation_score_workers,
apm_output_filepath=self._audioproc_wrapper.output_filepath,
reference_input_filepath=reference_signal_filepath,
output_path=evaluation_output_path)
@classmethod
def _get_paths_collection(cls, filepaths):
"""
Given a list of file paths, makes a collection with one pair for each item
in the list where the key is the file name without extension and the value
is the path.
def _CreatePathsCollection(cls, filepaths):
"""Creates a collection of file paths.
Given a list of file paths, makes a collection with one item for each file
path. The value is absolute path, the key is the file name without
extenstion.
Args:
filepaths: list of file paths.
Returns:
A dict.
"""
filepaths_collection = {}
for filepath in filepaths: