"""AndroidBluetoothDecorator class. This decorator is used for giving an AndroidDevice Bluetooth-specific functionality. """ import datetime import logging import os import queue import random import re import string import time from typing import Dict, Any, Text, Optional, Tuple, Sequence, Union, List from mobly import logger as mobly_logger from mobly import signals from mobly import utils from mobly.controllers import android_device from mobly.controllers.android_device_lib import adb from mobly.controllers.android_device_lib import jsonrpc_client_base from mobly.controllers.android_device_lib.services import sl4a_service from blueberry.controllers import derived_bt_device from blueberry.utils import bt_constants from blueberry.utils import ble_scan_adv_constants from blueberry.utils import bt_test_utils # Map for media passthrough commands and the corresponding events. MEDIA_CMD_MAP = { bt_constants.CMD_MEDIA_PAUSE: bt_constants.EVENT_PAUSE_RECEIVED, bt_constants.CMD_MEDIA_PLAY: bt_constants.EVENT_PLAY_RECEIVED, bt_constants.CMD_MEDIA_SKIP_PREV: bt_constants.EVENT_SKIP_PREV_RECEIVED, bt_constants.CMD_MEDIA_SKIP_NEXT: bt_constants.EVENT_SKIP_NEXT_RECEIVED } # Timeout for track change and playback state update in second. MEDIA_UPDATE_TIMEOUT_SEC = 3 # Timeout for the event of Media passthrough commands in second. MEDIA_EVENT_TIMEOUT_SEC = 1 BT_CONNECTION_WAITING_TIME_SECONDS = 10 ADB_WAITING_TIME_SECONDS = 1 # Common timeout for toggle status in seconds. COMMON_TIMEOUT_SECONDS = 5 # Local constant _DATETIME_FMT = '%m-%d %H:%M:%S.%f' # Interval time between ping requests in second. PING_INTERVAL_TIME_SEC = 2 # Timeout to wait for ping success in second. PING_TIMEOUT_SEC = 60 # A URL is used to verify internet by ping request. TEST_URL = 'http://www.google.com' # Timeout to wait for device boot success in second. WAIT_FOR_DEVICE_TIMEOUT_SEC = 180 class DeviceBootError(signals.ControllerError): """Exception raised for Android device boot failures.""" pass class Error(Exception): """Raised when an operation in this module fails.""" pass class DiscoveryError(signals.ControllerError): """Exception raised for Bluetooth device discovery failures.""" pass class AndroidBluetoothDecorator(android_device.AndroidDevice): """Decorates an AndroidDevice with Bluetooth-specific functionality.""" def __init__(self, ad: android_device.AndroidDevice): self._ad = ad self._user_params = None if not self._ad or not isinstance(self._ad, android_device.AndroidDevice): raise TypeError('Must apply AndroidBluetoothDecorator to an ' 'AndroidDevice') self.ble_advertise_callback = None self.regex_logcat_time = re.compile(r'(?P[\d]{2}-[\d]{2} [\d]{2}:[\d]{2}:[\d]{2}.[\d]{3})' r'[ ]+\d+.*') self._regex_bt_crash = re.compile(r'Bluetooth crashed (?P\d+) times') def __getattr__(self, name: Any) -> Any: return getattr(self._ad, name) def _is_device_connected(self, mac_address): """Wrapper method to help with unit testability of this class.""" return self._ad.sl4a.bluetoothIsDeviceConnected(mac_address) def _is_profile_connected(self, mac_address, profile): """Checks if the profile is connected.""" status = None pri_ad = self._ad if profile == bt_constants.BluetoothProfile.HEADSET_CLIENT: status = pri_ad.sl4a.bluetoothHfpClientGetConnectionStatus(mac_address) elif profile == bt_constants.BluetoothProfile.A2DP_SINK: status = pri_ad.sl4a.bluetoothA2dpSinkGetConnectionStatus(mac_address) elif profile == bt_constants.BluetoothProfile.PBAP_CLIENT: status = pri_ad.sl4a.bluetoothPbapClientGetConnectionStatus(mac_address) elif profile == bt_constants.BluetoothProfile.MAP_MCE: connected_devices = self._ad.sl4a.bluetoothMapClientGetConnectedDevices() return any(mac_address in device['address'] for device in connected_devices) else: pri_ad.log.warning('The connection check for profile %s is not supported ' 'yet', profile) return False return status == bt_constants.BluetoothConnectionStatus.STATE_CONNECTED def _get_bluetooth_le_state(self): """Wrapper method to help with unit testability of this class.""" return self._ad.sl4a.bluetoothGetLeState def _generate_id_by_size(self, size): """Generate string of random ascii letters and digits. Args: size: required size of string. Returns: String of random chars. """ return ''.join(random.choice(string.ascii_letters + string.digits) for _ in range(size)) def _wait_for_bluetooth_manager_state(self, state=None, timeout=10, threshold=5): """Waits for Bluetooth normalized state or normalized explicit state. Args: state: expected Bluetooth state timeout: max timeout threshold threshold: list len of bt state Returns: True if successful, false if unsuccessful. """ all_states = [] start_time = time.time() while time.time() < start_time + timeout: all_states.append(self._get_bluetooth_le_state()) if len(all_states) >= threshold: # for any normalized state if state is None: if len(all_states[-threshold:]) == 1: logging.info('State normalized %s', all_states[-threshold:]) return True else: # explicit check against normalized state if state in all_states[-threshold:]: return True time.sleep(0.5) logging.error('Bluetooth state fails to normalize' if state is None else 'Failed to match bluetooth state, current state {} expected state {}'. format(self._get_bluetooth_le_state(), state)) return False def init_setup(self) -> None: """Sets up android device for bluetooth tests.""" self._ad.services.register('sl4a', sl4a_service.Sl4aService) self._ad.load_snippet('mbs', 'com.google.android.mobly.snippet.bundled') self._ad.adb.shell('setenforce 0') # Adds 2 seconds waiting time to see it can fix the NullPointerException # when executing the following sl4a.bluetoothStartPairingHelper method. time.sleep(2) self._ad.sl4a.bluetoothStartPairingHelper() self.factory_reset_bluetooth() def sl4a_setup(self) -> None: """A common setup routine for android device sl4a function. Things this method setup: 1. Set Bluetooth local name to random string of size 4 2. Disable BLE background scanning. """ sl4a = self._ad.sl4a sl4a.bluetoothStartConnectionStateChangeMonitor('') setup_result = sl4a.bluetoothSetLocalName(self._generate_id_by_size(4)) if not setup_result: self.log.error('Failed to set device name.') return sl4a.bluetoothDisableBLE() bonded_devices = sl4a.bluetoothGetBondedDevices() for b in bonded_devices: self.log.info('Removing bond for device {}'.format(b['address'])) sl4a.bluetoothUnbond(b['address']) def set_user_params(self, params: Dict[str, Any]) -> None: self._user_params = params def get_user_params(self) -> Dict[str, Any]: return self._user_params def is_sim_state_loaded(self) -> bool: """Checks if SIM state is loaded. Returns: True if SIM state is loaded else False. """ state = self._ad.adb.shell('getprop gsm.sim.state').decode().strip() return state == 'LOADED' def is_package_installed(self, package_name: str) -> bool: """Checks if a package is installed. Args: package_name: string, a package to be checked. Returns: True if the package is installed else False. """ # The package is installed if result is 1, not installed if result is 0. result = int(self._ad.adb.shell('pm list packages | grep -i %s$ | wc -l' % package_name)) return bool(result) def connect_with_rfcomm(self, other_ad: android_device.AndroidDevice) -> bool: """Establishes an RFCOMM connection with other android device. Connects this android device (as a client) to the other android device (as a server). Args: other_ad: the Android device accepting the connection from this device. Returns: True if connection was successful, False if unsuccessful. """ server_address = other_ad.sl4a.bluetoothGetLocalAddress() logging.info('Pairing and connecting devices') if not self._ad.sl4a.bluetoothDiscoverAndBond(server_address): logging.info('Failed to pair and connect devices') return False # Create RFCOMM connection logging.info('establishing RFCOMM connection') return self.orchestrate_rfcomm_connection(other_ad) def orchestrate_rfcomm_connection(self, other_ad: android_device.AndroidDevice, accept_timeout_ms: int = bt_constants.DEFAULT_RFCOMM_TIMEOUT_MS, uuid: Optional[Text] = None) -> bool: """Sets up the RFCOMM connection to another android device. It sets up the connection with a Bluetooth Socket connection with other device. Args: other_ad: the Android device accepting the connection from this device. accept_timeout_ms: the timeout in ms for the connection. uuid: universally unique identifier. Returns: True if connection was successful, False if unsuccessful. """ if uuid is None: uuid = bt_constants.BT_RFCOMM_UUIDS['default_uuid'] other_ad.sl4a.bluetoothStartPairingHelper() self._ad.sl4a.bluetoothStartPairingHelper() other_ad.sl4a.bluetoothSocketConnBeginAcceptThreadUuid(uuid, accept_timeout_ms) self._ad.sl4a.bluetoothSocketConnBeginConnectThreadUuid(other_ad.sl4a.bluetoothGetLocalAddress(), uuid) end_time = time.time() + bt_constants.BT_DEFAULT_TIMEOUT_SECONDS test_result = True while time.time() < end_time: number_socket_connections = len(other_ad.sl4a.bluetoothSocketConnActiveConnections()) connected = number_socket_connections > 0 if connected: test_result = True other_ad.log.info('Bluetooth socket Client Connection Active') break else: test_result = False time.sleep(1) if not test_result: other_ad.log.error('Failed to establish a Bluetooth socket connection') return False return True def wait_for_discovery_success(self, mac_address: str, timeout: float = 30) -> float: """Waits for a device to be discovered by AndroidDevice. Args: mac_address: The Bluetooth mac address of the peripheral device. timeout: Number of seconds to wait for device discovery. Returns: discovery_time: The time it takes to pair in seconds. Raises: DiscoveryError """ device_start_time = self.get_device_time() start_time = time.time() event_name = f'Discovery{mac_address}' try: self._ad.ed.wait_for_event(event_name, lambda x: x['data']['Status'], timeout) discovery_time = time.time() - start_time return discovery_time except queue.Empty: # TODO(user): Remove this check when this bug is fixed. if self.logcat_filter(device_start_time, event_name): self._ad.log.info('Actually the event "%s" was posted within %d seconds.', event_name, timeout) return timeout raise DiscoveryError('Failed to discover device %s after %d seconds' % (mac_address, timeout)) def wait_for_pairing_success(self, mac_address: str, timeout: float = 30) -> float: """Waits for a device to pair with the AndroidDevice. Args: mac_address: The Bluetooth mac address of the peripheral device. timeout: Number of seconds to wait for the devices to pair. Returns: pairing_time: The time it takes to pair in seconds. Raises: ControllerError """ start_time = time.time() try: self._ad.ed.wait_for_event('Bond%s' % mac_address, lambda x: x['data']['Status'], timeout) pairing_time = time.time() - start_time return pairing_time except queue.Empty: raise signals.ControllerError('Failed to bond with device %s after %d seconds' % (mac_address, timeout)) def wait_for_connection_success(self, mac_address: str, timeout: int = 30) -> float: """Waits for a device to connect with the AndroidDevice. Args: mac_address: The Bluetooth mac address of the peripheral device. timeout: Number of seconds to wait for the devices to connect. Returns: connection_time: The time it takes to connect in seconds. Raises: ControllerError """ start_time = time.time() end_time = start_time + timeout while time.time() < end_time: if self._is_device_connected(mac_address): connection_time = (time.time() - start_time) logging.info('Connected device %s in %d seconds', mac_address, connection_time) return connection_time raise signals.ControllerError('Failed to connect device within %d seconds.' % timeout) def factory_reset_bluetooth(self) -> None: """Factory resets Bluetooth on an AndroidDevice.""" logging.info('Factory resetting Bluetooth for AndroidDevice.') self._ad.sl4a.bluetoothToggleState(True) paired_devices = self._ad.mbs.btGetPairedDevices() for device in paired_devices: self._ad.sl4a.bluetoothUnbond(device['Address']) self._ad.sl4a.bluetoothFactoryReset() self._wait_for_bluetooth_manager_state() self.wait_for_bluetooth_toggle_state(True) def get_device_info(self) -> Dict[str, Any]: """Gets the configuration info of an AndroidDevice. Returns: dict, A dictionary mapping metric keys to their respective values. """ device_info = { 'device_class': 'AndroidDevice', 'device_model': self._ad.device_info['model'], 'hardware_version': self._ad.adb.getprop('ro.boot.hardware.revision'), 'software_version': self._ad.build_info['build_id'], 'android_build_type': self._ad.build_info['build_type'], 'android_build_number': self._ad.adb.getprop('ro.build.version.incremental'), 'android_release_id': self._ad.build_info['build_id'] } return device_info def pair_and_connect_bluetooth(self, mac_address: str, attempts: int = 3, enable_pairing_retry: bool = True) -> Tuple[float, float]: """Pairs and connects an AndroidDevice with a peripheral Bluetooth device. Ensures that an AndroidDevice is paired and connected to a peripheral device. If the devices are already connected, does nothing. If the devices are paired but not connected, connects the devices. If the devices are neither paired nor connected, this method pairs and connects the devices. Suggests to use the retry mechanism on Discovery because it sometimes fail even if the devices are testing in shielding. In order to avoid the remote device may not respond a incoming pairing request causing to bonding failure , it suggests to retry pairing too. Args: mac_address: The Bluetooth mac address of the peripheral device. attempts: Number of attempts to discover and pair the peripheral device. enable_pairing_retry: Bool to control whether the retry mechanism is used on bonding failure, it's enabled if True. Returns: pairing_time: The time, in seconds, it takes to pair the devices. connection_time: The time, in seconds, it takes to connect the devices after pairing is completed. Raises: DiscoveryError: Raised if failed to discover the peripheral device. ControllerError: Raised if failed to bond the peripheral device. """ connected = self._is_device_connected(mac_address) pairing_time = 0 connection_time = 0 if connected: logging.info('Device %s already paired and connected', mac_address) return pairing_time, connection_time paired_devices = [device['address'] for device in self._ad.sl4a.bluetoothGetBondedDevices()] if mac_address in paired_devices: self._ad.sl4a.bluetoothConnectBonded(mac_address) return pairing_time, self.wait_for_connection_success(mac_address) logging.info('Initiate pairing to the device "%s".', mac_address) for i in range(attempts): self._ad.sl4a.bluetoothDiscoverAndBond(mac_address) try: self.wait_for_discovery_success(mac_address) pairing_time = self.wait_for_pairing_success(mac_address) break except DiscoveryError: if i + 1 < attempts: logging.error('Failed to find the device "%s" on Attempt %d. ' 'Retrying discovery...', mac_address, i + 1) continue raise DiscoveryError('Failed to find the device "%s".' % mac_address) except signals.ControllerError: if i + 1 < attempts and enable_pairing_retry: logging.error('Failed to bond the device "%s" on Attempt %d. ' 'Retrying pairing...', mac_address, i + 1) continue raise signals.ControllerError('Failed to bond the device "%s".' % mac_address) connection_time = self.wait_for_connection_success(mac_address) return pairing_time, connection_time def disconnect_bluetooth(self, mac_address: str, timeout: float = 30) -> float: """Disconnects Bluetooth between an AndroidDevice and peripheral device. Args: mac_address: The Bluetooth mac address of the peripheral device. timeout: Number of seconds to wait for the devices to disconnect the peripheral device. Returns: disconnection_time: The time, in seconds, it takes to disconnect the peripheral device. Raises: ControllerError: Raised if failed to disconnect the peripheral device. """ if not self._is_device_connected(mac_address): logging.info('Device %s already disconnected', mac_address) return 0 self._ad.sl4a.bluetoothDisconnectConnected(mac_address) start_time = time.time() end_time = time.time() + timeout while time.time() < end_time: connected = self._is_device_connected(mac_address) if not connected: logging.info('Device %s disconnected successfully.', mac_address) return time.time() - start_time raise signals.ControllerError('Failed to disconnect device within %d seconds.' % timeout) def connect_bluetooth(self, mac_address: str, timeout: float = 30) -> float: """Connects Bluetooth between an AndroidDevice and peripheral device. Args: mac_address: The Bluetooth mac address of the peripheral device. timeout: Number of seconds to wait for the devices to connect the peripheral device. Returns: connection_time: The time, in seconds, it takes to connect the peripheral device. Raises: ControllerError: Raised if failed to connect the peripheral device. """ if self._is_device_connected(mac_address): logging.info('Device %s already connected', mac_address) return 0 self._ad.sl4a.bluetoothConnectBonded(mac_address) connect_time = self.wait_for_connection_success(mac_address) return connect_time def activate_pairing_mode(self) -> None: """Activates pairing mode on an AndroidDevice.""" logging.info('Activating pairing mode on AndroidDevice.') self._ad.sl4a.bluetoothMakeDiscoverable() self._ad.sl4a.bluetoothStartPairingHelper() def activate_ble_pairing_mode(self) -> None: """Activates BLE pairing mode on an AndroidDevice.""" self.ble_advertise_callback = self._ad.sl4a.bleGenBleAdvertiseCallback() self._ad.sl4a.bleSetAdvertiseDataIncludeDeviceName(True) # Sets advertise mode to low latency. self._ad.sl4a.bleSetAdvertiseSettingsAdvertiseMode(ble_scan_adv_constants.BleAdvertiseSettingsMode.LOW_LATENCY) self._ad.sl4a.bleSetAdvertiseSettingsIsConnectable(True) # Sets TX power level to High. self._ad.sl4a.bleSetAdvertiseSettingsTxPowerLevel(ble_scan_adv_constants.BleAdvertiseSettingsTxPower.HIGH) advertise_data = self._ad.sl4a.bleBuildAdvertiseData() advertise_settings = self._ad.sl4a.bleBuildAdvertiseSettings() logging.info('Activating BLE pairing mode on AndroidDevice.') self._ad.sl4a.bleStartBleAdvertising(self.ble_advertise_callback, advertise_data, advertise_settings) def deactivate_ble_pairing_mode(self) -> None: """Deactivates BLE pairing mode on an AndroidDevice.""" if not self.ble_advertise_callback: self._ad.log.debug('BLE pairing mode is not activated.') return logging.info('Deactivating BLE pairing mode on AndroidDevice.') self._ad.sl4a.bleStopBleAdvertising(self.ble_advertise_callback) self.ble_advertise_callback = None def get_bluetooth_mac_address(self) -> str: """Gets Bluetooth mac address of an AndroidDevice.""" logging.info('Getting Bluetooth mac address for AndroidDevice.') mac_address = self._ad.sl4a.bluetoothGetLocalAddress() logging.info('Bluetooth mac address of AndroidDevice: %s', mac_address) return mac_address def scan_and_get_ble_device_address(self, device_name: str, timeout_sec: float = 30) -> str: """Searchs a BLE device by BLE scanner and returns it's BLE mac address. Args: device_name: string, the name of BLE device. timeout_sec: int, number of seconds to wait for finding the advertisement. Returns: String of the BLE mac address. Raises: ControllerError: Raised if failed to get the BLE device address """ filter_list = self._ad.sl4a.bleGenFilterList() scan_settings = self._ad.sl4a.bleBuildScanSetting() scan_callback = self._ad.sl4a.bleGenScanCallback() self._ad.sl4a.bleSetScanFilterDeviceName(device_name) self._ad.sl4a.bleBuildScanFilter(filter_list) self._ad.sl4a.bleStartBleScan(filter_list, scan_settings, scan_callback) try: event = self._ad.ed.pop_event('BleScan%sonScanResults' % scan_callback, timeout_sec) except queue.Empty: raise signals.ControllerError('Timed out %ds after waiting for phone finding BLE device: %s.' % (timeout_sec, device_name)) finally: self._ad.sl4a.bleStopBleScan(scan_callback) return event['data']['Result']['deviceInfo']['address'] def get_device_name(self) -> str: """Gets Bluetooth device name of an AndroidDevice.""" logging.info('Getting Bluetooth device name for AndroidDevice.') device_name = self._ad.sl4a.bluetoothGetLocalName() logging.info('Bluetooth device name of AndroidDevice: %s', device_name) return device_name def is_bluetooth_sco_on(self) -> bool: """Checks whether communications use Bluetooth SCO.""" cmd = 'dumpsys bluetooth_manager | grep "isBluetoothScoOn"' get_status = self._ad.adb.shell(cmd) if isinstance(get_status, bytes): get_status = get_status.decode() return 'true' in get_status def connect_with_profile(self, snd_ad_mac_address: str, profile: bt_constants.BluetoothProfile) -> bool: """Connects with the profile. The connection can only be completed after the bluetooth devices are paired. To connected with the profile, the bluetooth connection policy is set to forbidden first and then set to allowed. The paired bluetooth devices will start to make connection. The connection time could be long. The waitting time is set to BT_CONNECTION_WAITING_TIME_SECONDS (currently 10 seconds). Args: snd_ad_mac_address: the mac address of the device accepting connection. profile: the profiles to be set Returns: The profile connection succeed/fail """ if profile == bt_constants.BluetoothProfile.MAP_MCE: self._ad.sl4a.bluetoothMapClientConnect(snd_ad_mac_address) elif profile == bt_constants.BluetoothProfile.PBAP_CLIENT: self.set_profile_policy(snd_ad_mac_address, profile, bt_constants.BluetoothConnectionPolicy.CONNECTION_POLICY_ALLOWED) self._ad.sl4a.bluetoothPbapClientConnect(snd_ad_mac_address) else: self.set_profile_policy(snd_ad_mac_address, profile, bt_constants.BluetoothConnectionPolicy.CONNECTION_POLICY_FORBIDDEN) self.set_profile_policy(snd_ad_mac_address, profile, bt_constants.BluetoothConnectionPolicy.CONNECTION_POLICY_ALLOWED) self._ad.sl4a.bluetoothConnectBonded(snd_ad_mac_address) time.sleep(BT_CONNECTION_WAITING_TIME_SECONDS) is_connected = self._is_profile_connected(snd_ad_mac_address, profile) self.log.info('The connection between %s and %s for profile %s succeed: %s', self.serial, snd_ad_mac_address, profile, is_connected) return is_connected def connect_to_snd_with_profile(self, snd_ad: android_device.AndroidDevice, profile: bt_constants.BluetoothProfile, attempts: int = 5) -> bool: """Connects pri android device to snd android device with profile. Args: snd_ad: android device accepting connection profile: the profile to be connected attempts: Number of attempts to try until failure. Returns: Boolean of connecting result """ pri_ad = self._ad curr_attempts = 0 snd_ad_mac_address = snd_ad.sl4a.bluetoothGetLocalAddress() if not self.is_bt_paired(snd_ad_mac_address): self.log.error('Devices %s and %s not paired before connecting', self.serial, snd_ad.serial) return False while curr_attempts < attempts: curr_attempts += 1 self.log.info('Connection of profile %s at curr attempt %d (total %d)', profile, curr_attempts, attempts) if self.connect_with_profile(snd_ad_mac_address, profile): self.log.info('Connection between devices %s and %s succeeds at %d try', pri_ad.serial, snd_ad.serial, curr_attempts) return True self.log.error('Connection of profile %s failed after %d attempts', profile, attempts) return False def is_bt_paired(self, mac_address: str) -> bool: """Check if the bluetooth device with mac_address is paired to ad. Args: mac_address: the mac address of the bluetooth device for pairing Returns: True if they are paired """ bonded_info = self._ad.sl4a.bluetoothGetBondedDevices() return mac_address in [info['address'] for info in bonded_info] def is_a2dp_sink_connected(self, mac_address: str) -> bool: """Checks if the Android device connects to a A2DP sink device. Args: mac_address: String, Bluetooth MAC address of the A2DP sink device. Returns: True if connected else False. """ connected_devices = self._ad.sl4a.bluetoothA2dpGetConnectedDevices() return mac_address in [d['address'] for d in connected_devices] def hfp_connect(self, ag_ad: android_device.AndroidDevice) -> bool: """Hfp connecting hf android device to ag android device. The android device should support the Headset Client profile. For example, the android device with git_master-bds-dev build. Args: ag_ad: Audio Gateway (ag) android device Returns: Boolean of connecting result """ return self.connect_to_snd_with_profile(ag_ad, bt_constants.BluetoothProfile.HEADSET_CLIENT) def a2dp_sink_connect(self, src_ad: android_device.AndroidDevice) -> bool: """Connects pri android device to secondary android device. The android device should support the A2dp Sink profile. For example, the android device with git_master-bds-dev build. Args: src_ad: A2dp source android device Returns: Boolean of connecting result """ return self.connect_to_snd_with_profile(src_ad, bt_constants.BluetoothProfile.A2DP_SINK) def map_connect(self, map_ad: android_device.AndroidDevice) -> bool: """Connects primary device to secondary device via MAP MCE profile. The primary device should support the MAP MCE profile. For example, the android device with git_master-bds-dev build. Args: map_ad: AndroidDevice, a android device supporting MAP profile. Returns: Boolean of connecting result """ return self.connect_to_snd_with_profile(map_ad, bt_constants.BluetoothProfile.MAP_MCE) def map_disconnect(self, bluetooth_address: str) -> bool: """Disconnects a MAP MSE device with specified Bluetooth MAC address. Args: bluetooth_address: a connected device's bluetooth address. Returns: True if the device is disconnected else False. """ self._ad.sl4a.bluetoothMapClientDisconnect(bluetooth_address) return bt_test_utils.wait_until(timeout_sec=COMMON_TIMEOUT_SECONDS, condition_func=self._is_profile_connected, func_args=[bluetooth_address, bt_constants.BluetoothProfile.MAP_MCE], expected_value=False) def pbap_connect(self, pbap_ad: android_device.AndroidDevice) -> bool: """Connects primary device to secondary device via PBAP client profile. The primary device should support the PBAP client profile. For example, the android device with git_master-bds-dev build. Args: pbap_ad: AndroidDevice, a android device supporting PBAP profile. Returns: Boolean of connecting result """ return self.connect_to_snd_with_profile(pbap_ad, bt_constants.BluetoothProfile.PBAP_CLIENT) def set_bluetooth_tethering(self, status_enabled: bool) -> None: """Sets Bluetooth tethering to be specific status. Args: status_enabled: Bool, Bluetooth tethering will be set to enable if True, else disable. """ if self._ad.sl4a.bluetoothPanIsTetheringOn() == status_enabled: self._ad.log.info('Already %s Bluetooth tethering.' % ('enabled' if status_enabled else 'disabled')) return self._ad.log.info('%s Bluetooth tethering.' % ('Enable' if status_enabled else 'Disable')) self._ad.sl4a.bluetoothPanSetBluetoothTethering(status_enabled) bt_test_utils.wait_until(timeout_sec=COMMON_TIMEOUT_SECONDS, condition_func=self._ad.sl4a.bluetoothPanIsTetheringOn, func_args=[], expected_value=status_enabled, exception=signals.ControllerError('Failed to %s Bluetooth tethering.' % ('enable' if status_enabled else 'disable'))) def set_profile_policy(self, snd_ad_mac_address: str, profile: bt_constants.BluetoothProfile, policy: bt_constants.BluetoothConnectionPolicy) -> None: """Sets policy of the profile car related profiles to OFF. This avoids autoconnect being triggered randomly. The use of this function is encouraged when you're testing individual profiles in isolation. Args: snd_ad_mac_address: the mac address of the device accepting connection. profile: the profiles to be set policy: the policy value to be set """ pri_ad = self._ad pri_ad_local_name = pri_ad.sl4a.bluetoothGetLocalName() pri_ad.log.info('Sets profile %s on %s for %s to policy %s', profile, pri_ad_local_name, snd_ad_mac_address, policy) if profile == bt_constants.BluetoothProfile.A2DP: pri_ad.sl4a.bluetoothA2dpSetPriority(snd_ad_mac_address, policy.value) elif profile == bt_constants.BluetoothProfile.A2DP_SINK: pri_ad.sl4a.bluetoothA2dpSinkSetPriority(snd_ad_mac_address, policy.value) elif profile == bt_constants.BluetoothProfile.HEADSET_CLIENT: pri_ad.sl4a.bluetoothHfpClientSetPriority(snd_ad_mac_address, policy.value) elif profile == bt_constants.BluetoothProfile.PBAP_CLIENT: pri_ad.sl4a.bluetoothPbapClientSetPriority(snd_ad_mac_address, policy.value) elif profile == bt_constants.BluetoothProfile.HID_HOST: pri_ad.sl4a.bluetoothHidSetPriority(snd_ad_mac_address, policy.value) else: pri_ad.log.error('Profile %s not yet supported for policy settings', profile) def set_profiles_policy(self, snd_ad: android_device.AndroidDevice, profile_list: Sequence[bt_constants.BluetoothProfile], policy: bt_constants.BluetoothConnectionPolicy) -> None: """Sets the policy of said profile(s) on pri_ad for snd_ad. Args: snd_ad: android device accepting connection profile_list: list of the profiles to be set policy: the policy to be set """ mac_address = snd_ad.sl4a.bluetoothGetLocalAddress() for profile in profile_list: self.set_profile_policy(mac_address, profile, policy) def set_profiles_policy_off(self, snd_ad: android_device.AndroidDevice, profile_list: Sequence[bt_constants.BluetoothProfile]) -> None: """Sets policy of the profiles to OFF. This avoids autoconnect being triggered randomly. The use of this function is encouraged when you're testing individual profiles in isolation Args: snd_ad: android device accepting connection profile_list: list of the profiles to be turned off """ self.set_profiles_policy(snd_ad, profile_list, bt_constants.BluetoothConnectionPolicy.CONNECTION_POLICY_FORBIDDEN) def wait_for_call_state(self, call_state: Union[int, bt_constants.CallState], timeout_sec: float, wait_interval: int = 3) -> bool: """Waits for call state of the device to be changed. Args: call_state: int, the expected call state. Call state values are: 0: IDLE 1: RINGING 2: OFFHOOK timeout_sec: int, number of seconds of expiration time wait_interval: int, number of seconds of waiting in each cycle Returns: True if the call state has been changed else False. """ # TODO(user): Force external call to use CallState instead of int if isinstance(call_state, bt_constants.CallState): call_state = call_state.value expiration_time = time.time() + timeout_sec which_cycle = 1 while time.time() < expiration_time: # Waits for the call state change in every cycle. time.sleep(wait_interval) self._ad.log.info('in cycle %d of waiting for call state %d', which_cycle, call_state) if call_state == self._ad.mbs.getTelephonyCallState(): return True self._ad.log.info('The call state did not change to %d before timeout', call_state) return False def add_call_log(self, call_log_type: Union[int, bt_constants.CallLogType], phone_number: str, call_time: int) -> None: """Add call number and time to specified log. Args: call_log_type: int, number of call log type. Call log type values are: 1: Incoming call 2: Outgoing call 3: Missed call phone_number: string, phone number to be added in call log. call_time: int, call time to be added in call log. Returns: None """ # TODO(user): Force external call to use CallLogType instead of int if isinstance(call_log_type, bt_constants.CallLogType): call_log_type = call_log_type.value new_call_log = {} new_call_log['type'] = str(call_log_type) new_call_log['number'] = phone_number new_call_log['time'] = str(call_time) self._ad.sl4a.callLogsPut(new_call_log) def get_call_volume(self) -> int: """Gets current call volume of an AndroidDevice when Bluetooth SCO On. Returns: An integer specifying the number of current call volume level. Raises: Error: If the pattern search failed. """ cmd = 'dumpsys audio | grep "STREAM_BLUETOOTH_SCO|STREAM_VOICE_CALL" | tail -1' out = self._ad.adb.shell(cmd).decode() pattern = r'(?<=SCO index:)\d+' result = re.search(pattern, out) if result is None: raise Error(f'Pattern "{pattern}" search failed, dump output: {out}') return int(result.group()) def make_phone_call(self, callee: android_device.AndroidDevice, timeout_sec: float = 30) -> None: """Make a phone call to callee and check if callee is ringing. Args: callee: AndroidDevice, The callee in the phone call. timeout_sec: int, number of seconds to wait for the callee ringing. Raises: TestError """ self._ad.sl4a.telecomCallNumber(callee.dimensions['phone_number']) is_ringing = callee.wait_for_call_state(bt_constants.CALL_STATE_RINGING, timeout_sec) if not is_ringing: raise signals.TestError('Timed out after %ds waiting for call state: RINGING' % timeout_sec) def wait_for_disconnection_success(self, mac_address: str, timeout: float = 30) -> float: """Waits for a device to connect with the AndroidDevice. Args: mac_address: The Bluetooth mac address of the peripheral device. timeout: Number of seconds to wait for the devices to connect. Returns: connection_time: The time it takes to connect in seconds. Raises: ControllerError """ start_time = time.time() end_time = start_time + timeout while time.time() < end_time: if not self._ad.sl4a.bluetoothIsDeviceConnected(mac_address): disconnection_time = (time.time() - start_time) logging.info('Disconnected device %s in %d seconds', mac_address, disconnection_time) return disconnection_time raise signals.ControllerError('Failed to disconnect device within %d seconds.' % timeout) def first_pair_and_connect_bluetooth(self, bt_device: Any) -> None: """Pairs and connects an AndroidDevice with a Bluetooth device. This method does factory reset bluetooth first and then pairs and connects the devices. Args: bt_device: A device object which implements basic Bluetooth function related methods. Returns: None """ bt_device.factory_reset_bluetooth() mac_address = bt_device.get_bluetooth_mac_address() bt_device.activate_pairing_mode() self.pair_and_connect_bluetooth(mac_address) def get_device_time(self) -> str: """Get device epoch time and transfer to logcat timestamp format. Returns: String of the device time. """ return self._ad.adb.shell('date +"%m-%d %H:%M:%S.000"').decode().splitlines()[0] def logcat_filter(self, start_time: str, text_filter: str = '') -> str: """Returns logcat after a given time. This method calls from the android_device logcat service file and filters all logcat line prior to the start_time. Args: start_time: start time in string format of _DATETIME_FMT. text_filter: only return logcat lines that include this string. Returns: A logcat output. Raises: ValueError Exception if start_time is invalid format. """ try: start_time_conv = datetime.datetime.strptime(start_time, _DATETIME_FMT) except ValueError as ex: logging.error('Invalid time format!') raise ex logcat_response = '' with open(self._ad.adb_logcat_file_path, 'r', errors='replace') \ as logcat_file: post_start_time = False for line in logcat_file: match = self.regex_logcat_time.match(line) if match: if (datetime.datetime.strptime(match.group('datetime'), _DATETIME_FMT) >= start_time_conv): post_start_time = True if post_start_time and line.find(text_filter) >= 0: logcat_response += line return logcat_response def logcat_filter_message(self, current_time: str, text: str = '') -> str: """DEPRECATED Builds the logcat command. This method builds the logcat command to check for a specified log message after the specified time. If text=None, the logcat returned will be unfiltered. Args: current_time: time cutoff for grepping for the specified message, format = ('%m-%d %H:%M:%S.000'). text: text to search for. Returns: The response of the logcat filter. """ return self.logcat_filter(current_time, text) def send_media_passthrough_cmd(self, command: str, event_receiver: Optional[android_device.AndroidDevice] = None) -> None: """Sends a media passthrough command. Args: command: string, media passthrough command. event_receiver: AndroidDevice, a device which starts BluetoothSL4AAudioSrcMBS. Raises: signals.ControllerError: raised if the event is not received. """ if event_receiver is None: event_receiver = self._ad self._ad.log.info('Sending Media Passthough: %s' % command) self._ad.sl4a.bluetoothMediaPassthrough(command) if not event_receiver: event_receiver = self._ad try: event_receiver.ed.pop_event(MEDIA_CMD_MAP[command], MEDIA_EVENT_TIMEOUT_SEC) except queue.Empty: raise signals.ControllerError('Device "%s" failed to receive the event "%s" ' 'when the command "%s" was sent.' % (event_receiver.serial, MEDIA_CMD_MAP[command], command)) def pause(self) -> None: """Sends the AVRCP command "pause".""" self.send_media_passthrough_cmd(bt_constants.CMD_MEDIA_PAUSE) def play(self) -> None: """Sends the AVRCP command "play".""" self.send_media_passthrough_cmd(bt_constants.CMD_MEDIA_PLAY) def track_previous(self) -> None: """Sends the AVRCP command "skipPrev".""" self.send_media_passthrough_cmd(bt_constants.CMD_MEDIA_SKIP_PREV) def track_next(self) -> None: """Sends the AVRCP command "skipNext".""" self.send_media_passthrough_cmd(bt_constants.CMD_MEDIA_SKIP_NEXT) def get_current_track_info(self) -> Dict[str, Any]: """Returns Dict (Media metadata) representing the current track.""" return self._ad.sl4a.bluetoothMediaGetCurrentMediaMetaData() def get_current_playback_state(self) -> int: """Returns Integer representing the current playback state.""" return self._ad.sl4a.bluetoothMediaGetCurrentPlaybackState()['state'] def verify_playback_state_changed(self, expected_state: str, exception: Optional[Exception] = None) -> bool: """Verifies the playback state is changed to be the expected state. Args: expected_state: string, the changed state as expected. exception: Exception, raised when the state is not changed if needed. """ bt_test_utils.wait_until(timeout_sec=MEDIA_UPDATE_TIMEOUT_SEC, condition_func=self.get_current_playback_state, func_args=[], expected_value=expected_state, exception=exception, interval_sec=1) def verify_current_track_changed(self, expected_track: str, exception: Optional[Exception] = None) -> bool: """Verifies the Now playing track is changed to be the expected track. Args: expected_track: string, the changed track as expected. exception: Exception, raised when the track is not changed if needed. """ bt_test_utils.wait_until(timeout_sec=MEDIA_UPDATE_TIMEOUT_SEC, condition_func=self.get_current_track_info, func_args=[], expected_value=expected_track, exception=exception, interval_sec=1) def verify_avrcp_event(self, event_name: bt_constants.AvrcpEvent, check_time: str, timeout_sec: float = 20) -> bool: """Verifies that an AVRCP event was received by an AndroidDevice. Checks logcat to verify that an AVRCP event was received after a given time. Args: event_name: enum, AVRCP event name. Currently supports play, pause, track_previous, and track_next. check_time: string, The earliest desired cutoff time to check the logcat. Must be in format '%m-%d %H:%M:%S.000'. Use datetime.datetime.now().strftime('%m-%d %H:%M:%S.%f') to get current time in this format. timeout_sec: int, Number of seconds to wait for the specified AVRCP event be found in logcat. Raises: TestError Returns: True if the event was received. """ avrcp_events = [ 'State:NOT_PLAYING->PLAYING', 'State:PLAYING->NOT_PLAYING', 'sendMediaKeyEvent: keyEvent=76', 'sendMediaKeyEvent: keyEvent=75' ] if event_name.value not in avrcp_events: raise signals.TestError('An unexpected AVRCP event is specified.') end_time = time.time() + timeout_sec while time.time() < end_time: if self.logcat_filter_message(check_time, event_name.value): logging.info('%s event received successfully.', event_name) return True time.sleep(1) logging.error('AndroidDevice failed to receive %s event.', event_name) logging.info('Logcat:\n%s', self.logcat_filter_message(check_time)) return False def add_google_account(self, retries: int = 5) -> bool: """Login Google account. Args: retries: int, the number of retries. Returns: True if account is added successfully. Raises: TestError """ for _ in range(retries): output = self._ad.adb.shell( 'am instrument -w -e account "%s" -e password ' '"%s" -e sync true -e wait-for-checkin false ' 'com.google.android.tradefed.account/.AddAccount' % (self._ad.dimensions['google_account'], self._ad.dimensions['google_account_password'])).decode() if 'result=SUCCESS' in output: logging.info('Google account is added successfully') time.sleep(3) # Wait for account to steady state return True raise signals.TestError('Failed to add google account: %s' % output) def remove_google_account(self, retries: int = 5) -> bool: """Remove Google account. Args: retries: int, the number of retries. Returns: True if account is removed successfully. Raises: TestError """ for _ in range(retries): output = self._ad.adb.shell('am instrument -w com.google.android.tradefed.account/.RemoveAccounts').decode() if 'result=SUCCESS' in output: logging.info('Google account is removed successfully') return True time.sleep(1) # Buffer between retries. raise signals.TestError('Failed to remove google account: %s' % output) def detect_and_pull_ssrdump(self, ramdump_type: str = 'ramdump_bt') -> bool: """Detect and pull RAMDUMP log. Args: ramdump_type: str, the partial of file names to search for in ramdump files path. 'ramdump_bt' is used for searching Bluetooth ramdump log files. Returns: True if there is a file with file name matching the ramdump type. """ files = self._ad.adb.shell('ls %s' % bt_constants.RAMDUMP_PATH).decode() if ramdump_type in files: logging.info('RAMDUMP is found.') log_name_timestamp = mobly_logger.get_log_file_timestamp() destination = os.path.join(self._ad.log_path, 'RamdumpLogs', log_name_timestamp) utils.create_dir(destination) self._ad.adb.pull([bt_constants.RAMDUMP_PATH, destination]) return True return False def get_bt_num_of_crashes(self) -> int: """Get number of Bluetooth crash times from bluetooth_manager. Returns: Number of Bluetooth crashed times. """ out = self._regex_bt_crash.search(self._ad.adb.shell('dumpsys bluetooth_manager').decode()) # TODO(user): Need to consider the case "out=None" when miss in # matching return int(out.group('num_bt_crashes')) def clean_ssrdump(self) -> None: """Clean RAMDUMP log. Returns: None """ self._ad.adb.shell('rm -rf %s/*' % bt_constants.RAMDUMP_PATH) def set_target(self, bt_device: derived_bt_device.BtDevice) -> None: """Allows for use to get target device object for target interaction.""" self._target_device = bt_device def wait_for_hsp_connection_state(self, mac_address: str, connected: bool, raise_error: bool = True, timeout_sec: float = 30) -> bool: """Waits for HSP connection to be in a expected state on Android device. Args: mac_address: The Bluetooth mac address of the peripheral device. connected: True if HSP connection state is connected as expected. raise_error: Error will be raised if True. timeout_sec: Number of seconds to wait for HSP connection state change. Returns: True if HSP connection state is the expected state. """ expected_state = bt_constants.BluetoothConnectionStatus.STATE_DISCONNECTED if connected: expected_state = bt_constants.BluetoothConnectionStatus.STATE_CONNECTED msg = ('Failed to %s the device "%s" within %d seconds via HSP.' % ('connect' if connected else 'disconnect', mac_address, timeout_sec)) return bt_test_utils.wait_until(timeout_sec=timeout_sec, condition_func=self._ad.sl4a.bluetoothHspGetConnectionStatus, func_args=[mac_address], expected_value=expected_state, exception=signals.TestError(msg) if raise_error else None) def wait_for_bluetooth_toggle_state(self, enabled: bool = True, timeout_sec: float = 30) -> bool: """Waits for Bluetooth to be in an expected state. Args: enabled: True if Bluetooth status is enabled as expected. timeout_sec: Number of seconds to wait for Bluetooth to be in the expected state. """ bt_test_utils.wait_until( timeout_sec=timeout_sec, condition_func=self._ad.mbs.btIsEnabled, func_args=[], expected_value=enabled, exception=signals.TestError('Bluetooth is not %s within %d seconds on the device "%s".' % ('enabled' if enabled else 'disabled', timeout_sec, self._ad.serial))) def wait_for_a2dp_connection_state(self, mac_address: str, connected: bool, raise_error: bool = True, timeout_sec: float = 30) -> bool: """Waits for A2DP connection to be in a expected state on Android device. Args: mac_address: The Bluetooth mac address of the peripheral device. connected: True if A2DP connection state is connected as expected. raise_error: Error will be raised if True. timeout_sec: Number of seconds to wait for A2DP connection state change. Returns: True if A2DP connection state is in the expected state. """ msg = ('Failed to %s the device "%s" within %d seconds via A2DP.' % ('connect' if connected else 'disconnect', mac_address, timeout_sec)) return bt_test_utils.wait_until(timeout_sec=timeout_sec, condition_func=self.is_a2dp_sink_connected, func_args=[mac_address], expected_value=connected, exception=signals.TestError(msg) if raise_error else None) def wait_for_nap_service_connection(self, connected_mac_addr: str, state_connected: bool, exception: Exception) -> bool: """Waits for NAP service connection to be expected state. Args: connected_mac_addr: String, Bluetooth Mac address is needed to be checked. state_connected: Bool, NAP service connection is established as expected if True, else terminated as expected. exception: Exception, Raised if NAP service connection is not expected state. Raises: exception: Raised if NAP service connection is not expected state. """ def is_device_connected(): """Returns True if connected else False.""" connected_devices = self._ad.sl4a.bluetoothPanGetConnectedDevices() # Check if the Bluetooth mac address is in the connected device list. return connected_mac_addr in [d['address'] for d in connected_devices] bt_test_utils.wait_until(timeout_sec=bt_constants.NAP_CONNECTION_TIMEOUT_SECS, condition_func=is_device_connected, func_args=[], expected_value=state_connected, exception=exception) def verify_internet(self, allow_access: bool, exception: Exception, test_url: str = TEST_URL, interval_sec: int = PING_INTERVAL_TIME_SEC, timeout_sec: float = PING_TIMEOUT_SEC) -> bool: """Verifies that internet is in expected state. Continuously make ping request to a URL for internet verification. Args: allow_access: Bool, Device can have internet access as expected if True, else no internet access as expected. exception: Exception, Raised if internet is not in expected state. test_url: String, A URL is used to verify internet by ping request. interval_sec: Int, Interval time between ping requests in second. timeout_sec: Int, Number of seconds to wait for ping success if allow_access is True else wait for ping failure if allow_access is False. Raises: exception: Raised if internet is not in expected state. """ self._ad.log.info('Verify that internet %s be used.' % ('can' if allow_access else 'can not')) def http_ping(): """Returns True if http ping success else False.""" try: return bool(self._ad.sl4a.httpPing(test_url)) except jsonrpc_client_base.ApiError as e: # ApiError is raised by httpPing() when no internet. self._ad.log.debug(str(e)) return False bt_test_utils.wait_until(timeout_sec=timeout_sec, condition_func=http_ping, func_args=[], expected_value=allow_access, exception=exception, interval_sec=interval_sec) def allow_extra_permissions(self) -> None: """A method to allow extra permissions. This method has no any logics. It is used to skip the operation when it is called if a test is not Wear OS use case. """ def is_service_running(self, mac_address: str, timeout_sec: float) -> bool: """Checks bluetooth profile state. Check bluetooth headset/a2dp profile connection status from bluetooth manager log. Args: mac_address: The Bluetooth mac address of the peripheral device. timeout_sec: Number of seconds to wait for the specified message be found in bluetooth manager log. Returns: True: If pattern match with bluetooth_manager_log. """ pattern_headset = (r'\sm\w+e:\sC\w+d') pattern_a2dp = (r'StateMachine:.*state=Connected') output_headset = self._ad.adb.shell('dumpsys bluetooth_manager | egrep -A20 "Profile: HeadsetService"').decode() output_a2dp = self._ad.adb.shell('dumpsys bluetooth_manager | egrep -A30 "Profile: A2dpService"').decode() service_type = {'a2dp': ((pattern_a2dp), (output_a2dp)), 'headset': ((pattern_headset), (output_headset))} start_time = time.time() end_time = start_time + timeout_sec while start_time < end_time: try: match = service_type if match and mac_address in service_type: return True except adb.AdbError as e: logging.exception(e) time.sleep(ADB_WAITING_TIME_SECONDS) return False def connect_wifi_from_other_device_hotspot(self, wifi_hotspot_device: android_device.AndroidDevice) -> None: """Turns on 2.4G Wifi hotspot from the other android device and connect on the android device. Args: wifi_hotspot_device: Android device, turn on 2.4G Wifi hotspot. """ wifi_hotspot_2_4g_config = bt_constants.WIFI_HOTSPOT_2_4G.copy() if int(wifi_hotspot_device.build_info['build_version_sdk']) > 29: wifi_hotspot_2_4g_config['apBand'] = 1 # Turn on 2.4G Wifi hotspot on the secondary phone. wifi_hotspot_device.sl4a.wifiSetWifiApConfiguration(wifi_hotspot_2_4g_config) wifi_hotspot_device.sl4a.connectivityStartTethering(0, False) # Connect the 2.4G Wifi on the primary phone. self._ad.mbs.wifiEnable() self._ad.mbs.wifiConnectSimple(wifi_hotspot_2_4g_config['SSID'], wifi_hotspot_2_4g_config['password']) def get_paired_device_supported_codecs(self, mac_address: str) -> List[str]: """Gets the supported A2DP codecs of the paired Bluetooth device. Gets the supported A2DP codecs of the paired Bluetooth device from bluetooth manager log. Args: mac_address: The Bluetooth mac address of the paired Bluetooth device. Returns: A list of the A2DP codecs that the paired Bluetooth device supports. """ if not self.is_bt_paired(mac_address): raise signals.TestError(f'Devices {self.serial} and {mac_address} are not paired.') cmd = (f'dumpsys bluetooth_manager | ' f'egrep -A12 "A2dpStateMachine for {mac_address}" | ' f'egrep -A5 "mCodecsSelectableCapabilities"') paired_device_selectable_codecs = self._ad.adb.shell(cmd).decode() pattern = 'codecName:(.*),mCodecType' return re.findall(pattern, paired_device_selectable_codecs) def get_current_a2dp_codec(self) -> bt_constants.BluetoothA2dpCodec: """Gets current A2DP codec type. Returns: A number representing the current A2DP codec type. Codec type values are: 0: SBC 1: AAC 2: aptX 3: aptX HD 4: LDAC """ codec_type = self._ad.sl4a.bluetoothA2dpGetCurrentCodecConfig()['codecType'] return bt_constants.BluetoothA2dpCodec(codec_type) def is_variable_bit_rate_enabled(self) -> bool: """Checks if Variable Bit Rate (VBR) support is enabled for A2DP AAC codec. Returns: True if Variable Bit Rate support is enabled else False. """ return bt_constants.TRUE in self._ad.adb.getprop(bt_constants.AAC_VBR_SUPPORTED_PROPERTY) def toggle_variable_bit_rate(self, enabled: bool = True) -> bool: """Toggles Variable Bit Rate (VBR) support status for A2DP AAC codec. After calling this method, the android device needs to restart Bluetooth for taking effect. If Variable Bit Rate support status is disabled, the android device will use Constant Bit Rate (CBR). Args: enabled: Enable Variable Bit Rate support if True. Returns: True if the status is changed successfully else False. """ self._ad.adb.shell(f'su root setprop {bt_constants.AAC_VBR_SUPPORTED_PROPERTY} ' f'{bt_constants.TRUE if enabled else bt_constants.FALSE}') return enabled == self.is_variable_bit_rate_enabled() def pair_and_connect_ble_device(self, peripheral_ble_device: android_device.AndroidDevice) -> None: """Pairs Android phone with BLE device. Initiates pairing from the phone and checks if it is bonded and connected to the BLE device. Args: peripheral_ble_device: An android device. AndroidDevice instance to pair and connect with. Raises: signals.ControllerError: raised if it failed to connect BLE device. """ peripheral_ble_device.activate_ble_pairing_mode() mac_address = self.scan_and_get_ble_device_address(peripheral_ble_device.get_device_name()) self.pair_and_connect_bluetooth(mac_address) def toggle_charging(self, enabled: bool) -> None: """Toggles charging on the device. Args: enabled: Enable charging if True. """ set_value = '0' if enabled else '1' config_file = bt_constants.CHARGING_CONTROL_CONFIG_DICT[self._ad.build_info['hardware']] self._ad.adb.shell(f'echo {set_value} > {config_file}') def enable_airplane_mode(self, wait_secs=1) -> None: """Enables airplane mode on device. Args: wait_secs: float, the amount of time to wait after sending the airplane mode broadcast. Returns: None """ self._ad.adb.shell(['settings', 'put', 'global', 'airplane_mode_on', '1']) self._ad.adb.shell(['am', 'broadcast', '-a', 'android.intent.action.AIRPLANE_MODE', '--ez', 'state', 'true']) time.sleep(wait_secs) def disable_airplane_mode(self, wait_secs=1) -> None: """Disables airplane mode on device. Args: wait_secs: float, the amount of time to wait after sending the airplane mode broadcast. Returns: None """ self._ad.adb.shell(['settings', 'put', 'global', 'airplane_mode_on', '0']) self._ad.adb.shell(['am', 'broadcast', '-a', 'android.intent.action.AIRPLANE_MODE', '--ez', 'state', 'false']) time.sleep(wait_secs) def disable_verity_check(self) -> None: """Disables Android dm verity check. Returns: None """ if 'verity is already disabled' in str(self._ad.adb.disable_verity()): return self._ad.reboot() self._ad.root_adb() self._ad.wait_for_boot_completion() self._ad.adb.remount()