blob: 2f53ec52974b70e2da39e8c731afd83e5877ce63 [file] [log] [blame]
# Copyright 2014 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.
"""A factory test for the audio function.
This test perform tests on audio plaback and recording devices. It supports 2
loopback modes:
1. Loop from headphone out to headphone in.
2. Loop from speaker to digital microphone.
And 3 test scenarios:
1. Audiofun test, which plays different tones and checks recorded frequency.
This test can be conducted simultaneously on different devices. This test can
not be conducted with dongle inserted.
2. Sinewav test, which plays simple sine wav and checks if the recorded
frequency is in the range specified. Optionally checks the RMS and amplitude
3. Noise test, which plays nothing and record, then checks the RMS and amplitude
Since this test is sensitive to different loopback dongles, user can set a list
of output volume candidates. The test can pass if it can pass at any one of
output volume candidates.
Test Procedure
1. Operator inserts the dongle (if required).
2. The playback starts automatically, and analyze recordings afterward.
- Device API ````.
Here are some test list examples for different test cases. First, you need to
figure out the particular input/output device you want to perform test on. For
ALSA input devices, the command `arecord -l` can be used to list all available
input devices.
For instance, if the device showing as ``card 0: kblrt5514rt5663
[kblrt5514rt5663max], device 1: Audio Record (*)`` is what you want, the
input_dev should be set to ["kblrt5514rt5663max", "1"]. Similarly, the
output_dev might be ["kblrt5514rt5663max", "0"]. These settings are used in the
following examples.
Audiofuntest external mic (default) of input_dev and speakers of output_dev::
"pytest_name": "audio_loop",
"args": {
"input_dev": ["kblrt5514rt5663max", "1"],
"output_dev": ["kblrt5514rt5663max", "0"],
"output_volume": 10,
"require_dongle": false,
"check_dongle": true,
"initial_actions": [
["1", "init_speakerdmic"]
"tests_to_conduct": [
"duration": 4,
"threshold": 80,
"type": "audiofun"
Audiofuntest on 'mlb' mics of input_dev and speaker channel 0 of output_dev::
"pytest_name": "audio_loop",
"args": {
"input_dev": ["kblrt5514rt5663max", "1"],
"output_dev": ["kblrt5514rt5663max", "0"],
"output_volume": 10,
"require_dongle": false,
"check_dongle": true,
"mic_source": "MLBDmic",
"initial_actions": [
["1", "init_speakerdmic"]
"tests_to_conduct": [
"threshold": 80,
"capture_rate": 16000,
"type": "audiofun",
"output_channels": [0]
"pytest_name": "audio_loop",
"args": {
"input_dev": ["kblrt5514rt5663max", "1"],
"output_dev": ["kblrt5514rt5663max", "0"],
"require_dongle": false,
"check_dongle": true,
"initial_actions": [
["1", "init_speakerdmic"]
"tests_to_conduct": [
"duration": 2,
"amplitude_threshold": [-0.9, 0.9],
"type": "noise",
"rms_threshold": [null, 0.5]
"pytest_name": "audio_loop",
"args": {
"input_dev": ["kblrt5514rt5663max", "1"],
"output_dev": ["kblrt5514rt5663max", "0"],
"output_volume": 15,
"require_dongle": true,
"check_dongle": true,
"initial_actions": [
["1", "init_audiojack"]
"tests_to_conduct": [
"freq_threshold": 50,
"type": "sinewav",
"rms_threshold": [0.08, null]
from __future__ import print_function
import os
import re
import tempfile
import time
import factory_common # pylint: disable=unused-import
from import base
from cros.factory.device import device_utils
from cros.factory.test import session
from cros.factory.test import test_case
from cros.factory.test.utils import audio_utils
from cros.factory.testlog import testlog
from cros.factory.utils.arg_utils import Arg
from cros.factory.utils import process_utils
# Default setting
# the duration(secs) for sine tone to playback.
# it must be long enough for record process.
# Regular expressions to match audiofuntest message.
_AUDIOFUNTEST_MIC_CHANNEL_RE = re.compile(r'.*Microphone channels:\s*(.*)$')
_AUDIOFUNTEST_RUN_START_RE = re.compile('^carrier')
# Default minimum success rate of audiofun test to pass.
# Default iterations to do the audiofun test.
# Default channels of the input_dev to be tested.
# Default channels of the output_dev to be tested.
# Default capture sample rate used for audiofuntest.
# Default audio gain used for audiofuntest.
# Default duration to do the sinewav test, in seconds.
# Default frequency tolerance, in Hz.
# Default duration to do the noise test, in seconds.
# Default RMS thresholds when checking recorded file.
# Default Amplitude thresholds when checking recorded file.
# Default duration in seconds to trim in the beginning of recorded file.
# Default sample format for player used by aplay, S16 = Signed 16 Bit.
# Default sample format for recorder used by arecord, S16 = Signed 16 Bit.
# Default minimum frequency.
# Default maximum frequency.
class AudioLoopTest(test_case.TestCase):
"""Audio Loop test to test two kind of situations.
1. Speaker to digital microphone.
2. Headphone out to headphone in.
ARGS = [
Arg('audio_conf', str, 'Audio config file path', default=None),
Arg('initial_actions', list,
'List of [card, actions]. If actions is None, the Initialize method '
'will be invoked.',
Arg('input_dev', list,
'Input ALSA device. [card_name, sub_device].'
'For example: ["audio_card", "0"].', ['0', '0']),
Arg('num_input_channels', int,
'Number of input channels.', default=2),
Arg('output_dev', list,
'Output ALSA device. [card_name, sub_device].'
'For example: ["audio_card", "0"].', ['0', '0']),
Arg('output_volume', (int, list),
'An int of output volume or a list of output volume candidates',
Arg('autostart', bool, 'Auto start option', default=False),
Arg('require_dongle', bool, 'Require dongle option', default=False),
Arg('check_dongle', bool,
'Check dongle status whether match require_dongle', default=False),
Arg('check_cras', bool, 'Do we need to check if CRAS is running',
Arg('cras_enabled', bool, 'Whether cras should be running or not',
Arg('mic_source', base.InputDevices, 'Microphone source',
Arg('test_title', str,
'Title on the test screen.'
'It can be used to tell operators the test info'
'For example: "LRGM Mic", "LRMG Mic"', default=''),
Arg('mic_jack_type', str, 'Microphone jack Type: nocheck, lrgm, lrmg',
Arg('audiofuntest_run_delay', (int, float),
'Delay between consecutive calls to audiofuntest',
Arg('tests_to_conduct', list,
'A list of dicts. A dict should contain at least one key named\n'
'**type** indicating the test type, which can be **audiofun**,\n'
'**sinewav**, or **noise**.\n'
'If type is **audiofun**, the dict can optionally contain:\n'
' - **iteration**: Iterations to run the test.\n'
' - **threshold**: The minimum success rate to pass the test.\n'
' - **input_channels**: A list of input channels to be tested.\n'
' - **output_channels**: A list of output channels to be tested.\n'
' - **capture_rate**: The capturing sample rate use for testing.\n'
' - **volume_gain**: The volume gain set to audiofuntest for \n'
' controlling the volume of generated audio frames. The \n'
' range is from 0 to 100.'
' - **recorder_sample_format**: The sample format for the input \n'
' device. Use arecord to see all possible formats.'
' - **player_sample_format**: The sample format for the output \n'
' device. Use aplay to see all possible formats.'
' - **min_frequency**: The minimum frequency set to audiofuntest.\n'
' - **max_frequency**: The maximum frequency set to audiofuntest.\n'
'If type is **sinewav**, the dict can optionally contain:\n'
' - **duration**: The test duration, in seconds.\n'
' - **freq_threshold**: Acceptable frequency margin.\n'
' - **rms_threshold**: **[min, max]** that will make\n'
' sure the following inequality is true: *min <= recorded\n'
' audio RMS (root mean square) value <= max*, otherwise,\n'
' fail the test. Both of **min** and **max** can be set to\n'
' None, which means no limit.\n'
' - **amplitude_threshold**: **[min, max]** and it will\n'
' make sure the inequality is true: *min <= minimum measured\n'
' amplitude <= maximum measured amplitude <= max*,\n'
' otherwise, fail the test. Both of **min** and **max** can\n'
' be set to None, which means no limit.\n'
'If type is **noise**, the dict can optionally contain:\n'
' - **duration**: The test duration, in seconds.\n'
' - **rms_threshold**: **[min, max]** that will make\n'
' sure the following inequality is true: *min <= recorded\n'
' audio RMS (root mean square) value <= max*, otherwise,\n'
' fail the test. Both of **min** and **max** can be set to\n'
' None, which means no limit.\n'
' - **amplitude_threshold**: **[min, max]** and it will\n'
' make sure the inequality is true: *min <= minimum measured\n'
' amplitude <= maximum measured amplitude <= max*,\n'
' otherwise, fail the test. Both of **min** and **max** can\n'
' be set to None, which means no limit.'),
Arg('keep_raw_logs', bool,
'Whether to attach the audio by Testlog when the test fail.',
def setUp(self):
self._dut = device_utils.CreateDUTInterface()
if self.args.audio_conf:
# Transfer input and output device format
self._in_card =[0])
self._in_device = self.args.input_dev[1]
self._out_card =[0])
self._out_device = self.args.output_dev[1]
# Backward compatible for non-porting case, which use ALSA device name.
# only works on chromebook device
# TODO(mojahsu) Remove them later.
self._alsa_input_device = 'hw:%s,%s' % (self._in_card, self._in_device)
self._alsa_output_device = 'hw:%s,%s' % (self._out_card, self._out_device)
self._output_volumes = self.args.output_volume
if not isinstance(self._output_volumes, list):
self._output_volumes = [self._output_volumes]
self._output_volume_index = 0
self._freq = _DEFAULT_FREQ_HZ
# The test results under each output volume candidate.
# If any one of tests to conduct fails, test fails under that output
# volume candidate. If test fails under all output volume candidates,
# the whole test fails.
self._test_results = [True] * len(self._output_volumes)
self._test_message = []
self._mic_jack_type = {
'nocheck': None,
'lrgm': base.MicJackType.lrgm,
'lrmg': base.MicJackType.lrmg
if self.args.initial_actions is None:
for card, action in self.args.initial_actions:
if card.isdigit() is False:
card =
if action is None:
else:, card)
self._current_test_args = None
if self.args.check_cras:
# Check cras status
if self.args.cras_enabled:
cras_status = 'start/running'
cras_status = 'stop/waiting'
self._dut.CallOutput(['status', 'cras']),
'cras status is wrong (expected status: %s). '
'Please make sure that you have appropriate setting for '
'\'"disable_services": ["cras"]\' in the test item.' % cras_status)
self._dut_temp_dir = self._dut.temp.mktemp(True, '', 'audio_loop')
# If the test fails, attach the audio file; otherwise, remove it.
self._audio_file_path = []
def tearDown(self):
self._dut.CheckCall(['rm', '-rf', self._dut_temp_dir])
def runTest(self):
# If autostart, JS triggers start_run_test event.
# Otherwise, it binds start_run_test with 's' key pressed.
self.args.require_dongle, self.args.test_title)
if self.args.autostart:
self.ui.RunJS('window.template.innerHTML = "";')
# Run each tests to conduct under each output volume candidate.
for self._output_volume_index, output_volume in enumerate(
if output_volume is not None:
if self.args.require_dongle:, self._out_card)
else:, self._out_card)
for test in self.args.tests_to_conduct:
self._current_test_args = test
if test['type'] == 'audiofun':
elif test['type'] == 'sinewav':
elif test['type'] == 'noise':
raise ValueError('Test type "%s" not supported.' % test['type'])
if self.MayPassTest():
for file_path in self._audio_file_path:
if self.args.keep_raw_logs:
for file_path in self._audio_file_path:
description='recorded audio of the test',
for file_path in self._audio_file_path:
def AppendErrorMessage(self, error_message):
"""Sets the test result to fail and append a new error message."""
self._test_results[self._output_volume_index] = False
'Under output volume %r' % self._output_volumes[
def _MatchPatternLines(self, in_stream, re_pattern, num_lines=None):
"""Try to match the re pattern in the given number of lines.
Try to read lines one-by-one from input stream and perform re matching.
Stop when matching successes or reaching the number of lines limit.
in_stream: input stream to read from.
re_pattern: re pattern used for matching.
num_lines: maximum number of lines to stop for matching.
None for read until end of input stream.
num_read = 0
while True:
line = in_stream.readline()
if not line:
return None
num_read += 1
m = re_pattern.match(line)
if m is not None:
return m
if num_lines is not None and num_read >= num_lines:
return None
def _ParseSingleRunOutput(self, audiofun_output, input_channels):
"""Parse a single run output from audiofuntest
Sample single run output:
O: channel = 0, success = 1, fail = 0, rate = 100.0
X: channel = 1, success = 0, fail = 1, rate = 0.0
audiofun_output: output stream of audiofuntest to parse from
input_channels: a list of mic channels used for testing
all_channel_rate = {}
for expected_channel in input_channels:
m = self._MatchPatternLines(
audiofun_output, _AUDIOFUNTEST_SUCCESS_RATE_RE, 1)
if m is None or int( != expected_channel:
'Failed to get expected %d channel output from audiofuntest'
% expected_channel)
return None
all_channel_rate[expected_channel] = float(
return all_channel_rate
def AudioFunTestWithOutputChannel(self, capture_rate, input_channels,
"""Runs audiofuntest program to get the frequency from microphone
immediately according to speaker and microphone setting.
Sample audiofuntest message:
Config values.
Player parameter: aplay -r 48000 -f s16 -t raw -c 2 -B 0 -
Recorder parameter: arecord -r 48000 -f s16 -t raw -c 2 -B 0 -
Player FIFO name:
Recorder FIFO name:
Number of test rounds: 20
Pass threshold: 3
Allowed delay: 1200(ms)
Sample rate: 48000
FFT size: 2048
Microphone channels: 2
Speaker channels: 2
Microphone active channels: 0, 1,
Speaker active channels: 0, 1,
Tone length (in second): 3.00
Volume range: 1.00 ~ 1.00
carrier = 119
O: channel = 0, success = 1, fail = 0, rate = 100.0
X: channel = 1, success = 0, fail = 1, rate = 0.0
carrier = 89
O: channel = 0, success = 2, fail = 0, rate = 100.0
X: channel = 1, success = 1, fail = 1, rate = 50.0
output_channel: output device channel used for testing
"""'Test output channel %d', output_channel)
iteration = self._current_test_args.get(
volume_gain = self._current_test_args.get(
self.assertTrue(0 <= volume_gain <= 100)
player_sample_format = self._current_test_args.get(
'player_sample_format', _DEFAULT_PLAYER_SAMPLE_FORMAT)
recorder_sample_format = self._current_test_args.get(
'recorder_sample_format', _DEFAULT_RECORDER_SAMPLE_FORMAT)
min_frequency = self._current_test_args.get(
'min_frequency', _DEFAULT_MIN_FREQUENCY)
self.assertGreaterEqual(min_frequency, 0)
max_frequency = self._current_test_args.get(
'max_frequency', _DEFAULT_MAX_FREQUENCY)
self.assertLessEqual(min_frequency, max_frequency)
player_cmd = 'aplay -D %s -r %d -f %s -t raw -c 2 -B 0 -' % (
self._alsa_output_device, capture_rate, player_sample_format)
recorder_cmd = 'arecord -D %s -r %d -f %s -t raw -c %d -B 0 -' % (
self._alsa_input_device, capture_rate, recorder_sample_format,
process = self._dut.Popen(
'-P', player_cmd,
'-R', recorder_cmd,
'-r', '%d' % capture_rate,
'-T', '%d' % iteration,
'-a', '%d' % output_channel,
'-m', ','.join(map(str, input_channels)),
'-c', '%d' % self.args.num_input_channels,
'-g', '%d' % volume_gain,
'-i', '%d' % min_frequency,
'-x', '%d' % max_frequency],
stdout=process_utils.PIPE, stderr=process_utils.PIPE)
last_success_rate = None
while self._MatchPatternLines(process.stdout,
last_success_rate = self._ParseSingleRunOutput(process.stdout,
if last_success_rate is None:
rate_msg = ', '.join(
'Mic %d: %.1f%%' %
(channel, rate) for channel, rate in last_success_rate.viewitems())
self.ui.CallJSFunction('testInProgress', rate_msg)
if last_success_rate is None:
self.AppendErrorMessage('Failed to parse audiofuntest output')
threshold = self._current_test_args.get(
if any(rate < threshold for rate in last_success_rate.viewvalues()):
'For output device channel %s, the success rate is "'
'%s", too low!' % (output_channel, rate_msg))
self.ui.CallJSFunction('testFailResult', rate_msg)
def AudioFunTest(self):
"""Setup speaker and microphone test pairs and run audiofuntest program."""'Run audiofuntest from %r to %r',
self._alsa_output_device, self._alsa_input_device)
input_channels = self._current_test_args.get(
output_channels = self._current_test_args.get(
capture_rate = self._current_test_args.get(
for output_channel in output_channels:
self.AudioFunTestWithOutputChannel(capture_rate, input_channels,
if self.args.audiofuntest_run_delay is not None:
def TestLoopbackChannel(self, num_channels):
"""Tests loopback on all channels.
num_channels: Number of channels to test
# TODO(phoenixshen): Support quad channels here.
# This test assumes number of input channels == number of output channels,
# and ID of valid channels should be the same,
# Need to redesign the args to provide more flexbility.
duration = self._current_test_args.get('duration',
for channel in xrange(num_channels):
# file path in host
record_file_path = '/tmp/record-%d-%d-%s.raw' % (
channel, time.time())
sine_wav_path = '/tmp/%d_%d.wav' % (self._freq, channel)
dut_sine_wav_path = self._dut.path.join(self._dut_temp_dir,
'sine_%d.wav' % channel)'DUT sine wav path %s', dut_sine_wav_path)
# Generate sine .wav file locally and push it to the DUT.
# It's hard to estimate the overhead in audio record thing of different
# platform, To make sure we can record the whole sine tone in the record
# duration, we will playback a long period sine tone, and stop the
# playback process after we finish recording.
cmd = audio_utils.GetGenerateSineWavArgs(sine_wav_path, channel,
process_utils.Spawn(cmd.split(' '), log=True, check_call=True), dut_sine_wav_path), self._out_card,
self._out_device, blocking=False)
self.RecordFile(duration, record_file_path)
sox_output = audio_utils.SoxStatOutput(record_file_path, channel)
def SinewavTest(self):
self.ui.CallJSFunction('testInProgress', None)
# Playback sine tone and check the recorded audio frequency.
def NoiseTest(self):
self.ui.CallJSFunction('testInProgress', None)
# Record the noise file.
duration = self._current_test_args.get(
noise_file_path = '/tmp/noise-%s.wav' % time.time()
# Do not trim because we want to check all possible noises and artifacts.
self.RecordFile(duration, noise_file_path, None)
# Since we have actually only 1 channel, we can just give channel=0 here.
sox_output = audio_utils.SoxStatOutput(noise_file_path, 0)
def RecordFile(self, duration, file_path, trim=_DEFAULT_TRIM_SECONDS):
"""Records file for *duration* seconds.
The caller is responsible for removing the file at last.
duration: Recording duration, in seconds.
file_path: The file path to recorded file in host.
trim: If not None, the number of seconds in the beginning to trim.
"""'RecordFile : %s.', file_path)
record_path = (tempfile.NamedTemporaryFile(delete=False).name if trim
else file_path)
with self._dut.temp.TempFile() as dut_record_path:, self._in_card,
self._in_device, duration, 2, 48000), record_path)
if trim:
audio_utils.TrimAudioFile(in_path=record_path, out_path=file_path,
start=trim, end=None, num_channel=2)
def CheckRecordedAudio(self, sox_output):
rms_value = audio_utils.GetAudioRms(sox_output)'Got audio RMS value: %f.', rms_value)
rms_threshold = self._current_test_args.get(
'rms_threshold', _DEFAULT_SOX_RMS_THRESHOLD)
if rms_threshold[0] is not None and rms_threshold[0] > rms_value:
self.AppendErrorMessage('Audio RMS value %f too low. Minimum pass is %f.'
% (rms_value, rms_threshold[0]))
if rms_threshold[1] is not None and rms_threshold[1] < rms_value:
self.AppendErrorMessage('Audio RMS value %f too high. Maximum pass is %f.'
% (rms_value, rms_threshold[1]))
amplitude_threshold = self._current_test_args.get(
'amplitude_threshold', _DEFAULT_SOX_AMPLITUDE_THRESHOLD)
min_value = audio_utils.GetAudioMinimumAmplitude(sox_output)'Got audio min amplitude: %f.', min_value)
if (amplitude_threshold[0] is not None and
amplitude_threshold[0] > min_value):
'Audio minimum amplitude %f too low. Minimum pass is %f.' % (
min_value, amplitude_threshold[0]))
max_value = audio_utils.GetAudioMaximumAmplitude(sox_output)'Got audio max amplitude: %f.', max_value)
if (amplitude_threshold[1] is not None and
amplitude_threshold[1] < max_value):
'Audio maximum amplitude %f too high. Maximum pass is %f.' % (
max_value, amplitude_threshold[1]))
if self._current_test_args['type'] == 'sinewav':
freq = audio_utils.GetRoughFreq(sox_output)
freq_threshold = self._current_test_args.get(
'freq_threshold', _DEFAULT_SINEWAV_FREQ_THRESHOLD)'Expected frequency %r +- %d',
self._freq, freq_threshold)
if freq is None or (abs(freq - self._freq) > freq_threshold):
self.AppendErrorMessage('Test Fail at frequency %r' % freq)
else:'Got frequency %d', freq)
def MayPassTest(self):
"""Checks if test can pass with result of one output volume.
Returns: True if test passes, False otherwise.
"""'Test results for output volume %r: %r',
if self._test_results[self._output_volume_index]:
return True
return False
def FailTest(self):
"""Fails test."""'Test results for each output volumes: %r',
zip(self._output_volumes, self._test_results))
self.FailTask('; '.join(self._test_message))
def CheckDongleStatus(self):
# When audio jack detection feature is ready on a platform, we can
# enable check_dongle option to check jack status matches we expected.
if self.args.check_dongle:
mic_status =
headphone_status =
plug_status = mic_status or headphone_status
# We've encountered false positive running audiofuntest tool against
# audio fun-plug on a few platforms; so it is suggested not to run
# audiofuntest with HP/MIC jack
if plug_status is True:
if any((t['type'] == 'audiofun') for t in self.args.tests_to_conduct):'Audiofuntest does not require dongle.')
raise ValueError('Audiofuntest does not require dongle.')
if self.args.require_dongle is False:'Dongle Status is wrong, don\'t need dongle.')
raise ValueError('Dongle Status is wrong.')
# for require dongle case, we need to check both microphone and headphone
# are all detected.
if self.args.require_dongle:
if (mic_status and headphone_status) is False:'Dongle Status is wrong. mic %s, headphone %s',
mic_status, headphone_status)
raise ValueError('Dongle Status is wrong.')
if self._mic_jack_type:
mictype =
if mictype != self._mic_jack_type:'Mic Jack Type is wrong. need %s, but %s',
raise ValueError('Mic Jack Type is wrong.')
def SetupAudio(self):
# Enable/disable devices according to require_dongle.
# We don't use plug_status because plug_status may not be ready at early
# stage.
if self.args.require_dongle:
else:, self._in_card)