blob: b9902ce10d7682688b573fe10d855c00a7426bdb [file] [log] [blame]
# Lint as: python2, python3
# Copyright 2020 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.
"""Facade to access the bluetooth-related functionality."""
from __future__ import absolute_import
from __future__ import division
from __future__ import print_function
import base64
import binascii
import collections
from datetime import datetime, timedelta
import glob
# AU tests use ToT client code, but ToT -3 client version.
try:
from gi.repository import GLib, GObject
except ImportError:
import gobject as GObject
import json
import logging
import logging.handlers
import os
# TODO(b/215715213) - Wait until ebuild runs as python3 to remove this try
try:
import pydbus
except Exception as e:
import platform
logging.error('Unable to import pydbus at version=%s: %s',
platform.python_version(), e)
pydbus = {}
import re
import subprocess
import functools
import time
import threading
import traceback
import common
from autotest_lib.client.bin import utils
from autotest_lib.client.common_lib.cros.bluetooth import bluetooth_socket
from autotest_lib.client.common_lib import error
from autotest_lib.client.cros.udev_helpers import UdevadmInfo, UdevadmTrigger
from autotest_lib.client.cros.audio import (audio_test_data as
audio_test_data_module)
from autotest_lib.client.cros.audio import check_quality
from autotest_lib.client.cros.audio import cras_utils
from autotest_lib.client.cros.audio.sox_utils import (
convert_format, convert_raw_file, get_file_length,
trim_silence_from_wav_file)
from autotest_lib.client.cros.bluetooth import advertisement
from autotest_lib.client.cros.bluetooth import adv_monitor_helper
from autotest_lib.client.cros.bluetooth import output_recorder
from autotest_lib.client.cros.bluetooth import logger_helper
from autotest_lib.client.cros.bluetooth.floss.adapter_client import (
FlossAdapterClient, BluetoothCallbacks, BluetoothConnectionCallbacks,
BondState, SspVariant, Transport)
from autotest_lib.client.cros.bluetooth.floss.manager_client import FlossManagerClient
from autotest_lib.client.cros.bluetooth.floss.utils import GLIB_THREAD_NAME
from autotest_lib.client.cros.power import power_suspend_delay
from autotest_lib.client.cros.power import sys_power
import six
from six.moves import map
from six.moves import range
CheckQualityArgsClass = collections.namedtuple(
'args_type', ['filename', 'rate', 'channel', 'bit_width'])
def _dbus_byte_array_to_b64_string(dbus_byte_array):
"""Base64 encodes a dbus byte array for use with the xml rpc proxy.
Input is encoded to bytes using base64 encoding. Then the base64 bytes is
decoded as string.
"""
return base64.standard_b64encode(bytearray(dbus_byte_array)).decode()
def _b64_string_to_dbus_byte_array(b64_string):
"""Base64 decodes a dbus byte array for use with the xml rpc proxy."""
dbus_array = []
bytes = bytearray(base64.standard_b64decode(b64_string))
for byte in bytes:
dbus_array.append(byte)
return dbus_array
def dbus_safe(default_return_value, return_error=False):
"""Catch all DBus exceptions and return a default value instead.
Wrap a function with a try block that catches DBus exceptions and
returns the error with the specified return status. The exception is logged
to aid in debugging.
If |return_error| is set, the call will return a tuple with
(default_return_value, str(error)).
@param default_return_value: What value to return in case of errors.
@param return_error: Whether to return the error string as well.
@return Either the return value from the method call if successful or
the |default_return_value| or a tuple(default_return_value,
str(error))
"""
def decorator(wrapped_function):
"""Call a function and catch DBus errors.
@param wrapped_function function to call in dbus safe context.
@return function return value or default_return_value on failure.
"""
@functools.wraps(wrapped_function)
def wrapper(*args, **kwargs):
"""Pass args and kwargs to a dbus safe function.
@param args formal python arguments.
@param kwargs keyword python arguments.
@return function return value or default_return_value on failure.
"""
logging.debug('%s()', wrapped_function.__name__)
try:
return wrapped_function(*args, **kwargs)
except GLib.Error as e:
logging.debug('Exception while performing operation %s: %s',
wrapped_function.__name__, e)
if return_error:
return (default_return_value, str(e))
else:
return default_return_value
except Exception as e:
logging.debug('Exception in %s: %s', wrapped_function.__name__,
e)
logging.debug(traceback.format_exc())
raise
return wrapper
return decorator
def raw_dbus_call_sync(bus,
proxy,
iface,
method,
variant_in_args,
variant_out_type,
timeout_ms=None):
"""Makes a raw D-Bus call and returns the unpacked result.
@param bus: System bus object.
@param proxy: Proxy object.
@param iface: D-Bus interface that exposes this method.
@param method: Name of method to call.
@param variant_in_args: A Glib.Variant that corresponds to the method's
inputs.
@param variant_out_type: A Glib.VariantType that describes the output. This
is the type that will be unpacked from the result.
@param timeout_ms: Timeout in milliseconds for this method call.
@returns: Unpacked result from the method call.
"""
if timeout_ms is None:
timeout_ms = GLib.MAXINT
return bus.con.call_sync(proxy._bus_name, proxy._path, iface, method,
variant_in_args, variant_out_type, 0, timeout_ms,
None).unpack()
def unpack_if_variant(value):
"""If given value is GLib.Variant, unpack it to the actual type."""
if isinstance(value, GLib.Variant):
return value.unpack()
return value
class UpstartClient:
"""Upstart D-Bus client that allows actions on upstart targets."""
UPSTART_MANAGER_SERVICE = 'com.ubuntu.Upstart'
UPSTART_MANAGER_PATH = '/com/ubuntu/Upstart'
UPSTART_MANAGER_IFACE = 'com.ubuntu.Upstart0_6'
UPSTART_JOB_IFACE = 'com.ubuntu.Upstart0_6.Job'
UPSTART_ERROR_UNKNOWNINSTANCE = (
'com.ubuntu.Upstart0_6.Error.UnknownInstance')
UPSTART_ERROR_ALREADYSTARTED = (
'com.ubuntu.Upstart0_6.Error.AlreadyStarted')
@classmethod
def _get_job(cls, job_name):
"""Get job by name."""
bus = pydbus.SystemBus()
obj = bus.get(cls.UPSTART_MANAGER_SERVICE, cls.UPSTART_MANAGER_PATH)
job_path = obj[cls.UPSTART_MANAGER_IFACE].GetJobByName(job_name)
return bus.get(cls.UPSTART_MANAGER_SERVICE,
job_path)[cls.UPSTART_JOB_IFACE]
@staticmethod
def _convert_instance_args(source):
"""Convert instance args dict to array."""
return ['{}={}'.format(k, v) for k, v in source.items()]
@classmethod
def start(cls, job_name, instance_args = {}):
"""Starts a job.
@param job_name: Name of upstart job to start.
@param instance_args: Instance arguments. Will be converted to array of
"key=value".
@return True if job start was sent successfully.
"""
try:
job = cls._get_job(job_name)
converted_args = cls._convert_instance_args(instance_args)
job.Start(converted_args, True)
except TypeError as t:
# Can occur if cls._get_job fails
logging.error('Error starting {}: {}'.format(job_name, t))
return False
except GLib.Error as e:
# An already started error is ok. All other dbus errors should
# return False.
if cls.UPSTART_ERROR_ALREADYSTARTED not in str(e):
logging.error('Error starting {}: {}'.format(job_name, e))
return False
return True
@classmethod
def stop(cls, job_name, instance_args = {}):
"""Stops a job.
@param job_name: Name of upstart job to stop.
@param instance_args: Instance arguments. Will be converted to
array of "key=value".
@return True if job stop was sent successfully.
"""
try:
job = cls._get_job(job_name)
converted_args = cls._convert_instance_args(instance_args)
job.Stop(converted_args, True)
except TypeError as t:
# Can occur if cls._get_job fails
logging.error('Error stopping {}: {}'.format(job_name, t))
return False
except GLib.Error as e:
# If the job was already stopped, we will see an UnknownInstance
# exception. All other failure reasons should be treated as
# a failure to stop.
if cls.UPSTART_ERROR_UNKNOWNINSTANCE not in str(e):
logging.error('Error starting {}: {}'.format(job_name, e))
return False
return True
class BluetoothBaseFacadeLocal(object):
"""Base facade shared by Bluez and Floss daemons. This takes care of any
functionality that is common across the two daemons.
"""
# Both bluez and floss share the same lib dir for configuration and cache
BLUETOOTH_LIBDIR = '/var/lib/bluetooth'
SYSLOG_LEVELS = [
'EMERG', 'ALERT', 'CRIT', 'ERR', 'WARNING', 'NOTICE', 'INFO',
'DEBUG'
]
# How long to wait for hid device
HID_TIMEOUT = 15
HID_CHECK_SECS = 2
# Due to problems transferring a date object, we convert to stringtime first
# This is the standard format that we will use.
OUT_DATE_FORMAT = '%Y-%m-%d %H:%M:%S.%f'
# Upstart job name for the Floss Manager daemon
MANAGER_JOB = "btmanagerd"
# File path for btmanagerd
BTMANGERD_FILE_PATH = '/usr/bin/btmanagerd'
# How long we wait for the manager daemon to come up after we start it
DAEMON_TIMEOUT_SEC = 5
# Upstart job name for ChromeOS Audio daemon
CRAS_JOB = "cras"
# The VID:PID recorded here is the PCI vid:pid. lspci -nn command can be
# used to find it.
CHIPSET_TO_VIDPID = {
'MVL-8897': [(('0x02df', '0x912d'), 'SDIO')],
'MVL-8997': [(('0x1b4b', '0x2b42'), 'USB')],
'QCA-6174A-5-USB': [(('0x168c', '0x003e'), 'USB')],
'QCA-6174A-3-UART': [(('0x0271', '0x050a'), 'UART')],
'QCA-WCN6856': [(('0x17cb', '0x1103'), 'USB')],
'Intel-AX200': [(('0x8086', '0x2723'), 'USB')], # CcP2
'Intel-AX201': [
(('0x8086', '0x02f0'), 'USB'),
(('0x8086', '0x4df0'), 'USB'),
(('0x8086', '0xa0f0'), 'USB'),
], # HrP2
'Intel-AX211': [(('0x8086', '0x51f0'), 'USB')], # GfP2
'Intel-AC9260': [(('0x8086', '0x2526'), 'USB')], # ThP2
'Intel-AC9560': [
(('0x8086', '0x31dc'), 'USB'), # JfP2
(('0x8086', '0x9df0'), 'USB')
],
'Intel-AC7260': [
(('0x8086', '0x08b1'), 'USB'), # WP2
(('0x8086', '0x08b2'), 'USB')
],
'Intel-AC7265': [
(('0x8086', '0x095a'), 'USB'), # StP2
(('0x8086', '0x095b'), 'USB')
],
'Realtek-RTL8822C-USB': [(('0x10ec', '0xc822'), 'USB')],
'Realtek-RTL8822C-UART': [(('0x10ec', '0xc822'), 'UART')],
'Realtek-RTL8852A-USB': [(('0x10ec', '0x8852'), 'USB')],
'Realtek-RTL8852C-USB': [(('0x10ec', '0xc852'), 'USB')],
'Mediatek-MTK7921-USB': [(('0x14c3', '0x7961'), 'USB')],
'Mediatek-MTK7921-SDIO': [(('0x037a', '0x7901'), 'SDIO')]
# The following doesn't expose vid:pid
# 'WCN3991-UART'
}
def __init__(self):
# Initialize a messages object to record general logging.
self.messages = logger_helper.LogManager()
# Set up cras test client for audio tests
self._cras_test_client = cras_utils.CrasTestClient()
def configure_floss(self, enabled):
"""Start and configure the Floss manager daemon.
In order to manage whether we use bluez or floss, we need to start the
Floss manager daemon and then set floss enabled. This exists in the base
implementation because bluez tests will need to start the manager to
disable Floss.
@param enabled: Whether to enable Floss
@return Whether Floss was configured successfully.
"""
# Start manager daemon or exit early
if not UpstartClient.start(self.MANAGER_JOB):
return False
# Since we've just started the manager daemon, we also need to recreate
# the client.
self.manager_client = FlossManagerClient(self.bus)
# Wait for the manager daemon to come up
try:
utils.poll_for_condition(
condition=(lambda: self.manager_client.has_proxy()),
desc='Wait for manager daemon to come up',
sleep_interval=0.5,
timeout=self.DAEMON_TIMEOUT_SEC)
except Exception as e:
logging.error('timeout: error starting manager daemon: %s', e)
# We need to observe callbacks for proper operation.
if not self.manager_client.register_callbacks():
logging.error('manager_client: Failed to register callbacks')
return False
# Floss may not yet be enabled so make sure to enable it here.
if self.manager_client.get_floss_enabled() != enabled:
self.manager_client.set_floss_enabled(enabled)
default_adapter = self.manager_client.get_default_adapter()
try:
utils.poll_for_condition(
condition=(lambda: self.manager_client.
get_adapter_enabled(default_adapter
) == enabled),
desc='Wait for set floss enabled to complete',
sleep_interval=0.5,
timeout=self.DAEMON_TIMEOUT_SEC)
except Exception as e:
logging.error('timeout: error waiting for set_floss_enabled')
# Also configure cras to enable/disable floss
self.configure_cras_floss(enabled)
return True
def configure_cras_floss(self, enabled):
"""Configure whether CRAS has floss enabled."""
cras_utils.set_floss_enabled(enabled)
def _restart_cras(self, enable_floss=False):
"""Restarts CRAS and sets whether Floss is enabled."""
UpstartClient.stop(self.CRAS_JOB)
started = UpstartClient.start(self.CRAS_JOB)
def _set_floss():
try:
self.configure_cras_floss(enable_floss)
return True
except:
return False
try:
if started:
utils.poll_for_condition(
condition=_set_floss,
desc='Wait for CRAS to come up and configure floss',
sleep_interval=1,
timeout=self.DAEMON_TIMEOUT_SEC)
except Exception as e:
logging.error('timeout: error waiting to set floss on cras')
return False
# Did we successfully start the cras daemon?
return started
def log_message(self, msg):
""" log a message to /var/log/messages."""
try:
cmd = ['logger', msg]
subprocess.call(cmd)
except Exception as e:
logging.error("log_message %s failed with %s", cmd, str(e))
def messages_start(self):
"""Start messages monitoring.
@returns: True if logging started successfully, else False
"""
try:
self.messages.StartRecording()
return True
except Exception as e:
logging.error('Failed to start log recording with error: %s', e)
return False
def messages_stop(self):
"""Stop messages monitoring.
@returns: True if logs were successfully gathered since logging started,
else False
"""
try:
self.messages.StopRecording()
return True
except Exception as e:
logging.error('Failed to stop log recording with error: %s', e)
return False
def messages_find(self, pattern_str):
"""Find if a pattern string exists in messages output.
@param pattern_str: the pattern string to find.
@returns: True on success. False otherwise.
"""
return self.messages.LogContains(pattern_str)
def clean_bluetooth_kernel_log(self, log_level):
"""Remove Bluetooth kernel logs in /var/log/messages with loglevel
equal to or greater than |log_level|
@param log_level: int in range [0..7]
"""
reg_exp = '[^ ]+ ({LEVEL}) kernel: \[.*\] Bluetooth: .*'.format(
LEVEL='|'.join(self.SYSLOG_LEVELS[log_level:]))
logging.debug('Set kernel filter to level %d', log_level)
self.messages.FilterOut(reg_exp)
def _encode_base64_json(self, data):
"""Base64 encode and json encode the data.
Required to handle non-ascii data
@param data: data to be base64 and JSON encoded
@return: base64 and JSON encoded data
"""
logging.debug('_encode_base64_json raw data is %s', data)
b64_encoded = utils.base64_recursive_encode(data)
logging.debug('base64 encoded data is %s', b64_encoded)
json_encoded = json.dumps(b64_encoded)
logging.debug('JSON encoded data is %s', json_encoded)
return json_encoded
def is_wrt_supported(self):
"""Check if Bluetooth adapter support WRT logs
WRT is supported on Intel adapters other than (StP2 and WP2)
@returns : True if adapter is Intel made.
"""
# Dict of Intel Adapters that support WRT and vid:pid
vid_pid_dict = {
'HrP2': '8086:02f0',
'ThP2': '8086:2526',
'JfP2': '8086:31dc',
'JfP2-2': '8086:9df0'
} # On Sarien/Arcada
def _get_lspci_vid_pid(output):
""" parse output of lspci -knn and get the vid:pid
output is of the form '01:00.0 Network controller [0280]:
\Intel Corporation Device [8086:2526] (rev 29)\n'
@returns : 'vid:pid' or None
"""
try:
for i in output.split(b'\n'):
if 'Network controller' in i.decode('utf-8'):
logging.debug('Got line %s', i)
if 'Intel Corporation' in i.decode('utf-8'):
return i.split(b'[')[2].split(b']')[0]
return None
except Exception as e:
logging.debug('Exception in _get_lspci_vidpid %s', str(e))
return None
try:
cmd = ['lspci', '-knn']
output = subprocess.check_output(cmd, encoding='UTF-8')
vid_pid = _get_lspci_vid_pid(output)
logging.debug("got vid_pid %s", vid_pid)
if vid_pid is not None:
if vid_pid in list(vid_pid_dict.values()):
return True
except Exception as e:
logging.error('is_intel_adapter failed with %s', cmd, str(e))
return False
def __execute_cmd(self, cmd_str, msg=''):
"""Wrapper around subprocess.check_output.
@params cmd: Command to be executed as a string
@params msg: Optional description of the command
@returns: (True, output) if execution succeeded
(False, None) if execution failed
"""
try:
logging.info('Executing %s cmd', msg)
cmd = cmd_str.split(' ')
logging.debug('command is "%s"', cmd)
output = subprocess.check_output(cmd, encoding='UTF-8')
logging.info('%s cmd successfully executed', msg)
logging.debug('output is %s', output)
return (True, output)
except Exception as e:
logging.error('Exception %s while executing %s command', str(e),
msg)
return (False, None)
def read_inquiry_mode(self):
"""Read inquiry mode of DUT"""
read_inquiry_mode_cmd = 'hciconfig hci0 inqmode'
def _extract_inquiry_mode(hciconfig_str):
if 'Extended Inquiry' in hciconfig_str:
return 2
if 'RSSI' in hciconfig_str:
return 1
if 'Standard' in hciconfig_str:
return 0
return -1 # not recognized
try:
logging.info('Read inquiry mode')
status, output = self.__execute_cmd(read_inquiry_mode_cmd,
'Read inquiry mode')
if not status:
logging.info('Read inquiry mode command execution failed')
return -1
logging.info('Inquiry mode returned')
code = _extract_inquiry_mode(output)
if code == -1:
logging.error('hciconfig inquiry mode not recognized')
return code
except Exception as e:
logging.error('Exception %s while getting inquiry mode', str(e))
return -1
def write_inquiry_mode(self, mode=2):
"""Write inquiry mode to DUT"""
if mode > 2 or mode < 0:
logging.info('Wrong inquiry mode value: out of range [0-2]')
return False
write_inquiry_mode_cmd = 'hciconfig hci0 inqmode ' + str(mode)
try:
logging.info('Write inquiry mode')
status, _ = self.__execute_cmd(write_inquiry_mode_cmd,
'Write inquiry mode')
if not status:
logging.info('Write inquiry mode command execution failed')
return False
return True
except Exception as e:
logging.error('Exception %s while setting inquiry mode', str(e))
return False
def enable_wrt_logs(self):
""" Enable WRT logs for Intel Bluetooth adapters.
This is applicable only to Intel adapters.
Execute a series of custom hciconfig commands to
setup WRT log collection
Precondition :
1) Check if the DUT has Intel controller other than StP2
2) Make sure the controller is powered on
"""
fw_trace_cmd = (
'hcitool cmd 3f 7c 01 10 00 00 00 FE 81 02 80 04 00 00'
' 00 01 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00'
' 00 00 00 00 00 00 00')
ddc_read_cmd = 'hcitool cmd 3f 8c 28 01'
ddc_write_cmd_prefix = 'hcitool cmd 3f 8b 03 28 01'
hw_trace_cmd = (
'hcitool cmd 3f 6f 01 08 00 00 00 00 00 00 00 00 01 00'
' 00 03 01 03 03 03 10 03 6A 0A 6A 0A 6A 0A 6A 0A 00 00'
' 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00'
' 00 00 00 00 00 00')
multi_comm_trace_str = ('000000F600000000005002000000003F3F3F3'
'F3F003F000000000000000001000000000000000000'
'000000000000000000000000000000000000000000'
'00000000000000000000000000000000000000000'
'00000000000000000')
multi_comm_trace_file = ('/sys/kernel/debug/ieee80211'
'/phy0/iwlwifi/iwlmvm/send_hcmd')
def _get_ddc_write_cmd(ddc_read_result, ddc_write_cmd_prefix):
""" Create ddc_write_cmd from read command
This function performs the following
1) Take the output of ddc_read_cmd which is in following form
'< HCI Command: ogf 0x3f, ocf 0x008c, plen 1\n
01 \n>
HCI Event: 0x0e plen 6\n 01 8C FC 12 00 18 \n'
2) Take the last value of the output
01 8C FC 12 00 ===>> 18 <====
3) Bitwise or with 0x40
0x18 | 0x40 = 0x58
4) Add it to the end of the ddc_write_cmd
'hcitool 01 8C FC 00 28 01 ===> 58 <===='
"""
last_line = [
i for i in ddc_read_result.strip().split(b'\n') if i != ''
][-1]
last_byte = [i for i in last_line.split(b' ') if i != ''][-1]
processed_byte = hex(int(last_byte, 16) | 0x40).split('0x')[1]
cmd = ddc_write_cmd_prefix + ' ' + processed_byte
logging.debug('ddc_write_cmd is %s', cmd)
return cmd
try:
logging.info('Enabling WRT logs')
status, _ = self.__execute_cmd(fw_trace_cmd, 'FW trace cmd')
if not status:
logging.info('FW trace command execution failed')
return False
status, ddc_read_result = self.__execute_cmd(
ddc_read_cmd, 'DDC Read')
if not status:
logging.info('DDC Read command execution failed')
return False
ddc_write_cmd = _get_ddc_write_cmd(ddc_read_result,
ddc_write_cmd_prefix)
logging.debug('DDC Write command is %s', ddc_write_cmd)
status, _ = self.__execute_cmd(ddc_write_cmd, 'DDC Write')
if not status:
logging.info('DDC Write commanad execution failed')
return False
status, hw_trace_result = self.__execute_cmd(
hw_trace_cmd, 'HW trace')
if not status:
logging.info('HW Trace command execution failed')
return False
logging.debug('Executing the multi_comm_trace cmd %s to file %s',
multi_comm_trace_str, multi_comm_trace_file)
with open(multi_comm_trace_file, 'w') as f:
f.write(multi_comm_trace_str + '\n')
f.flush()
logging.info('WRT Logs enabled')
return True
except Exception as e:
logging.error('Exception %s while enabling WRT logs', str(e))
return False
def collect_wrt_logs(self):
"""Collect the WRT logs for Intel Bluetooth adapters
This is applicable only to Intel adapters.
Execute following command to collect WRT log. The logs are
copied to /var/spool/crash/
'echo 1 > sudo tee /sys/kernel/debug/ieee80211/phy0'
'/iwlwifi/iwlmvm/fw_dbg_collect'
This is to be called only after enable_wrt_logs is called
Precondition:
1) enable_wrt_logs has been called
"""
def _collect_logs():
"""Execute command to collect wrt logs."""
try:
with open(
'/sys/kernel/debug/ieee80211/phy0/iwlwifi/'
'iwlmvm/fw_dbg_collect', 'w') as f:
f.write('1')
f.flush()
# There is some flakiness in log collection. This sleep
# is due to the flakiness
time.sleep(10)
return True
except Exception as e:
logging.error('Exception %s in _collect logs ', str(e))
return False
def _get_num_log_files():
"""Return number of WRT log files."""
try:
return len(glob.glob('/var/spool/crash/devcoredump_iwlwifi*'))
except Exception as e:
logging.debug('Exception %s raised in _get_num_log_files',
str(e))
return 0
try:
logging.info('Collecting WRT logs')
#
# The command to trigger the logs does seems to work always.
# As a workaround for this flakiness, execute it multiple times
# until a new log is created
#
num_logs_present = _get_num_log_files()
logging.debug('%s logs present', num_logs_present)
for i in range(10):
time.sleep(1)
logging.debug('Executing command to collect WRT logs ')
if _collect_logs():
logging.debug('Command to collect WRT logs executed')
else:
logging.debug('Command to collect WRT logs failed')
continue
if _get_num_log_files() > num_logs_present:
logging.info('Successfully collected WRT logs ')
return True
else:
logging.debug('Log file not written. Trying again')
logging.info('Unable to collect WRT logs')
return False
except Exception as e:
logging.error('Exception %s while collecting WRT logs', str(e))
return False
def _get_wake_enabled_path(self):
# Walk up the parents from hci0 sysfs path and find the first one with
# a power/wakeup property. Return that path (including power/wakeup).
# Resolve hci path to get full device path (i.e. w/ usb or uart)
search_at = os.path.realpath('/sys/class/bluetooth/hci0')
# Exit early if path doesn't exist
if not os.path.exists(search_at):
return None
# Walk up parents and try to find one with 'power/wakeup'
for _ in range(search_at.count('/') - 1):
search_at = os.path.normpath(os.path.join(search_at, '..'))
try:
path = os.path.join(search_at, 'power', 'wakeup')
with open(path, 'r') as f:
return path
except IOError:
# No power wakeup at the given location so keep going
continue
return None
def _is_wake_enabled(self):
search_at = self._get_wake_enabled_path()
if search_at is not None:
try:
with open(search_at, 'r') as f:
value = f.read()
logging.info('Power/wakeup found at {}: {}'.format(
search_at, value))
return 'enabled' in value
except IOError:
# Path was not readable
return False
logging.debug('No power/wakeup path found')
return False
def _set_wake_enabled(self, value):
path = self._get_wake_enabled_path()
if path is not None:
try:
with open(path, 'w') as f:
f.write('enabled' if value else 'disabled')
return True
except IOError:
# Path was not writeable
return False
return False
def is_wake_enabled(self):
"""Checks whether the bluetooth adapter has wake enabled.
This will walk through all parents of the hci0 sysfs path and try to
find one with a 'power/wakeup' entry and returns whether its value is
'enabled'.
@return True if 'power/wakeup' of an hci0 parent is 'enabled'
"""
enabled = self._is_wake_enabled()
return enabled
def set_wake_enabled(self, value):
"""Sets wake enabled to the value if path exists.
This will walk through all parents of the hci0 sysfs path and write the
value to the first one it finds.
@param value: Sets power/wakeup to "enabled" if value is true, else
"disabled"
@return True if it wrote value to a power/wakeup, False otherwise
"""
return self._set_wake_enabled(value)
def wait_for_hid_device(self, device_address, timeout, sleep_interval):
"""Waits for hid device with given device address.
@param device_address: Peripheral address
@param timeout: maximum number of seconds to wait
@param sleep_interval: time to sleep between polls
@return True if hid device found, False otherwise
"""
def _match_hid_to_device(hidpath, device_address):
"""Check if given hid syspath is for the given device address """
# If the syspath has a uniq property that matches the peripheral
# device's address, then it has matched
props = UdevadmInfo.GetProperties(hidpath)
if (props.get(b'uniq', b'').lower().decode() == device_address):
logging.info('Found hid device for address {} at {}'.format(
device_address, hidpath))
return True
else:
logging.info('Path {} is not right device.'.format(hidpath))
return False
def _hid_is_created(device_address):
existing_inputs = UdevadmTrigger(
subsystem_match=['input']).DryRun()
for entry in existing_inputs:
entry = entry.decode()
bt_hid = any([t in entry for t in ['uhid', 'hci']])
logging.info('udevadm trigger entry is {}: {}'.format(
bt_hid, entry))
if (bt_hid and _match_hid_to_device(entry,
device_address.lower())):
return True
return False
if timeout is None:
timeout = self.HID_TIMEOUT
if sleep_interval is None:
sleep_interval = self.HID_CHECK_SECS
method_name = 'wait_for_hid_device'
try:
utils.poll_for_condition(
condition=(lambda: _hid_is_created(device_address)),
timeout=timeout,
sleep_interval=sleep_interval,
desc=('Waiting for HID device to be created from %s' %
device_address))
return True
except utils.TimeoutError as e:
logging.error('%s: %s', method_name, e)
except Exception as e:
logging.error('%s: unexpected error: %s', method_name, e)
return False
def _powerd_last_resume_details(self, before=5, after=0):
""" Look at powerd logs for last suspend/resume attempt.
Note that logs are in reverse order (chronologically). Keep that in mind
for the 'before' and 'after' parameters.
@param before: Number of context lines before search item to show.
@param after: Number of context lines after search item to show.
@return Most recent lines containing suspend resume details or ''.
"""
event_file = '/var/log/power_manager/powerd.LATEST'
# Each powerd_suspend wakeup has a log "powerd_suspend returned 0",
# with the return code of the suspend. We search for the last
# occurrence in the log, and then find the collocated event_count log,
# indicating the wakeup cause. -B option for grep will actually grab the
# *next* 5 logs in time, since we are piping the powerd file backwards
# with tac command
resume_indicator = 'powerd_suspend returned'
cmd = 'tac {} | grep -A {} -B {} -m1 "{}"'.format(
event_file, after, before, resume_indicator)
try:
return utils.run(cmd).stdout
except error.CmdError:
logging.error('Could not locate recent suspend')
return ''
def bt_caused_last_resume(self):
"""Checks if last resume from suspend was caused by bluetooth
@return: True if BT wake path was cause of resume, False otherwise
"""
# When the resume cause is printed to powerd log, it omits the
# /power/wakeup portion of wake path
bt_wake_path = self._get_wake_enabled_path()
# If bluetooth does not have a valid wake path, it could not have caused
# the resume
if not bt_wake_path:
return False
bt_wake_path = bt_wake_path.replace('/power/wakeup', '')
last_resume_details = self._powerd_last_resume_details().rstrip(
'\n ').split('\n')
logging.debug('/var/log/power_manager/powerd.LATEST: 5 lines after '
'powerd_suspend returns:')
for l in last_resume_details[::-1]:
logging.debug(l)
# If BT caused wake, there will be a line describing the bt wake
# path's event_count before and after the resume
for line in last_resume_details:
if 'event_count' in line:
logging.info('Checking wake event: {}'.format(line))
if bt_wake_path in line:
logging.debug('BT event woke the DUT')
return True
return False
def find_last_suspend_via_powerd_logs(self):
""" Finds the last suspend attempt via powerd logs.
Finds the last suspend attempt using powerd logs by searching backwards
through the logs to find the latest entries with 'powerd_suspend'. If we
can't find a suspend attempt, we return None.
@return: Tuple (suspend start time, suspend end time, suspend result) or
None if we can't find a suspend attempt
"""
# Logs look like this (ignore newline):
# 2021-02-11T18:53:43.561880Z INFO powerd:
# [daemon.cc(724)] powerd_suspend returned 0
# ... stuff in between ...
# 2021-02-11T18:53:13.277695Z INFO powerd:
# [suspender.cc(574)] Starting suspend
# Date format for strptime and strftime
date_format = '%Y-%m-%dT%H:%M:%S.%fZ'
date_group_re = ('(?P<date>[0-9]+-[0-9]+-[0-9]+T'
'[0-9]+:[0-9]+:[0-9]+[.][0-9]+Z)\s')
finish_suspend_re = re.compile(
'^{date_regex}'
'.*daemon.*powerd_suspend returned '
'(?P<exitcode>[0-9]+)'.format(date_regex=date_group_re))
start_suspend_re = re.compile(
'^{date_regex}.*suspender.*'
'Starting suspend'.format(date_regex=date_group_re))
now = datetime.now()
last_resume_details = self._powerd_last_resume_details(before=0,
after=8)
if last_resume_details:
start_time, end_time, ret = None, None, None
try:
for line in last_resume_details.split('\n'):
logging.debug('Last suspend search: %s', line)
m = finish_suspend_re.match(line)
if m:
logging.debug('Found suspend end: date(%s) ret(%s)',
m.group('date'), m.group('exitcode'))
end_time = datetime.strptime(
m.group('date'),
date_format).replace(year=now.year)
ret = int(m.group('exitcode'))
m = start_suspend_re.match(line)
if m:
logging.debug('Found suspend start: date(%s)',
m.group('date'))
start_time = datetime.strptime(
m.group('date'),
date_format).replace(year=now.year)
break
if all([x is not None for x in [start_time, end_time, ret]]):
# Return dates in string format due to inconsistency between
# python2/3 usage on host and dut
return (start_time.strftime(self.OUT_DATE_FORMAT),
end_time.strftime(self.OUT_DATE_FORMAT), ret)
else:
logging.error(
'Failed to parse details from last suspend. %s %s %s',
str(start_time), str(end_time), str(ret))
except Exception as e:
logging.error('Failed to parse last suspend: %s', str(e))
else:
logging.error('No powerd_suspend attempt found')
return None
def do_suspend(self, seconds, expect_bt_wake):
"""Suspend DUT using the power manager.
@param seconds: The number of seconds to suspend the device.
@param expect_bt_wake: Whether we expect bluetooth to wake us from
suspend. If true, we expect this resume will occur early
@throws: SuspendFailure on resume with unexpected timing or wake source.
The raised exception will be handled as a non-zero retcode over the
RPC, signalling for the test to fail.
"""
early_wake = False
try:
sys_power.do_suspend(seconds)
except sys_power.SpuriousWakeupError:
logging.info('Early resume detected...')
early_wake = True
# Handle error conditions based on test expectations, whether resume
# was early, and cause of the resume
bt_caused_wake = self.bt_caused_last_resume()
logging.info('Cause for resume: {}'.format(
'BT' if bt_caused_wake else 'Not BT'))
if not expect_bt_wake and bt_caused_wake:
raise sys_power.SuspendFailure('BT woke us unexpectedly')
# TODO(b/160803597) - Uncomment when BT wake reason is correctly
# captured in powerd log.
#
# if expect_bt_wake and not bt_caused_wake:
# raise sys_power.SuspendFailure('BT should have woken us')
#
# if bt_caused_wake and not early_wake:
# raise sys_power.SuspendFailure('BT wake did not come early')
return True
def suspend_delay(self, suspend_delay_secs, suspend_delay_timeout_secs,
wakeup_timeout_secs):
"""Enforce a suspend delay before system suspending.
@param suspend_delay_secs: the suspend delay in seconds
@param suspend_delay_timeout_secs: the suspend delay timeout in seconds
@param wakeup_timeout_secs: the wakeup_timeout in seconds
"""
return power_suspend_delay.suspend_delay(suspend_delay_secs,
suspend_delay_timeout_secs,
wakeup_timeout_secs)
def get_wlan_vid_pid(self):
""" Return vendor id and product id of the wlan chip on BT/WiFi module
@returns: (vid,pid) on success; (None,None) on failure
"""
vid = None
pid = None
path_template = '/sys/class/net/%s/device/'
for dev_name in ['wlan0', 'mlan0']:
if os.path.exists(path_template % dev_name):
path_v = path_template % dev_name + 'vendor'
path_d = path_template % dev_name + 'device'
logging.debug('Paths are %s %s', path_v, path_d)
try:
vid = open(path_v).read().strip('\n')
pid = open(path_d).read().strip('\n')
break
except Exception as e:
logging.error('Exception %s while reading vid/pid', str(e))
logging.debug('returning vid:%s pid:%s', vid, pid)
return (vid, pid)
def get_bt_transport(self):
""" Return transport (UART/USB/SDIO) used by BT module
@returns: USB/UART/SDIO on success; None on failure
"""
try:
transport_str = os.path.realpath(
'/sys/class/bluetooth/hci0/device/driver/module')
logging.debug('transport is %s', transport_str)
transport = transport_str.split('/')[-1]
if transport == 'btusb':
return 'USB'
elif transport == 'hci_uart':
return 'UART'
elif transport in ['btmrvl_sdio', 'btmtksdio']:
return 'SDIO'
else:
return None
except Exception as e:
logging.error('Exception %s in get_bt_transport', str(e))
return None
def get_bt_module_name(self):
""" Return bluetooth module name for non-USB devices
@returns '' on failure. On success return chipset name, if found in
dict.Otherwise it returns the raw string read.
"""
# map the string read from device to chipset name
chipset_string_dict = {
'qcom,wcn3991-bt\x00': 'WCN3991',
'qcom,wcn6750-bt\x00': 'WCN6750',
}
hci_device = '/sys/class/bluetooth/hci0'
real_path = os.path.realpath(hci_device)
logging.debug('real path is %s', real_path)
if 'usb' in real_path:
return ''
device_path = os.path.join(real_path, 'device', 'of_node',
'compatible')
try:
chipset_string = open(device_path).read()
logging.debug('read string %s from %s', chipset_string,
device_path)
except Exception as e:
logging.error('Exception %s while reading from file', str(e),
device_path)
return ''
if chipset_string in chipset_string_dict:
return chipset_string_dict[chipset_string]
else:
logging.debug("Chipset not known. Returning %s", chipset_string)
return chipset_string
def get_chipset_name(self):
""" Get the name of BT/WiFi chipset on this host
@returns chipset name if successful else ''
"""
(vid, pid) = self.get_wlan_vid_pid()
logging.debug('Bluetooth module vid pid is %s %s', vid, pid)
transport = self.get_bt_transport()
logging.debug('Bluetooth transport is %s', transport)
if vid is None or pid is None:
# Controllers that aren't WLAN+BT combo chips does not expose
# Vendor ID/Product ID. Use alternate method.
# This will return one of ['WCN3991', ''] or a string containing
# the name of chipset read from DUT
return self.get_bt_module_name()
for name, l in self.CHIPSET_TO_VIDPID.items():
if ((vid, pid), transport) in l:
return name
return ''
def get_bt_usb_device_strs(self):
""" Return the usb endpoints for the bluetooth device, if they exist
We wish to be able to identify usb disconnect events that affect our
bluetooth operation. To do so, we must first identify the usb endpoint
that is associated with our bluetooth device.
@returns: Relevant usb endpoints for the bluetooth device,
i.e. ['1-1','1-1.2'] if they exist,
[] otherwise
"""
hci_device = '/sys/class/bluetooth/hci0'
real_path = os.path.realpath(hci_device)
# real_path for a usb bluetooth controller will look something like:
# ../../devices/pci0000:00/0000:00:14.0/usb1/1-4/1-4:1.0/bluetooth/hci0
if 'usb' not in real_path:
return []
logging.debug('Searching for usb path: {}'.format(real_path))
# Grab all numbered entries between 'usb' and 'bluetooth' descriptors
m = re.search(r'usb(.*)bluetooth', real_path)
if not m:
logging.error(
'Unable to extract usb dev from {}'.format(real_path))
return []
# Return the path as a list of individual usb descriptors
return m.group(1).split('/')
def get_bt_usb_disconnect_str(self):
""" Return the expected log error on USB disconnect
Locate the descriptor that will be used from the list of all usb
descriptors associated with our bluetooth chip, and format into the
expected string error for USB disconnect
@returns: string representing expected usb disconnect log entry if usb
device could be identified, None otherwise
"""
disconnect_log_template = 'usb {}: USB disconnect'
descriptors = self.get_bt_usb_device_strs()
# The usb disconnect log message seems to use the most detailed
# descriptor that does not use the ':1.0' entry
for d in sorted(descriptors, key=len, reverse=True):
if ':' not in d:
return disconnect_log_template.format(d)
return None
def get_device_utc_time(self):
""" Get the current device time in UTC. """
return datetime.utcnow().strftime(self.OUT_DATE_FORMAT)
def create_audio_record_directory(self, audio_record_dir):
"""Create the audio recording directory.
@param audio_record_dir: the audio recording directory
@returns: True on success. False otherwise.
"""
try:
if not os.path.exists(audio_record_dir):
os.makedirs(audio_record_dir)
return True
except Exception as e:
logging.error('Failed to create %s on the DUT: %s',
audio_record_dir, e)
return False
def start_capturing_audio_subprocess(self, audio_data, recording_device):
"""Start capturing audio in a subprocess.
@param audio_data: the audio test data
@param recording_device: which device recorded the audio,
possible values are 'recorded_by_dut' or 'recorded_by_peer'
@returns: True on success. False otherwise.
"""
audio_data = json.loads(audio_data)
return self._cras_test_client.start_capturing_subprocess(
audio_data[recording_device],
sample_format=audio_data['format'],
channels=audio_data['channels'],
rate=audio_data['rate'],
duration=audio_data['duration'])
def stop_capturing_audio_subprocess(self):
"""Stop capturing audio.
@returns: True on success. False otherwise.
"""
return self._cras_test_client.stop_capturing_subprocess()
def _generate_playback_file(self, audio_data):
"""Generate the playback file if it does not exist yet.
Some audio test files may be large. Generate them on the fly
to save the storage of the source tree.
@param audio_data: the audio test data
"""
if not os.path.exists(audio_data['file']):
data_format = dict(file_type='raw',
sample_format='S16_LE',
channel=audio_data['channels'],
rate=audio_data['rate'])
# Make the audio file a bit longer to handle any delay
# issue in capturing.
duration = audio_data['duration'] + 3
audio_test_data_module.GenerateAudioTestData(
data_format=data_format,
path=audio_data['file'],
duration_secs=duration,
volume_scale=audio_data.get('volume_scale'),
frequencies=audio_data['frequencies'])
logging.debug("Raw file generated: %s", audio_data['file'])
def start_playing_audio_subprocess(self, audio_data, pin_device=None):
"""Start playing audio in a subprocess.
@param audio_data: the audio test data.
@param pin_device: the device id to play audio.
@returns: True on success. False otherwise.
"""
audio_data = json.loads(audio_data)
self._generate_playback_file(audio_data)
try:
return self._cras_test_client.start_playing_subprocess(
audio_data['file'],
pin_device=pin_device,
channels=audio_data['channels'],
rate=audio_data['rate'],
duration=audio_data['duration'])
except Exception as e:
logging.error("start_playing_subprocess() failed: %s", str(e))
return False
def stop_playing_audio_subprocess(self):
"""Stop playing audio in the subprocess.
@returns: True on success. False otherwise.
"""
return self._cras_test_client.stop_playing_subprocess()
def play_audio(self, audio_data):
"""Play audio.
It blocks until it has completed playing back the audio.
@param audio_data: the audio test data
@returns: True on success. False otherwise.
"""
audio_data = json.loads(audio_data)
self._generate_playback_file(audio_data)
return self._cras_test_client.play(audio_data['file'],
channels=audio_data['channels'],
rate=audio_data['rate'],
duration=audio_data['duration'])
def check_audio_frames_legitimacy(self, audio_test_data, recording_device,
recorded_file):
"""Get the number of frames in the recorded audio file.
@param audio_test_data: the audio test data
@param recording_device: which device recorded the audio,
possible values are 'recorded_by_dut' or 'recorded_by_peer'
@param recorded_file: the recorded file name
@returns: True if audio frames are legitimate.
"""
if bool(recorded_file):
recorded_filename = recorded_file
else:
audio_test_data = json.loads(audio_test_data)
recorded_filename = audio_test_data[recording_device]
if recorded_filename.endswith('.raw'):
# Make sure that the recorded file does not contain all zeros.
filesize = os.path.getsize(recorded_filename)
cmd_str = 'cmp -s -n %d %s /dev/zero' % (filesize,
recorded_filename)
try:
result = subprocess.call(cmd_str.split())
return result != 0
except Exception as e:
logging.error("Failed: %s (%s)", cmd_str, str(e))
return False
else:
# The recorded wav file should not be empty.
wav_file = check_quality.WaveFile(recorded_filename)
return wav_file.get_number_frames() > 0
def convert_audio_sample_rate(self, input_file, out_file, test_data,
new_rate):
"""Convert audio file to new sample rate.
@param input_file: Path to file to upsample.
@param out_file: Path to create upsampled file.
@param test_data: Dictionary with information about file.
@param new_rate: New rate to upsample file to.
@returns: True if upsampling succeeded, False otherwise.
"""
test_data = json.loads(test_data)
logging.debug('Resampling file {} to new rate {}'.format(
input_file, new_rate))
convert_format(input_file,
test_data['channels'],
test_data['bit_width'],
test_data['rate'],
out_file,
test_data['channels'],
test_data['bit_width'],
new_rate,
1.0,
use_src_header=True,
use_dst_header=True)
return os.path.isfile(out_file)
def trim_wav_file(self,
in_file,
out_file,
new_duration,
test_data,
tolerance=0.1):
"""Trim long file to desired length.
Trims audio file to length by cutting out silence from beginning and
end.
@param in_file: Path to audio file to be trimmed.
@param out_file: Path to trimmed audio file to create.
@param new_duration: A float representing the desired duration of
the resulting trimmed file.
@param test_data: Dictionary containing information about the test file.
@param tolerance: (optional) A float representing the allowable
difference between trimmed file length and desired duration
@returns: True if file was trimmed successfully, False otherwise.
"""
test_data = json.loads(test_data)
trim_silence_from_wav_file(in_file, out_file, new_duration)
measured_length = get_file_length(out_file, test_data['channels'],
test_data['bit_width'],
test_data['rate'])
return abs(measured_length - new_duration) <= tolerance
def unzip_audio_test_data(self, tar_path, data_dir):
"""Unzip audio test data files.
@param tar_path: Path to audio test data tarball on DUT.
@oaram data_dir: Path to directory where to extract test data directory.
@returns: True if audio test data folder exists, False otherwise.
"""
logging.debug('Downloading audio test data on DUT')
# creates path to dir to extract test data to by taking name of the
# tarball without the extension eg. <dir>/file.ext to data_dir/file/
audio_test_dir = os.path.join(
data_dir,
os.path.split(tar_path)[1].split('.', 1)[0])
unzip_cmd = 'tar -xf {0} -C {1}'.format(tar_path, data_dir)
unzip_proc = subprocess.Popen(unzip_cmd.split(),
stdout=subprocess.PIPE,
stderr=subprocess.PIPE)
_, stderr = unzip_proc.communicate()
if stderr:
logging.error('Error occurred in unzipping audio data: {}'.format(
str(stderr)))
return False
return unzip_proc.returncode == 0 and os.path.isdir(audio_test_dir)
def convert_raw_to_wav(self, input_file, output_file, test_data):
"""Convert raw audio file to wav file.
@oaram input_file: the location of the raw file
@param output_file: the location to place the resulting wav file
@param test_data: the data for the file being converted
@returns: True if conversion was successful otherwise false
"""
test_data = json.loads(test_data)
convert_raw_file(input_file, test_data['channels'],
test_data['bit_width'], test_data['rate'],
output_file)
return os.path.isfile(output_file)
def get_primary_frequencies(self, audio_test_data, recording_device,
recorded_file):
"""Get primary frequencies of the audio test file.
@param audio_test_data: the audio test data
@param recording_device: which device recorded the audio,
possible values are 'recorded_by_dut' or 'recorded_by_peer'
@param recorded_file: the recorded file name
@returns: a list of primary frequencies of channels in the audio file
"""
audio_test_data = json.loads(audio_test_data)
if bool(recorded_file):
recorded_filename = recorded_file
else:
recorded_filename = audio_test_data[recording_device]
args = CheckQualityArgsClass(filename=recorded_filename,
rate=audio_test_data['rate'],
channel=audio_test_data['channels'],
bit_width=16)
raw_data, rate = check_quality.read_audio_file(args)
checker = check_quality.QualityChecker(raw_data, rate)
# The highest frequency recorded would be near 24 Khz
# as the max sample rate is 48000 in our tests.
# So let's set ignore_high_freq to be 48000.
checker.do_spectral_analysis(ignore_high_freq=48000,
check_quality=False,
quality_params=None)
spectra = checker._spectrals
primary_freq = [
float(spectra[i][0][0]) if spectra[i] else 0
for i in range(len(spectra))
]
primary_freq.sort()
return primary_freq
def enable_wbs(self, value):
"""Enable or disable wideband speech (wbs) per the value.
@param value: True to enable wbs.
@returns: True if the operation succeeds.
"""
return self._cras_test_client.enable_wbs(value)
def set_player_playback_status(self, status):
"""Set playback status for the registered media player.
@param status: playback status in string.
"""
return self._cras_test_client.set_player_playback_status(status)
def set_player_position(self, position):
"""Set media position for the registered media player.
@param position: position in micro seconds.
"""
return self._cras_test_client.set_player_position(position)
def set_player_metadata(self, metadata):
"""Set metadata for the registered media player.
@param metadata: dictionary of media metadata.
"""
return self._cras_test_client.set_player_metadata(metadata)
def set_player_length(self, length):
"""Set media length for the registered media player.
Media length is a part of metadata information. However, without
specify its type to int64. dbus-python will guess the variant type to
be int32 by default. Separate it from the metadata function to help
prepare the data differently.
@param length: length in micro seconds.
"""
return self._cras_test_client.set_player_length(length)
def select_input_device(self, device_name):
"""Select the audio input device.
@param device_name: the name of the Bluetooth peer device
@returns: True if the operation succeeds.
"""
return self._cras_test_client.select_input_device(device_name)
@dbus_safe(None)
def select_output_node(self, node_type):
"""Select the audio output node.
@param node_type: the node type of the Bluetooth peer device
@returns: True if the operation succeeds.
"""
return cras_utils.set_single_selected_output_node(node_type)
@dbus_safe(None)
def get_selected_output_device_type(self):
"""Get the selected audio output node type.
@returns: the node type of the selected output device.
"""
# Note: should convert the dbus.String to the regular string.
return str(cras_utils.get_selected_output_device_type())
@dbus_safe(None)
def get_device_id_from_node_type(self, node_type, is_input):
"""Gets device id from node type.
@param node_type: a node type defined in CRAS_NODE_TYPES.
@param is_input: True if the node is input. False otherwise.
@returns: a string for device id.
"""
return cras_utils.get_device_id_from_node_type(node_type, is_input)
def get_audio_thread_summary(self):
"""Dumps audio thread info.
@returns: a list of cras audio information.
"""
return cras_utils.get_audio_thread_summary()
def is_btmanagerd_present(self):
""" Check if /usr/bin/btmanagerd file is present
@returns: True if /usr/bin/btmanagerd is present and False if not
"""
return os.path.exists(self.BTMANGERD_FILE_PATH)
class BluezPairingAgent:
"""The agent handling the authentication process of bluetooth pairing.
BluezPairingAgent overrides RequestPinCode method to return a given pin code.
User can use this agent to pair bluetooth device which has a known
pin code.
TODO (josephsih): more pairing modes other than pin code would be
supported later.
"""
def __init__(self, bus, path, pin):
"""Constructor.
@param bus: system bus object.
@param path: Object path to register.
@param pin: Pin to respond with for |RequestPinCode|.
"""
self._pin = pin
self.path = path
self.obj = bus.register_object(path, self, None)
# D-Bus service definition (required by pydbus).
dbus = """
<node>
<interface name="org.bluez.Agent1">
<method name="RequestPinCode">
<arg type="o" name="device_path" direction="in" />
<arg type="s" name="response" direction="out" />
</method>
<method name="AuthorizeService">
<arg type="o" name="device_path" direction="in" />
<arg type="s" name="uuid" direction="in" />
<arg type="b" name="response" direction="out" />
</method>
</interface>
</node>
"""
def unregister(self):
"""Unregisters self from bus."""
self.obj.unregister()
def RequestPinCode(self, device_path):
"""Requests pin code for a device.
Returns the known pin code for the request.
@param device_path: The object path of the device.
@returns: The known pin code.
"""
logging.info('RequestPinCode for %s; return %s', device_path,
self._pin)
return self._pin
def AuthorizeService(self, device_path, uuid):
"""Authorize given service for device.
@param device_path: The object path of the device.
@param uuid: The service that needs to be authorized.
@returns: True (we authorize everything since this is a test)
"""
return True
class BluezFacadeLocal(BluetoothBaseFacadeLocal):
"""Exposes DUT methods called remotely during Bluetooth autotests for the
Bluez daemon.
All instance methods of this object without a preceding '_' are exposed via
an XML-RPC server. This is not a stateless handler object, which means that
if you store state inside the delegate, that state will remain around for
future calls.
"""
BLUETOOTHD_JOB = 'bluetoothd'
DBUS_ERROR_SERVICEUNKNOWN = 'org.freedesktop.DBus.Error.ServiceUnknown'
BLUEZ_SERVICE_NAME = 'org.bluez'
BLUEZ_MANAGER_PATH = '/'
BLUEZ_DEBUG_LOG_PATH = '/org/chromium/Bluetooth'
BLUEZ_DEBUG_LOG_IFACE = 'org.chromium.Bluetooth.Debug'
BLUEZ_MANAGER_IFACE = 'org.freedesktop.DBus.ObjectManager'
BLUEZ_ADAPTER_IFACE = 'org.bluez.Adapter1'
BLUEZ_ADMIN_POLICY_SET_IFACE = 'org.bluez.AdminPolicySet1'
BLUEZ_ADMIN_POLICY_STATUS_IFACE = 'org.bluez.AdminPolicyStatus1'
BLUEZ_BATTERY_IFACE = 'org.bluez.Battery1'
BLUEZ_DEVICE_IFACE = 'org.bluez.Device1'
BLUEZ_GATT_SERV_IFACE = 'org.bluez.GattService1'
BLUEZ_GATT_CHAR_IFACE = 'org.bluez.GattCharacteristic1'
BLUEZ_GATT_DESC_IFACE = 'org.bluez.GattDescriptor1'
BLUEZ_LE_ADVERTISING_MANAGER_IFACE = 'org.bluez.LEAdvertisingManager1'
BLUEZ_ADV_MONITOR_MANAGER_IFACE = 'org.bluez.AdvertisementMonitorManager1'
BLUEZ_AGENT_MANAGER_PATH = '/org/bluez'
BLUEZ_AGENT_MANAGER_IFACE = 'org.bluez.AgentManager1'
BLUEZ_PROFILE_MANAGER_PATH = '/org/bluez'
BLUEZ_PROFILE_MANAGER_IFACE = 'org.bluez.ProfileManager1'
BLUEZ_ERROR_ALREADY_EXISTS = 'org.bluez.Error.AlreadyExists'
BLUEZ_PLUGIN_DEVICE_IFACE = 'org.chromium.BluetoothDevice'
DBUS_PROP_IFACE = 'org.freedesktop.DBus.Properties'
AGENT_PATH = '/test/agent'
BTMON_STOP_DELAY_SECS = 3
# Timeout for how long we'll wait for BlueZ and the Adapter to show up
# after reset.
ADAPTER_TIMEOUT = 30
# How long we should wait for property update signal before we cancel it.
PROPERTY_UPDATE_TIMEOUT_MILLI_SECS = 5000
# How often we should check for property update exit.
PROPERTY_UPDATE_CHECK_MILLI_SECS = 500
def __init__(self):
# Init the BaseFacade first
super(BluezFacadeLocal, self).__init__()
# Open the Bluetooth Raw socket to the kernel which provides us direct,
# raw, access to the HCI controller.
self._raw = bluetooth_socket.BluetoothRawSocket()
# Open the Bluetooth Control socket to the kernel which provides us
# raw management access to the Bluetooth Host Subsystem. Read the list
# of adapter indexes to determine whether or not this device has a
# Bluetooth Adapter or not.
self._control = bluetooth_socket.BluetoothControlSocket()
self._has_adapter = len(self._control.read_index_list()) > 0
# Create an Advertisement Monitor App Manager instance.
# This needs to be created before making any dbus connections as
# AdvMonitorAppMgr internally forks a new helper process and due to
# a limitation of python, it is not possible to fork a new process
# once any dbus connections are established.
self.advmon_appmgr = adv_monitor_helper.AdvMonitorAppMgr()
# Set up the connection to the D-Bus System Bus, get the object for
# the Bluetooth Userspace Daemon (BlueZ) and that daemon's object for
# the Bluetooth Adapter, and the advertising manager.
self.bus = pydbus.SystemBus()
self._update_bluez()
self._update_adapter()
self._update_advertising()
self._update_adv_monitor_manager()
# The agent to handle pin code request, which will be
# created when user calls pair_legacy_device method.
self._pairing_agent = None
# The default capability of the agent.
self._capability = 'KeyboardDisplay'
# Initialize a btmon object to record bluetoothd's activity.
self.btmon = output_recorder.OutputRecorder(
['btmon', '-c', 'never'],
stop_delay_secs=self.BTMON_STOP_DELAY_SECS)
self.advertisements = []
self.advmon_interleave_logger = logger_helper.InterleaveLogger()
self._chrc_property = None
self._timeout_id = 0
self._signal_watch = None
self._dbus_mainloop = GObject.MainLoop()
@dbus_safe(False)
def set_debug_log_levels(self, bluez_vb, kernel_vb):
"""Enable or disable the debug logs of bluetooth
@param bluez_vb: verbosity of bluez debug log, either 0 or 1
@param kernel_vb: verbosity of kernel debug log, either 0 or 1
"""
debug_object = self.bus.get(self.BLUEZ_SERVICE_NAME,
self.BLUEZ_DEBUG_LOG_PATH)
# Make a raw synchronous call using GLib (pydbus doesn't correctly
# serialize '(yy)'.
raw_dbus_call_sync(self.bus, debug_object, self.BLUEZ_DEBUG_LOG_IFACE,
'SetLevels',
GLib.Variant('(yy)', (bluez_vb, kernel_vb)),
GLib.VariantType.new('()'))
return
@dbus_safe(False)
def set_ll_privacy(self, enable):
"""Enable or disable the link layer privacy feature in the DUT.
If the controller is on, the command will first turn it off, set
the value, then turn it back on.
@param enable: True to enable
"""
debug_object = self.bus.get(
self.BLUEZ_SERVICE_NAME,
self.BLUEZ_DEBUG_LOG_PATH)[self.BLUEZ_DEBUG_LOG_IFACE]
debug_object.SetLLPrivacy(enable)
return True
@dbus_safe(False)
def set_quality_debug_log(self, enable):
"""Enable or disable bluez quality debug log in the DUT
@param enable: True to enable all of the debug log,
False to disable all of the debug log.
"""
bluez_debug = self.bus.get(
self.BLUEZ_SERVICE_NAME, self.BLUEZ_DEBUG_LOG_PATH)[
self.BLUEZ_DEBUG_LOG_IFACE]
bluez_debug.SetQualityDebug(enable)
@dbus_safe(False)
def set_quality_report(self, action):
"""Enable or disable the Bluetooth quality debug
@param action: 1 to enable the quality report
0 to disable the quality report.
"""
bluez_debug = self.bus.get(
self.BLUEZ_SERVICE_NAME, self.BLUEZ_DEBUG_LOG_PATH)[
self.BLUEZ_DEBUG_LOG_IFACE]
bluez_debug.SetQuality(action)
@dbus_safe(False)
def start_bluetoothd(self):
"""start bluetoothd.
This includes powering up the adapter.
@returns: True if bluetoothd is started correctly.
False otherwise.
"""
# Always start bluez tests with Floss disabled
self.configure_floss(enabled=False)
# Start the daemon and exit if that fails.
if not UpstartClient.start(self.BLUETOOTHD_JOB):
return False
logging.debug('waiting for bluez start')
try:
utils.poll_for_condition(condition=self._update_bluez,
desc='Bluetooth Daemon has started.',
timeout=self.ADAPTER_TIMEOUT)
except Exception as e:
logging.error('timeout: error starting bluetoothd: %s', e)
return False
# Waiting for the self._adapter object.
# This does not mean that the adapter is powered on.
logging.debug('waiting for bluez to obtain adapter information')
try:
utils.poll_for_condition(
condition=self._update_adapter,
desc='Bluetooth Daemon has adapter information.',
timeout=self.ADAPTER_TIMEOUT)
except Exception as e:
logging.error('timeout: error starting adapter: %s', e)
return False
# Waiting for the self._advertising interface object.
logging.debug('waiting for bluez to obtain interface manager.')
try:
utils.poll_for_condition(
condition=self._update_advertising,
desc='Bluetooth Daemon has advertising interface.',
timeout=self.ADAPTER_TIMEOUT)
except utils.TimeoutError:
logging.error('timeout: error getting advertising interface')
return False
# Register the pairing agent so we can authorize connections
logging.debug('registering default pairing agent')
self._setup_pairing_agent(0)
return True
@dbus_safe(False)
def stop_bluetoothd(self):
"""stop bluetoothd.
@returns: True if bluetoothd is stopped correctly.
False otherwise.
"""
def bluez_stopped():
"""Checks the bluetooth daemon status.
@returns: True if bluez is stopped. False otherwise.
"""
return not self._update_bluez()
# Stop the daemon and exit if that fails.
if not UpstartClient.stop(self.BLUETOOTHD_JOB):
return False
logging.debug('waiting for bluez stop')
try:
utils.poll_for_condition(condition=bluez_stopped,
desc='Bluetooth Daemon has stopped.',
timeout=self.ADAPTER_TIMEOUT)
bluetoothd_stopped = True
except Exception as e:
logging.error('timeout: error stopping bluetoothd: %s', e)
bluetoothd_stopped = False
return bluetoothd_stopped
def restart_cras(self):
"""Restarts the cras daemon."""
return self._restart_cras()
def is_bluetoothd_running(self):
"""Is bluetoothd running?
@returns: True if bluetoothd is running
"""
return bool(self._get_dbus_proxy_for_bluetoothd())
def is_bluetoothd_proxy_valid(self):
"""Checks whether the proxy object for bluetoothd is ok.
The dbus proxy object (self._bluez) can become unusable if bluetoothd
crashes or restarts for any reason. This method checks whether this has
happened by attempting to use the object proxy. If bluetoothd has
restarted (or is not available), then the session will no longer be
valid and this will result in a dbus exception (GLib.Error).
Returns:
True if the bluez proxy is still usable. False otherwise.
"""
try:
return self.is_bluetoothd_running() and bool(
self._objmgr_proxy) and bool(
self._objmgr_proxy.GetManagedObjects())
except GLib.Error:
return False
def _update_bluez(self):
"""Store a D-Bus proxy for the Bluetooth daemon in self._bluez.
This may be called in a loop until it returns True to wait for the
daemon to be ready after it has been started.
@return True on success, False otherwise.
"""
self._bluez = self._get_dbus_proxy_for_bluetoothd()
return bool(self._bluez)
@property
def _objmgr_proxy(self):
"""Returns proxy object to object manager if bluez is valid."""
if self._bluez:
return self._bluez[self.BLUEZ_MANAGER_IFACE]
return None
@dbus_safe(False)
def _get_dbus_proxy_for_bluetoothd(self):
"""Get the D-Bus proxy for the Bluetooth daemon.
@return True on success, False otherwise.
"""
bluez = None
try:
bluez = self.bus.get(self.BLUEZ_SERVICE_NAME,
self.BLUEZ_MANAGER_PATH)
logging.debug('bluetoothd is running')
except GLib.Error as e:
# When bluetoothd is not running, the exception looks like
# org.freedesktop.DBus.Error.ServiceUnknown: The name org.bluez
# was not provided by any .service files
if self.DBUS_ERROR_SERVICEUNKNOWN in str(e):
logging.debug('bluetoothd is not running')
else:
logging.error('Error getting dbus proxy for Bluez: %s', e)
return bluez
def _update_adapter(self):
"""Store a D-Bus proxy for the local adapter in self._adapter.
This may be called in a loop until it returns True to wait for the
daemon to be ready, and have obtained the adapter information itself,
after it has been started.
Since not all devices will have adapters, this will also return True
in the case where we have obtained an empty adapter index list from the
kernel.
Note that this method does not power on the adapter.
@return True on success, including if there is no local adapter,
False otherwise.
"""
self._adapter = None
self._adapter_path = None
# Re-check kernel to make sure adapter is available
self._has_adapter = len(self._control.read_index_list()) > 0
if self._bluez is None:
logging.warning('Bluez not found!')
return False
if not self._has_adapter:
logging.debug('Device has no adapter; returning')
return True
(self._adapter, self._adapter_path) = self._get_adapter()
return bool(self._adapter)
def _update_advertising(self):
"""Store a D-Bus proxy for the local advertising interface manager.
This may be called repeatedly in a loop until True is returned;
otherwise we wait for bluetoothd to start. After bluetoothd starts, we
check the existence of a local adapter and proceed to get the
advertisement interface manager.
Since not all devices will have adapters, this will also return True
in the case where there is no adapter.
@return True on success, including if there is no local adapter,
False otherwise.
"""
self._advertising = None
if self._bluez is None:
logging.warning('Bluez not found!')
return False
if not self._has_adapter:
logging.debug('Device has no adapter; returning')
return True
self._advertising = self._advertising_proxy
return bool(self._advertising)
def _update_adv_monitor_manager(self):
"""Store a D-Bus proxy for the local advertisement monitor manager.
This may be called repeatedly in a loop until True is returned;
otherwise we wait for bluetoothd to start. After bluetoothd starts, we
check the existence of a local adapter and proceed to get the
advertisement monitor manager interface.
Since not all devices will have adapters, this will also return True
in the case where there is no adapter.
@return True on success, including if there is no local adapter,
False otherwise.
"""
self._adv_monitor_manager = None
if self._bluez is None:
logging.warning('Bluez not found!')
return False
if not self._has_adapter:
logging.debug('Device has no adapter; returning without '
'advertisement monitor manager')
return True
self._adv_monitor_manager = self._get_adv_monitor_manager()
return bool(self._adv_monitor_manager)
@dbus_safe(False)
def _get_adapter(self):
"""Get the D-Bus proxy for the local adapter.
@return Tuple of (adapter, object_path) on success else (None, None).
"""
objects = self._objmgr_proxy.GetManagedObjects()
for path, ifaces in six.iteritems(objects):
logging.debug('%s -> %r', path, list(ifaces.keys()))
if self.BLUEZ_ADAPTER_IFACE in ifaces:
logging.debug('using adapter %s', path)
adapter = self.bus.get(self.BLUEZ_SERVICE_NAME, path)
return (adapter, path)
else:
logging.warning('No adapter found in interface!')
return (None, None)
@property
def _adapter_proxy(self):
"""Returns proxy object to adapter interface if adapter is valid."""
if self._adapter and self._get_adapter() != (None, None):
return self._adapter[self.BLUEZ_ADAPTER_IFACE]
return None
@property
def _property_proxy(self):
"""Returns proxy object to adapter properties if adapter is valid."""
if self._adapter and self._get_adapter() != (None, None):
return self._adapter[self.DBUS_PROP_IFACE]
return None
@property
def _advertising_proxy(self):
"""Returns proxy object to advertising interface if adapter is valid."""
if self._adapter and self._get_adapter() != (None, None):
return self._adapter[self.BLUEZ_LE_ADVERTISING_MANAGER_IFACE]
return None
@dbus_safe(False)
def _get_adv_monitor_manager(self):
"""Get the D-Bus proxy for the local advertisement monitor manager.
@return the advertisement monitor manager interface object.
"""
if self._adapter and self._get_adapter() != (None, None):
return self._adapter[self.BLUEZ_ADV_MONITOR_MANAGER_IFACE]
return None
@dbus_safe(False)
def reset_on(self):
"""Reset the adapter and settings and power up the adapter.
@return True on success, False otherwise.
"""
return self._reset(set_power=True)
@dbus_safe(False)
def reset_off(self):
"""Reset the adapter and settings, leave the adapter powered off.
@return True on success, False otherwise.
"""
return self._reset(set_power=False)
def has_adapter(self):
"""Return if an adapter is present.
This will only return True if we have determined both that there is
a Bluetooth adapter on this device (kernel adapter index list is not
empty) and that the Bluetooth daemon has exported an object for it.
@return True if an adapter is present, False if not.
"""
return self._has_adapter and self._adapter is not None
def _reset(self, set_power=False):
"""Remove remote devices and set adapter to set_power state.
Do not restart bluetoothd as this may incur a side effect.
The unhappy chrome may disable the adapter randomly.
@param set_power: adapter power state to set (True or False).
@return True on success, False otherwise.
"""
logging.debug('_reset')
if not self._adapter:
logging.warning('Adapter not found!')
return False
objects = self._objmgr_proxy.GetManagedObjects()
devices = []
for path, ifaces in six.iteritems(objects):
if self.BLUEZ_DEVICE_IFACE in ifaces:
devices.append(objects[path][self.BLUEZ_DEVICE_IFACE])
# Turn on the adapter in order to remove all remote devices.
if not self.is_powered_on():
if not self.set_powered(True):
logging.warning('Unable to power on the adapter')
return False
for device in devices:
logging.debug('removing %s', device.get('Address'))
self.remove_device_object(device.get('Address'))
# Toggle power to the adapter.
if not self.set_powered(False):
logging.warning('Unable to power off adapter')
return False
if set_power and not self.set_powered(True):
logging.warning('Unable to power on adapter')
return False
return True
@dbus_safe(False)
def is_discoverable(self):
"""Returns whether the adapter is discoverable."""
return bool(self._get_adapter_properties().get('Discoverable', 0) == 1)
@dbus_safe(False)
def set_powered(self, powered):
"""Set the adapter power state.
@param powered: adapter power state to set (True or False).
@return True on success, False otherwise.
"""
if not self._property_proxy:
if not powered:
# Return success if we are trying to power off an adapter that's
# missing or gone away, since the expected result has happened.
return True
else:
logging.warning('Adapter Property Proxy not found!')
return False
logging.debug('_set_powered %r', powered)
self._property_proxy.Set(self.BLUEZ_ADAPTER_IFACE, 'Powered',
GLib.Variant('b', powered))
return True
@dbus_safe(False)
def set_discoverable(self, discoverable):
"""Set the adapter discoverable state.
@param discoverable: adapter discoverable state to set (True or False).
@return True on success, False otherwise.
"""
if not self._property_proxy:
if not discoverable:
# Return success if we are trying to make an adapter that's
# missing or gone away, undiscoverable, since the expected
# result has happened.
return True
return False
self._property_proxy.Set(self.BLUEZ_ADAPTER_IFACE, 'Discoverable',
GLib.Variant('b', discoverable))
return True
@dbus_safe(False)
def get_discoverable_timeout(self):
"""Get the adapter discoverable_timeout.
@return discoverable timeout on success, None otherwise.
"""
if not self._property_proxy:
return None
return int(
self._property_proxy.Get(self.BLUEZ_ADAPTER_IFACE,
'DiscoverableTimeout'))
@dbus_safe(False)
def set_discoverable_timeout(self, discoverable_timeout):
"""Set the adapter discoverable_timeout property.
@param discoverable_timeout: adapter discoverable_timeout value
in seconds to set (Integer).
@return True on success, False otherwise.
"""
if not self._property_proxy:
return False
self._property_proxy.Set(self.BLUEZ_ADAPTER_IFACE,
'DiscoverableTimeout',
GLib.Variant('u', discoverable_timeout))
return True
@dbus_safe(False)
def get_pairable_timeout(self):
"""Get the adapter pairable_timeout.
@return pairable timeout on success, None otherwise.
"""
if not self._property_proxy:
return None
return int(
self._property_proxy.Get(self.BLUEZ_ADAPTER_IFACE,
'PairableTimeout'))
@dbus_safe(False)
def set_pairable_timeout(self, pairable_timeout):
"""Set the adapter pairable_timeout property.
@param pairable_timeout: adapter pairable_timeout value
in seconds to set (Integer).
@return True on success, False otherwise.
"""
if not self._property_proxy:
return False
self._property_proxy.Set(self.BLUEZ_ADAPTER_IFACE, 'PairableTimeout',
GLib.Variant('u', pairable_timeout))
return True
@dbus_safe(False)
def get_pairable(self):
"""Gets the adapter pairable state.
@return Pairable property value on success, None otherwise.
"""
if not self._property_proxy:
return None
return bool(
self._property_proxy.Get(self.BLUEZ_ADAPTER_IFACE, 'Pairable'))
@dbus_safe(False)
def set_pairable(self, pairable):
"""Set the adapter pairable state.
@param pairable: adapter pairable state to set (True or False).
@return True on success, False otherwise.
"""
if not self._property_proxy:
return False
self._property_proxy.Set(self.BLUEZ_ADAPTER_IFACE, 'Pairable',
GLib.Variant('b', pairable))
return True
@dbus_safe(False)
def set_adapter_alias(self, alias):
"""Set the adapter alias.
@param alias: adapter alias to set with type String
@return True on success, False otherwise.
"""
if not self._property_proxy:
return False
self._property_proxy.Set(self.BLUEZ_ADAPTER_IFACE, 'Alias',
GLib.Variant('s', alias))
return True
def _get_adapter_properties(self):
"""Read the adapter properties from the Bluetooth Daemon.
@return the properties as a JSON-encoded dictionary on success,
an empty dict otherwise.
"""
@dbus_safe({})
def get_props():
"""Get props from dbus."""
objects = self._objmgr_proxy.GetManagedObjects()
try:
return objects[self._adapter_path][self.BLUEZ_ADAPTER_IFACE]
except KeyError:
logging.warning('Failed to find adapter property')
return {}
except:
raise
if self._bluez and self._adapter:
props = get_props().copy()
else:
props = {}
logging.debug('get_adapter_properties')
for i in props.items():
logging.debug(i)
return props
def get_adapter_properties(self):
return json.dumps(self._get_adapter_properties())
def is_powered_on(self):
"""Checks whether the adapter is currently powered."""
return bool(self._get_adapter_properties().get('Powered', False))
def get_address(self):
"""Gets the current bluez adapter address."""
return str(self._get_adapter_properties().get('Address', ''))
def get_bluez_version(self):
"""Get the BlueZ version.
Returns:
Bluez version like 'BlueZ 5.39'.
"""
return str(self._get_adapter_properties().get('Name', ''))
def get_bluetooth_class(self):
"""Get the bluetooth class of the adapter.
Example for Chromebook: 4718852
Returns:
Class of device for the adapter.
"""
return str(self._get_adapter_properties().get('Class', ''))
def read_version(self):
"""Read the version of the management interface from the Kernel.
@return the information as a JSON-encoded tuple of:
( version, revision )
"""
#TODO(howardchung): resolve 'cannot allocate memory' error when
# BluetoothControlSocket idle too long(about 3 secs)
# (b:137603211)
_control = bluetooth_socket.BluetoothControlSocket()
return json.dumps(_control.read_version())
def read_supported_commands(self):
"""Read the set of supported commands from the Kernel.
@return the information as a JSON-encoded tuple of:
( commands, events )
"""
#TODO(howardchung): resolve 'cannot allocate memory' error when
# BluetoothControlSocket idle too long(about 3 secs)
# (b:137603211)
_control = bluetooth_socket.BluetoothControlSocket()
return json.dumps(_control.read_supported_commands())
def read_index_list(self):
"""Read the list of currently known controllers from the Kernel.
@return the information as a JSON-encoded array of controller indexes.
"""
#TODO(howardchung): resolve 'cannot allocate memory' error when
# BluetoothControlSocket idle too long(about 3 secs)
# (b:137603211)
_control = bluetooth_socket.BluetoothControlSocket()
return json.dumps(_control.read_index_list())
def read_info(self):
"""Read the adapter information from the Kernel.
@return the information as a JSON-encoded tuple of:
( address, bluetooth_version, manufacturer_id,
supported_settings, current_settings, class_of_device,
name, short_name )
"""
#TODO(howardchung): resolve 'cannot allocate memory' error when
# BluetoothControlSocket idle too long(about 3 secs)
# (b:137603211)
_control = bluetooth_socket.BluetoothControlSocket()
return json.dumps(_control.read_info(0))
def add_device(self, address, address_type, action):
"""Add a device to the Kernel action list.
@param address: Address of the device to add.
@param address_type: Type of device in @address.
@param action: Action to take.
@return on success, a JSON-encoded typle of:
( address, address_type ), None on failure.
"""
#TODO(howardchung): resolve 'cannot allocate memory' error when
# BluetoothControlSocket idle too long(about 3 secs)
# (b:137603211)
_control = bluetooth_socket.BluetoothControlSocket()
return json.dumps(_control.add_device(0, address, address_type,
action))
def remove_device(self, address, address_type):
"""Remove a device from the Kernel action list.
@param address: Address of the device to remove.
@param address_type: Type of device in @address.
@return on success, a JSON-encoded typle of:
( address, address_type ), None on failure.
"""
#TODO(howardchung): resolve 'cannot allocate memory' error when
# BluetoothControlSocket idle too long(about 3 secs)
# (b:137603211)
_control = bluetooth_socket.BluetoothControlSocket()
return json.dumps(_control.remove_device(0, address, address_type))
@dbus_safe(False)
def _get_devices(self):
"""Read information about remote devices known to the adapter.
@return the properties of each device in a list
"""
objects = self._objmgr_proxy.GetManagedObjects()
devices = []
for path, ifaces in six.iteritems(objects):
if self.BLUEZ_DEVICE_IFACE in ifaces:
devices.append(objects[path][self.BLUEZ_DEVICE_IFACE])
return devices
def _encode_json(self, data):
"""Encodes input data as JSON object.
Note that for bytes elements in the input data, they are decoded as
unicode string.
@param data: data to be JSON encoded
@return: JSON encoded data
"""
logging.debug('_encode_json raw data is %s', data)
str_data = utils.bytes_to_str_recursive(data)
json_encoded = json.dumps(str_data)
logging.debug('JSON encoded data is %s', json_encoded)
return json_encoded
def get_devices(self):
"""Read information about remote devices known to the adapter.
@return the properties of each device as a JSON-encoded array of
dictionaries on success, the value False otherwise.
"""
devices = self._get_devices()
# Note that bluetooth facade now runs in Python 3.
# Refer to crrev.com/c/3268347.
return self._encode_json(devices)
def get_num_connected_devices(self):
""" Return number of remote devices currently connected to the DUT.
@returns: The number of devices known to bluez with the Connected
property active
"""
num_connected_devices = 0
for dev in self._get_devices():
if dev and dev.get('Connected', False):
num_connected_devices += 1
return num_connected_devices
@dbus_safe(None)
def get_device_property(self, address, prop_name):
"""Read a property of BT device by directly querying device dbus object
@param address: Address of the device to query
@param prop_name: Property to be queried
@return Base 64 JSON repr of property if device is found and has
property, otherwise None on failure. JSON is a recursive
converter, automatically converting dbus types to python natives
and base64 allows us to pass special characters over xmlrpc.
Decode is done in bluetooth_device.py
"""
prop_val = None
# Grab dbus object, _find_device will catch any thrown dbus error
device_obj = self._find_device(address)
if device_obj:
# Query dbus object for property
prop_val = unpack_if_variant(device_obj[self.DBUS_PROP_IFACE].Get(
self.BLUEZ_DEVICE_IFACE, prop_name))
return self._encode_json(prop_val)
@dbus_safe(None)
def get_battery_property(self, address, prop_name):
"""Read a property from Battery1 interface.
@param address: Address of the device to query
@param prop_name: Property to be queried
@return The battery percentage value, or None if does not exist.
"""
prop_val = None
# Grab dbus object, _find_battery will catch any thrown dbus error
battery_obj = self._find_battery(address)
if battery_obj:
# Query dbus object for property
prop_val = unpack_if_variant(battery_obj[self.DBUS_PROP_IFACE].Get(
self.BLUEZ_BATTERY_IFACE, prop_name))
return prop_val
@dbus_safe(False)
def set_discovery_filter(self, filter):
"""Set the discovery filter.
@param filter: The discovery filter to set.
@return True on success, False otherwise.
"""
if not self._adapter_proxy:
return False
converted_filter = {}
for key in filter:
converted_filter[key] = GLib.Variant('s', filter[key])
self._adapter_proxy.SetDiscoveryFilter(converted_filter)
return True
@dbus_safe(False, return_error=True)
def start_discovery(self):
"""Start discovery of remote devices.
Obtain the discovered device information using get_devices(), called
stop_discovery() when done.
@return True on success, False otherwise.
"""
if not self._adapter_proxy:
return (False, "Adapter Not Found")
self._adapter_proxy.StartDiscovery()
return (True, None)
@dbus_safe(False, return_error=True)
def stop_discovery(self):
"""Stop discovery of remote devices.
@return True on success, False otherwise.
"""
if not self._adapter_proxy:
return (False, "Adapter Not Found")
self._adapter_proxy.StopDiscovery()
return (True, None)
def is_discovering(self):
"""Check if adapter is discovering."""
return self._get_adapter_properties().get('Discovering', 0) == 1
def get_supported_le_roles(self):
"""Returns the supported LE roles of the adapter.
@return: List of str indicates the supported LE roles.
"""
return self._get_adapter_properties().get('Roles', [])
def get_dev_info(self):
"""Read raw HCI device information.
@return JSON-encoded tuple of:
(index, name, address, flags, device_type, bus_type,
features, pkt_type, link_policy, link_mode,
acl_mtu, acl_pkts, sco_mtu, sco_pkts,
err_rx, err_tx, cmd_tx, evt_rx, acl_tx, acl_rx,
sco_tx, sco_rx, byte_rx, byte_tx) on success,
None on failure.
"""
return json.dumps(self._raw.get_dev_info(0))
@dbus_safe(None, return_error=True)
def get_supported_capabilities(self):
""" Get supported capabilities of the adapter
@returns (capabilities, None) on Success. (None, <error>) on failure
"""
if not self._adapter_proxy:
return (None, "Adapter Not Found")
value = self._adapter_proxy.GetSupportedCapabilities()
return (json.dumps(value), None)
@dbus_safe(False)
def register_profile(self, path, uuid, options):
"""Register new profile (service).
@param path: Path to the profile object.
@param uuid: Service Class ID of the service as string.
@param options: Dictionary of options for the new service, compliant
with BlueZ D-Bus Profile API standard.
@return True on success, False otherwise.
"""
converted_options = {}
if 'ServiceRecord' in options:
converted_options['ServiceRecord'] = GLib.Variant(
's', options['ServiceRecord'])
profile_manager = self.bus.get(
self.BLUEZ_SERVICE_NAME, self.BLUEZ_PROFILE_MANAGER_PATH)[
self.BLUEZ_PROFILE_MANAGER_IFACE]
profile_manager.RegisterProfile(path, uuid, converted_options)
return True
def has_device(self, address):
"""Checks if the device with a given address exists.
@param address: Address of the device.
@returns: True if there is an interface object with that address.
False if the device is not found.
@raises: Exception if a D-Bus error is encountered.
"""
result = self._find_device(address)
logging.debug('has_device result: %s', str(result))
# The result being False indicates that there is a D-Bus error.
if result is False:
raise Exception('dbus.Interface error')
# Return True if the result is not None, e.g. a D-Bus interface object;
# False otherwise.
return bool(result)
@dbus_safe(False)
def _find_device(self, address):
"""Finds the device with a given address.
Find the device with a given address and returns the
device interface.
@param address: Address of the device.
@returns: An 'org.bluez.Device1' interface to the device.
None if device can not be found.
"""
path = self._get_device_path(address)
if path:
return self.bus.get(self.BLUEZ_SERVICE_NAME, path)
logging.info('Device not found')
return None
@dbus_safe(None)
def _find_battery(self, address):
"""Finds the battery with a given address.
Find the battery with a given address and returns the
battery interface.
@param address: Address of the device.
@returns: An 'org.bluez.Battery1' interface to the device.
None if device can not be found.
"""
path = self._get_device_path(address)
if path:
try:
obj = self.bus.get(self.BLUEZ_SERVICE_NAME, path)
if obj[self.BLUEZ_BATTERY_IFACE] is not None:
return obj
except:
pass
logging.info('Battery not found')
return None
@dbus_safe(False)
def _get_device_path(self, address):
"""Gets the path for a device with a given address.
Find the device with a given address and returns the
the path for the device.
@param address: Address of the device.
@returns: The path to the address of the device, or None if device is
not found in the object tree.
"""
# Create device path, i.e. '/org/bluez/hci0/dev_AA_BB_CC_DD_EE_FF' based
# on path assignment scheme used in bluez
address_up = address.replace(':', '_')
device_path = '{}/dev_{}'.format(self._adapter_path, address_up)
# Verify the Address property agrees to confirm we have the device
try:
device = self.bus.get(self.BLUEZ_SERVICE_NAME, device_path)
found_addr = device[self.DBUS_PROP_IFACE].Get(
self.BLUEZ_DEVICE_IFACE, 'Address')
if found_addr == address:
logging.info('Device found at {}'.format(device_path))
return device_path
except KeyError as ke:
logging.debug('Couldn\'t reach device: %s: %s', address, ke)
except GLib.Error as e:
log_msg = 'Couldn\'t reach device: {}'.format(str(e))
logging.debug(log_msg)
logging.debug('No device found at {}'.format(device_path))
return None
@dbus_safe(False)
def _setup_pairing_agent(self, pin):
"""Initializes and resiters a BluezPairingAgent to handle authentication.
@param pin: The pin code this agent will answer.
"""
if self._pairing_agent:
logging.info(
'Removing the old agent before initializing a new one')
self._pairing_agent.unregister()
self._pairing_agent = None
# Create and register pairing agent
self._pairing_agent = BluezPairingAgent(self.bus, self.AGENT_PATH, pin)
agent_manager = self.bus.get(
self.BLUEZ_SERVICE_NAME,
self.BLUEZ_AGENT_MANAGER_PATH)[self.BLUEZ_AGENT_MANAGER_IFACE]
try:
# Make sure agent is accessible on bus
#agent_obj = self.bus.get(self.BLUEZ_SERVICE_NAME, self.AGENT_PATH)
agent_manager.RegisterAgent(self.AGENT_PATH, str(self._capability))
except GLib.Error as e:
if self.BLUEZ_ERROR_ALREADY_EXISTS in str(e):
logging.info('Unregistering old agent and registering the new')
agent_manager.UnregisterAgent(self.AGENT_PATH)
agent_manager.RegisterAgent(self.AGENT_PATH,
str(self._capability))
else:
logging.error('Error setting up pin agent: %s&