# Copyright (c) 2013 The Chromium Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
import logging
import re
import subprocess
from autotest_lib.client.common_lib import error
from autotest_lib.client.common_lib import utils
from autotest_lib.client.cros.audio import cmd_utils
ACONNECT_PATH = '/usr/bin/aconnect'
ARECORD_PATH = '/usr/bin/arecord'
APLAY_PATH = '/usr/bin/aplay'
AMIXER_PATH = '/usr/bin/amixer'
CARD_NUM_RE = re.compile(r'(\d+) \[.*\]:')
CLIENT_NUM_RE = re.compile(r'client (\d+):')
DEV_NUM_RE = re.compile(r'.* \[.*\], device (\d+):')
CONTROL_NAME_RE = re.compile(r"name='(.*)'")
SCONTROL_NAME_RE = re.compile(r"Simple mixer control '(.*)'")
AUDIO_DEVICE_STATUS_CMD = 'cat /proc/asound/card%s/pcm%sp/sub0/status'
OUTPUT_DEVICE_CMD = 'cras_test_client --dump_audio_thread | grep "Output dev:"'
CARD_PREF_RECORD_DEV_IDX = {
'bxtda7219max': 3,
}
def _get_format_args(channels, bits, rate):
args = ['-c', str(channels)]
args += ['-f', 'S%d_LE' % bits]
args += ['-r', str(rate)]
return args
def get_num_soundcards():
'''Returns the number of soundcards.
Number of soundcards is parsed from /proc/asound/cards.
Sample content:
0 [PCH ]: HDA-Intel - HDA Intel PCH
HDA Intel PCH at 0xef340000 irq 103
1 [NVidia ]: HDA-Intel - HDA NVidia
HDA NVidia at 0xef080000 irq 36
'''
card_id = None
with open('/proc/asound/cards', 'r') as f:
for line in f:
match = CARD_NUM_RE.search(line)
if match:
card_id = int(match.group(1))
if card_id is None:
return 0
else:
return card_id + 1
def _get_soundcard_controls(card_id):
'''Gets the controls for a soundcard.
@param card_id: Soundcard ID.
@raise RuntimeError: If failed to get soundcard controls.
Controls for a soundcard is retrieved by 'amixer controls' command.
amixer output format:
numid=32,iface=CARD,name='Front Headphone Jack'
numid=28,iface=CARD,name='Front Mic Jack'
numid=1,iface=CARD,name='HDMI/DP,pcm=3 Jack'
numid=8,iface=CARD,name='HDMI/DP,pcm=7 Jack'
Controls with iface=CARD are parsed from the output and returned in a set.
'''
cmd = [AMIXER_PATH, '-c', str(card_id), 'controls']
p = cmd_utils.popen(cmd, stdout=subprocess.PIPE)
output, _ = p.communicate()
if p.wait() != 0:
raise RuntimeError('amixer command failed')
controls = set()
for line in output.splitlines():
if not 'iface=CARD' in line:
continue
match = CONTROL_NAME_RE.search(line)
if match:
controls.add(match.group(1))
return controls
def _get_soundcard_scontrols(card_id):
'''Gets the simple mixer controls for a soundcard.
@param card_id: Soundcard ID.
@raise RuntimeError: If failed to get soundcard simple mixer controls.
Simple mixer controls for a soundcard is retrieved by 'amixer scontrols'
command. amixer output format:
Simple mixer control 'Master',0
Simple mixer control 'Headphone',0
Simple mixer control 'Speaker',0
Simple mixer control 'PCM',0
Simple controls are parsed from the output and returned in a set.
'''
cmd = [AMIXER_PATH, '-c', str(card_id), 'scontrols']
p = cmd_utils.popen(cmd, stdout=subprocess.PIPE)
output, _ = p.communicate()
if p.wait() != 0:
raise RuntimeError('amixer command failed')
scontrols = set()
for line in output.splitlines():
match = SCONTROL_NAME_RE.findall(line)
if match:
scontrols.add(match[0])
return scontrols
def get_first_soundcard_with_control(cname, scname):
'''Returns the soundcard ID with matching control name.
@param cname: Control name to look for.
@param scname: Simple control name to look for.
'''
cpat = re.compile(r'\b%s\b' % cname, re.IGNORECASE)
scpat = re.compile(r'\b%s\b' % scname, re.IGNORECASE)
for card_id in xrange(get_num_soundcards()):
for pat, func in [(cpat, _get_soundcard_controls),
(scpat, _get_soundcard_scontrols)]:
if any(pat.search(c) for c in func(card_id)):
return card_id
return None
def get_soundcard_names():
'''Returns a dictionary of card names, keyed by card number.'''
cmd = "alsa_helpers -l"
try:
output = utils.system_output(command=cmd, retain_output=True)
except error.CmdError:
raise RuntimeError('alsa_helpers -l failed to return card names')
return dict((index, name) for index, name in (
line.split(',') for line in output.splitlines()))
def get_default_playback_device():
'''Gets the first playback device.
Returns the first playback device or None if it fails to find one.
'''
card_id = get_first_soundcard_with_control(cname='Headphone Jack',
scname='Headphone')
if card_id is None:
return None
return 'plughw:%d' % card_id
def get_record_card_name(card_idx):
'''Gets the recording sound card name for given card idx.
Returns the card name inside the square brackets of arecord output lines.
'''
card_name_re = re.compile(r'card %d: .*?\[(.*?)\]' % card_idx)
cmd = [ARECORD_PATH, '-l']
p = cmd_utils.popen(cmd, stdout=subprocess.PIPE)
output, _ = p.communicate()
if p.wait() != 0:
raise RuntimeError('arecord -l command failed')
for line in output.splitlines():
match = card_name_re.search(line)
if match:
return match.group(1)
return None
def get_record_device_supported_channels(device):
'''Gets the supported channels for the record device.
@param device: The device to record the audio. E.g. hw:0,1
Returns the supported values in integer in a list for the device.
If the value doesn't exist or the command fails, return None.
'''
cmd = "alsa_helpers --device %s --get_capture_channels" % device
try:
output = utils.system_output(command=cmd, retain_output=True)
except error.CmdError:
logging.error("Fail to get supported channels for %s", device)
return None
supported_channels = output.splitlines()
if not supported_channels:
logging.error("Supported channels are empty for %s", device)
return None
return [int(i) for i in supported_channels]
def get_default_record_device():
'''Gets the first record device.
Returns the first record device or None if it fails to find one.
'''
card_id = get_first_soundcard_with_control(cname='Mic Jack', scname='Mic')
if card_id is None:
return None
card_name = get_record_card_name(card_id)
if CARD_PREF_RECORD_DEV_IDX.has_key(card_name):
return 'plughw:%d,%d' % (card_id, CARD_PREF_RECORD_DEV_IDX[card_name])
# Get first device id of this card.
cmd = [ARECORD_PATH, '-l']
p = cmd_utils.popen(cmd, stdout=subprocess.PIPE)
output, _ = p.communicate()
if p.wait() != 0:
raise RuntimeError('arecord -l command failed')
dev_id = 0
for line in output.splitlines():
if 'card %d:' % card_id in line:
match = DEV_NUM_RE.search(line)
if match:
dev_id = int(match.group(1))
break
return 'plughw:%d,%d' % (card_id, dev_id)
def _get_sysdefault(cmd):
p = cmd_utils.popen(cmd, stdout=subprocess.PIPE)
output, _ = p.communicate()
if p.wait() != 0:
raise RuntimeError('%s failed' % cmd)
for line in output.splitlines():
if 'sysdefault' in line:
return line
return None
def get_sysdefault_playback_device():
'''Gets the sysdefault device from aplay -L output.'''
return _get_sysdefault([APLAY_PATH, '-L'])
def get_sysdefault_record_device():
'''Gets the sysdefault device from arecord -L output.'''
return _get_sysdefault([ARECORD_PATH, '-L'])
def playback(*args, **kwargs):
'''A helper funciton to execute playback_cmd.
@param kwargs: kwargs passed to playback_cmd.
'''
cmd_utils.execute(playback_cmd(*args, **kwargs))
def playback_cmd(
input, duration=None, channels=2, bits=16, rate=48000, device=None):
'''Plays the given input audio by the ALSA utility: 'aplay'.
@param input: The input audio to be played.
@param duration: The length of the playback (in seconds).
@param channels: The number of channels of the input audio.
@param bits: The number of bits of each audio sample.
@param rate: The sampling rate.
@param device: The device to play the audio on. E.g. hw:0,1
@raise RuntimeError: If no playback device is available.
'''
args = [APLAY_PATH]
if duration is not None:
args += ['-d', str(duration)]
args += _get_format_args(channels, bits, rate)
if device is None:
device = get_default_playback_device()
if device is None:
raise RuntimeError('no playback device')
else:
device = "plug%s" % device
args += ['-D', device]
args += [input]
return args
def record(*args, **kwargs):
'''A helper function to execute record_cmd.
@param kwargs: kwargs passed to record_cmd.
'''
cmd_utils.execute(record_cmd(*args, **kwargs))
def record_cmd(
output, duration=None, channels=1, bits=16, rate=48000, device=None):
'''Records the audio to the specified output by ALSA utility: 'arecord'.
@param output: The filename where the recorded audio will be stored to.
@param duration: The length of the recording (in seconds).
@param channels: The number of channels of the recorded audio.
@param bits: The number of bits of each audio sample.
@param rate: The sampling rate.
@param device: The device used to recorded the audio from. E.g. hw:0,1
@raise RuntimeError: If no record device is available.
'''
args = [ARECORD_PATH]
if duration is not None:
args += ['-d', str(duration)]
args += _get_format_args(channels, bits, rate)
if device is None:
device = get_default_record_device()
if device is None:
raise RuntimeError('no record device')
else:
device = "plug%s" % device
args += ['-D', device]
args += [output]
return args
def mixer_cmd(card_id, cmd):
'''Executes amixer command.
@param card_id: Soundcard ID.
@param cmd: Amixer command to execute.
@raise RuntimeError: If failed to execute command.
Amixer command like ['set', 'PCM', '2dB+'] with card_id 1 will be executed
as:
amixer -c 1 set PCM 2dB+
Command output will be returned if any.
'''
cmd = [AMIXER_PATH, '-c', str(card_id)] + cmd
p = cmd_utils.popen(cmd, stdout=subprocess.PIPE)
output, _ = p.communicate()
if p.wait() != 0:
raise RuntimeError('amixer command failed')
return output
def get_num_seq_clients():
'''Returns the number of seq clients.
The number of clients is parsed from aconnect -io.
This is run as the chronos user to catch permissions problems.
Sample content:
client 0: 'System' [type=kernel]
0 'Timer '
1 'Announce '
client 14: 'Midi Through' [type=kernel]
0 'Midi Through Port-0'
@raise RuntimeError: If no seq device is available.
'''
cmd = [ACONNECT_PATH, '-io']
output = cmd_utils.execute(cmd, stdout=subprocess.PIPE, run_as='chronos')
num_clients = 0
for line in output.splitlines():
match = CLIENT_NUM_RE.match(line)
if match:
num_clients += 1
return num_clients
def convert_device_name(cras_device_name):
'''Converts cras device name to alsa device name.
@returns: alsa device name that can be passed to aplay -D or arecord -D.
For example, if cras_device_name is "kbl_r5514_5663_max: :0,1",
this function will return "hw:0,1".
'''
tokens = cras_device_name.split(":")
return "hw:%s" % tokens[2]
def check_audio_stream_at_selected_device(device_name, device_type):
"""Checks the audio output at expected node
@param device_name: Audio output device name, Ex: kbl_r5514_5663_max: :0,1
@param device_type: Audio output device type, Ex: INTERNAL_SPEAKER
"""
if device_type == 'BLUETOOTH':
output_device_output = utils.system_output(OUTPUT_DEVICE_CMD).strip()
bt_device = output_device_output.split('Output dev:')[1].strip()
if bt_device != device_name:
raise error.TestFail("Audio is not routing through expected node")
logging.info('Audio is routing through %s', bt_device)
else:
card_device_search = re.search(r':(\d),(\d)', device_name)
if card_device_search:
card_num = card_device_search.group(1)
device_num = card_device_search.group(2)
logging.debug("Sound card number is %s", card_num)
logging.debug("Device number is %s", device_num)
if card_num is None or device_num is None:
raise error.TestError("Audio device name is not in expected format")
device_status_output = utils.system_output(AUDIO_DEVICE_STATUS_CMD %
(card_num, device_num))
logging.debug("Selected output device status is %s",
device_status_output)
if 'RUNNING' in device_status_output:
logging.info("Audio is routing through expected node!")
elif 'closed' in device_status_output:
raise error.TestFail("Audio is not routing through expected audio "
"node!")
else:
raise error.TestError("Audio routing error! Device may be "
"preparing")