blob: abe9da9260677aef8cb478efb1441a09c7909493 [file] [log] [blame]
#!/usr/bin/env python
# Copyright (c) 2017 The WebRTC project authors. All Rights Reserved.
# Use of this source code is governed by a BSD-style license
# that can be found in the LICENSE file in the root of the source
# tree. An additional intellectual property rights grant can be found
# in the file PATENTS. All contributing project authors may
# be found in the AUTHORS file in the root of the source tree.
This script is the wrapper that runs the low-bandwidth audio test.
After running the test, post-process steps for calculating audio quality of the
output files will be performed.
import argparse
import collections
import logging
import os
import re
import shutil
import subprocess
import sys
SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
SRC_DIR = os.path.normpath(os.path.join(SCRIPT_DIR, os.pardir, os.pardir,
def _LogCommand(command):'Running %r', command)
return command
def _ParseArgs():
parser = argparse.ArgumentParser(description='Run low-bandwidth audio tests.')
help='Path to the build directory (e.g. out/Release).')
parser.add_argument('--remove', action='store_true',
help='Remove output audio files after testing.')
parser.add_argument('--android', action='store_true',
help='Perform the test on a connected Android device instead.')
parser.add_argument('--adb-path', help='Path to adb binary.', default='adb')
args = parser.parse_args()
return args
def _GetPlatform():
if sys.platform == 'win32':
return 'win'
elif sys.platform == 'darwin':
return 'mac'
elif sys.platform.startswith('linux'):
return 'linux'
def _DownloadTools():
tools_dir = os.path.join(SRC_DIR, 'tools_webrtc')
toolchain_dir = os.path.join(tools_dir, 'audio_quality')
# Download PESQ and POLQA.
download_script = os.path.join(tools_dir, '')
command = [sys.executable, download_script, toolchain_dir]
pesq_path = os.path.join(toolchain_dir, _GetPlatform(), 'pesq')
polqa_path = os.path.join(toolchain_dir, _GetPlatform(), 'PolqaOem64')
return pesq_path, polqa_path
def ExtractTestRuns(lines, echo=False):
"""Extracts information about tests from the output of a test runner.
Produces tuples (android_device, test_name, reference_file, degraded_file).
for line in lines:
if echo:
# Output from Android has a prefix with the device name.
android_prefix_re = r'(?:I\b.+\brun_tests_on_device\((.+?)\)\s*)?'
test_re = r'^' + android_prefix_re + r'TEST (\w+) ([^ ]+?) ([^ ]+?)\s*$'
match =, line)
if match:
yield match.groups()
def _GetFile(file_path, out_dir, move=False,
android=False, adb_prefix=('adb',)):
out_file_name = os.path.basename(file_path)
out_file_path = os.path.join(out_dir, out_file_name)
if android:
# Pull the file from the connected Android device.
adb_command = adb_prefix + ('pull', file_path, out_dir)
if move:
# Remove that file.
adb_command = adb_prefix + ('shell', 'rm', file_path)
elif os.path.abspath(file_path) != os.path.abspath(out_file_path):
if move:
shutil.move(file_path, out_file_path)
shutil.copy(file_path, out_file_path)
return out_file_path
def _RunPesq(executable_path, reference_file, degraded_file,
directory = os.path.dirname(reference_file)
assert os.path.dirname(degraded_file) == directory
# Analyze audio.
command = [executable_path, '+%d' % sample_rate_hz,
# Need to provide paths in the current directory due to a bug in PESQ:
# On Mac, for some 'path/to/file.wav', if 'file.wav' is longer than
# 'path/to', PESQ crashes.
out = subprocess.check_output(_LogCommand(command),
cwd=directory, stderr=subprocess.STDOUT)
# Find the scores in stdout of PESQ.
match =
r'Prediction \(Raw MOS, MOS-LQO\):\s+=\s+([\d.]+)\s+([\d.]+)', out)
if match:
raw_mos, _ = match.groups()
return {'pesq_mos': (raw_mos, 'score')}
logging.error('PESQ: %s', out.splitlines()[-1])
return {}
def _RunPolqa(executable_path, reference_file, degraded_file):
# Analyze audio.
command = [executable_path, '-q', '-LC', 'NB',
'-Ref', reference_file, '-Test', degraded_file]
process = subprocess.Popen(_LogCommand(command),
stdout=subprocess.PIPE, stderr=subprocess.PIPE)
except OSError as e:
if e.errno == os.errno.ENOENT:
logging.warning('POLQA executable missing, skipping test.')
return {}
out, err = process.communicate()
# Find the scores in stdout of POLQA.
match ='\bMOS-LQO:\s+([\d.]+)', out)
if process.returncode != 0 or not match:
if process.returncode == 2:
logging.warning('%s (2)', err.strip())
logging.warning('POLQA license error, skipping test.')
logging.error('%s (%d)', err.strip(), process.returncode)
return {}
mos_lqo, = match.groups()
return {'polqa_mos_lqo': (mos_lqo, 'score')}
Analyzer = collections.namedtuple('Analyzer', ['func', 'executable',
def main():
# pylint: disable=W0101
args = _ParseArgs()
pesq_path, polqa_path = _DownloadTools()
out_dir = os.path.join(args.build_dir, '..')
test_command = [os.path.join(args.build_dir, 'bin',
'run_low_bandwidth_audio_test'), '-v']
test_command = [os.path.join(args.build_dir, 'low_bandwidth_audio_test')]
analyzers = [Analyzer(_RunPesq, pesq_path, 16000)]
# Check if POLQA can run at all, or skip the 48 kHz tests entirely.
example_path = os.path.join(SRC_DIR, 'resources',
'voice_engine', 'audio_tiny48.wav')
if _RunPolqa(polqa_path, example_path, example_path):
analyzers.append(Analyzer(_RunPolqa, polqa_path, 48000))
for analyzer in analyzers:
# Start the test executable that produces audio files.
test_process = subprocess.Popen(
_LogCommand(test_command + ['--sample_rate_hz=%d' %
stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
lines = iter(test_process.stdout.readline, '')
for result in ExtractTestRuns(lines, echo=True):
(android_device, test_name, reference_file, degraded_file) = result
adb_prefix = (args.adb_path,)
if android_device:
adb_prefix += ('-s', android_device)
reference_file = _GetFile(reference_file, out_dir,, adb_prefix=adb_prefix)
degraded_file = _GetFile(degraded_file, out_dir, move=True,, adb_prefix=adb_prefix)
analyzer_results = analyzer.func(analyzer.executable,
reference_file, degraded_file)
for metric, (value, units) in analyzer_results.items():
# Output a result for the perf dashboard.
print 'RESULT %s: %s= %s %s' % (metric, test_name, value, units)
if args.remove:
return test_process.wait()
if __name__ == '__main__':