| #!/usr/bin/env python3 |
| # |
| # Copyright 2019 - The Android Open Source Project |
| # |
| # Licensed under the Apache License, Version 2.0 (the "License"); |
| # you may not use this file except in compliance with the License. |
| # You may obtain a copy of the License at |
| # |
| # http://www.apache.org/licenses/LICENSE-2.0 |
| # |
| # Unless required by applicable law or agreed to in writing, software |
| # distributed under the License is distributed on an "AS IS" BASIS, |
| # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| # See the License for the specific language governing permissions and |
| # limitations under the License. |
| |
| import inspect |
| import logging |
| |
| from queue import Empty |
| |
| from acts.controllers.android_device import AndroidDevice |
| from acts.controllers.fuchsia_device import FuchsiaDevice |
| from acts_contrib.test_utils.bt.bt_constants import ble_scan_settings_modes |
| from acts_contrib.test_utils.bt.bt_constants import gatt_cb_strings |
| from acts_contrib.test_utils.bt.bt_constants import gatt_event |
| from acts_contrib.test_utils.bt.bt_constants import scan_result |
| from acts_contrib.test_utils.bt.bt_gatt_utils import GattTestUtilsError |
| from acts_contrib.test_utils.bt.bt_gatt_utils import disconnect_gatt_connection |
| from acts_contrib.test_utils.bt.bt_gatt_utils import setup_gatt_connection |
| from acts_contrib.test_utils.fuchsia.bt_test_utils import le_scan_for_device_by_name |
| |
| import acts_contrib.test_utils.bt.bt_test_utils as bt_test_utils |
| |
| |
| def create_bluetooth_device(hardware_device): |
| """Creates a generic Bluetooth device based on type of device that is sent |
| to the functions. |
| |
| Args: |
| hardware_device: A Bluetooth hardware device that is supported by ACTS. |
| """ |
| if isinstance(hardware_device, FuchsiaDevice): |
| return FuchsiaBluetoothDevice(hardware_device) |
| elif isinstance(hardware_device, AndroidDevice): |
| return AndroidBluetoothDevice(hardware_device) |
| else: |
| raise ValueError('Unable to create BluetoothDevice for type %s' % |
| type(hardware_device)) |
| |
| |
| class BluetoothDevice(object): |
| """Class representing a generic Bluetooth device. |
| |
| Each object of this class represents a generic Bluetooth device. |
| Android device and Fuchsia devices are the currently supported devices. |
| |
| Attributes: |
| device: A generic Bluetooth device. |
| """ |
| |
| def __init__(self, device): |
| self.device = device |
| self.log = logging |
| |
| def a2dp_initiate_open_stream(self): |
| """Base generic Bluetooth interface. Only called if not overridden by |
| another supported device. |
| """ |
| raise NotImplementedError("{} must be defined.".format( |
| inspect.currentframe().f_code.co_name)) |
| |
| def start_profile_a2dp_sink(self): |
| """Base generic Bluetooth interface. Only called if not overridden by |
| another supported device. |
| """ |
| raise NotImplementedError("{} must be defined.".format( |
| inspect.currentframe().f_code.co_name)) |
| |
| def stop_profile_a2dp_sink(self): |
| """Base generic Bluetooth interface. Only called if not overridden by |
| another supported device. |
| """ |
| raise NotImplementedError("{} must be defined.".format( |
| inspect.currentframe().f_code.co_name)) |
| |
| def start_pairing_helper(self): |
| """Base generic Bluetooth interface. Only called if not overridden by |
| another supported device. |
| """ |
| raise NotImplementedError("{} must be defined.".format( |
| inspect.currentframe().f_code.co_name)) |
| |
| def set_discoverable(self, is_discoverable): |
| """Base generic Bluetooth interface. Only called if not overridden by |
| another supported device. |
| """ |
| raise NotImplementedError("{} must be defined.".format( |
| inspect.currentframe().f_code.co_name)) |
| |
| def bluetooth_toggle_state(self, state): |
| """Base generic Bluetooth interface. Only called if not overridden by |
| another supported device. |
| """ |
| raise NotImplementedError("{} must be defined.".format( |
| inspect.currentframe().f_code.co_name)) |
| |
| def gatt_client_discover_characteristic_by_uuid(self, peer_identifier, |
| uuid): |
| """Base generic Bluetooth interface. Only called if not overridden by |
| another supported device. |
| """ |
| raise NotImplementedError("{} must be defined.".format( |
| inspect.currentframe().f_code.co_name)) |
| |
| def initialize_bluetooth_controller(self): |
| """Base generic Bluetooth interface. Only called if not overridden by |
| another supported device. |
| """ |
| raise NotImplementedError("{} must be defined.".format( |
| inspect.currentframe().f_code.co_name)) |
| |
| def get_pairing_pin(self): |
| """Base generic Bluetooth interface. Only called if not overridden by |
| another supported device. |
| """ |
| raise NotImplementedError("{} must be defined.".format( |
| inspect.currentframe().f_code.co_name)) |
| |
| def input_pairing_pin(self, pin): |
| """Base generic Bluetooth interface. Only called if not overridden by |
| another supported device. |
| """ |
| raise NotImplementedError("{} must be defined.".format( |
| inspect.currentframe().f_code.co_name)) |
| |
| def get_bluetooth_local_address(self): |
| """Base generic Bluetooth interface. Only called if not overridden by |
| another supported device. |
| """ |
| raise NotImplementedError("{} must be defined.".format( |
| inspect.currentframe().f_code.co_name)) |
| |
| def gatt_connect(self, peer_identifier, transport, autoconnect): |
| """Base generic Bluetooth interface. Only called if not overridden by |
| another supported device. |
| """ |
| raise NotImplementedError("{} must be defined.".format( |
| inspect.currentframe().f_code.co_name)) |
| |
| def gatt_client_write_characteristic_without_response_by_handle( |
| self, peer_identifier, handle, value): |
| """Base generic Bluetooth interface. Only called if not overridden by |
| another supported device. |
| """ |
| raise NotImplementedError("{} must be defined.".format( |
| inspect.currentframe().f_code.co_name)) |
| |
| def gatt_client_write_characteristic_by_handle(self, peer_identifier, |
| handle, offset, value): |
| """Base generic Bluetooth interface. Only called if not overridden by |
| another supported device. |
| """ |
| raise NotImplementedError("{} must be defined.".format( |
| inspect.currentframe().f_code.co_name)) |
| |
| def gatt_client_read_characteristic_by_handle(self, peer_identifier, |
| handle): |
| """Base generic Bluetooth interface. Only called if not overridden by |
| another supported device. |
| """ |
| raise NotImplementedError("{} must be defined.".format( |
| inspect.currentframe().f_code.co_name)) |
| |
| def gatt_client_read_characteristic_by_uuid(self, peer_identifier, uuid): |
| """Base generic Bluetooth interface. Only called if not overridden by |
| another supported device. |
| """ |
| raise NotImplementedError("{} must be defined.".format( |
| inspect.currentframe().f_code.co_name)) |
| |
| def gatt_client_read_long_characteristic_by_handle(self, peer_identifier, |
| handle, offset, |
| max_bytes): |
| """Base generic Bluetooth interface. Only called if not overridden by |
| another supported device. |
| """ |
| raise NotImplementedError("{} must be defined.".format( |
| inspect.currentframe().f_code.co_name)) |
| |
| def gatt_client_enable_notifiy_characteristic_by_handle( |
| self, peer_identifier, handle): |
| """Base generic Bluetooth interface. Only called if not overridden by |
| another supported device. |
| """ |
| raise NotImplementedError("{} must be defined.".format( |
| inspect.currentframe().f_code.co_name)) |
| |
| def gatt_client_disable_notifiy_characteristic_by_handle( |
| self, peer_identifier, handle): |
| """Base generic Bluetooth interface. Only called if not overridden by |
| another supported device. |
| """ |
| raise NotImplementedError("{} must be defined.".format( |
| inspect.currentframe().f_code.co_name)) |
| |
| def gatt_client_read_descriptor_by_handle(self, peer_identifier, handle): |
| """Base generic Bluetooth interface. Only called if not overridden by |
| another supported device. |
| """ |
| raise NotImplementedError("{} must be defined.".format( |
| inspect.currentframe().f_code.co_name)) |
| |
| def gatt_client_write_descriptor_by_handle(self, peer_identifier, handle, |
| offset, value): |
| """Base generic Bluetooth interface. Only called if not overridden by |
| another supported device. |
| """ |
| raise NotImplementedError("{} must be defined.".format( |
| inspect.currentframe().f_code.co_name)) |
| |
| def gatt_client_long_read_descriptor_by_handle(self, peer_identifier, |
| handle, offset, max_bytes): |
| """Base generic Bluetooth interface. Only called if not overridden by |
| another supported device. |
| """ |
| raise NotImplementedError("{} must be defined.".format( |
| inspect.currentframe().f_code.co_name)) |
| |
| def gatt_disconnect(self, peer_identifier): |
| """Base generic Bluetooth interface. Only called if not overridden by |
| another supported device. |
| """ |
| raise NotImplementedError("{} must be defined.".format( |
| inspect.currentframe().f_code.co_name)) |
| |
| def gatt_client_refresh(self, peer_identifier): |
| """Base generic Bluetooth interface. Only called if not overridden by |
| another supported device. |
| """ |
| raise NotImplementedError("{} must be defined.".format( |
| inspect.currentframe().f_code.co_name)) |
| |
| def le_scan_with_name_filter(self, name, timeout): |
| """Base generic Bluetooth interface. Only called if not overridden by |
| another supported device. |
| """ |
| raise NotImplementedError("{} must be defined.".format( |
| inspect.currentframe().f_code.co_name)) |
| |
| def log_info(self, log): |
| """Base generic Bluetooth interface. Only called if not overridden by |
| another supported device. |
| """ |
| raise NotImplementedError("{} must be defined.".format( |
| inspect.currentframe().f_code.co_name)) |
| |
| def reset_bluetooth(self): |
| """Base generic Bluetooth interface. Only called if not overridden by |
| another supported device. |
| """ |
| raise NotImplementedError("{} must be defined.".format( |
| inspect.currentframe().f_code.co_name)) |
| |
| def sdp_add_search(self, attribute_list, profile_id): |
| """Base generic Bluetooth interface. Only called if not overridden by |
| another supported device. |
| """ |
| raise NotImplementedError("{} must be defined.".format( |
| inspect.currentframe().f_code.co_name)) |
| |
| def sdp_add_service(self, sdp_record): |
| """Base generic Bluetooth interface. Only called if not overridden by |
| another supported device. |
| """ |
| raise NotImplementedError("{} must be defined.".format( |
| inspect.currentframe().f_code.co_name)) |
| |
| def sdp_clean_up(self): |
| """Base generic Bluetooth interface. Only called if not overridden by |
| another supported device. |
| """ |
| raise NotImplementedError("{} must be defined.".format( |
| inspect.currentframe().f_code.co_name)) |
| |
| def sdp_init(self): |
| """Base generic Bluetooth interface. Only called if not overridden by |
| another supported device. |
| """ |
| raise NotImplementedError("{} must be defined.".format( |
| inspect.currentframe().f_code.co_name)) |
| |
| def sdp_remove_service(self, service_id): |
| """Base generic Bluetooth interface. Only called if not overridden by |
| another supported device. |
| """ |
| raise NotImplementedError("{} must be defined.".format( |
| inspect.currentframe().f_code.co_name)) |
| |
| def start_le_advertisement(self, adv_data, scan_response, adv_interval, |
| connectable): |
| """Base generic Bluetooth interface. Only called if not overridden by |
| another supported device. |
| """ |
| raise NotImplementedError("{} must be defined.".format( |
| inspect.currentframe().f_code.co_name)) |
| |
| def stop_le_advertisement(self): |
| """Base generic Bluetooth interface. Only called if not overridden by |
| another supported device. |
| """ |
| raise NotImplementedError("{} must be defined.".format( |
| inspect.currentframe().f_code.co_name)) |
| |
| def set_bluetooth_local_name(self, name): |
| """Base generic Bluetooth interface. Only called if not overridden by |
| another supported device. |
| """ |
| raise NotImplementedError("{} must be defined.".format( |
| inspect.currentframe().f_code.co_name)) |
| |
| def setup_gatt_server(self, database): |
| """Base generic Bluetooth interface. Only called if not overridden by |
| another supported device. |
| """ |
| raise NotImplementedError("{} must be defined.".format( |
| inspect.currentframe().f_code.co_name)) |
| |
| def close_gatt_server(self): |
| """Base generic Bluetooth interface. Only called if not overridden by |
| another supported device. |
| """ |
| raise NotImplementedError("{} must be defined.".format( |
| inspect.currentframe().f_code.co_name)) |
| |
| def unbond_device(self, peer_identifier): |
| """Base generic Bluetooth interface. Only called if not overridden by |
| another supported device. |
| """ |
| raise NotImplementedError("{} must be defined.".format( |
| inspect.currentframe().f_code.co_name)) |
| |
| def unbond_all_known_devices(self): |
| """Base generic Bluetooth interface. Only called if not overridden by |
| another supported device. |
| """ |
| raise NotImplementedError("{} must be defined.".format( |
| inspect.currentframe().f_code.co_name)) |
| |
| def init_pair(self, peer_identifier, security_level, non_bondable, |
| transport): |
| """Base generic Bluetooth interface. Only called if not overridden by |
| another supported device. |
| """ |
| raise NotImplementedError("{} must be defined.".format( |
| inspect.currentframe().f_code.co_name)) |
| |
| |
| class AndroidBluetoothDevice(BluetoothDevice): |
| """Class wrapper for an Android Bluetooth device. |
| |
| Each object of this class represents a generic Bluetooth device. |
| Android device and Fuchsia devices are the currently supported devices/ |
| |
| Attributes: |
| android_device: An Android Bluetooth device. |
| """ |
| |
| def __init__(self, android_device): |
| super().__init__(android_device) |
| self.gatt_timeout = 10 |
| self.peer_mapping = {} |
| self.discovered_services_index = None |
| |
| def _client_wait(self, gatt_event, gatt_callback): |
| return self._timed_pop(gatt_event, gatt_callback) |
| |
| def _timed_pop(self, gatt_event, gatt_callback): |
| expected_event = gatt_event["evt"].format(gatt_callback) |
| try: |
| return self.device.ed.pop_event(expected_event, self.gatt_timeout) |
| except Empty as emp: |
| raise AssertionError(gatt_event["err"].format(expected_event)) |
| |
| def _setup_discovered_services_index(self, bluetooth_gatt): |
| """ Sets the discovered services index for the gatt connection |
| related to the Bluetooth GATT callback object. |
| |
| Args: |
| bluetooth_gatt: The BluetoothGatt callback id |
| """ |
| if not self.discovered_services_index: |
| self.device.droid.gattClientDiscoverServices(bluetooth_gatt) |
| expected_event = gatt_cb_strings['gatt_serv_disc'].format( |
| self.gatt_callback) |
| event = self.dut.ed.pop_event(expected_event, self.gatt_timeout) |
| self.discovered_services_index = event['data']['ServicesIndex'] |
| |
| def a2dp_initiate_open_stream(self): |
| raise NotImplementedError("{} not yet implemented.".format( |
| inspect.currentframe().f_code.co_name)) |
| |
| def start_profile_a2dp_sink(self): |
| raise NotImplementedError("{} not yet implemented.".format( |
| inspect.currentframe().f_code.co_name)) |
| |
| def stop_profile_a2dp_sink(self): |
| raise NotImplementedError("{} not yet implemented.".format( |
| inspect.currentframe().f_code.co_name)) |
| |
| def bluetooth_toggle_state(self, state): |
| self.device.droid.bluetoothToggleState(state) |
| |
| def set_discoverable(self, is_discoverable): |
| """ Sets the device's discoverability. |
| |
| Args: |
| is_discoverable: True if discoverable, false if not discoverable |
| """ |
| if is_discoverable: |
| self.device.droid.bluetoothMakeDiscoverable() |
| else: |
| self.device.droid.bluetoothMakeUndiscoverable() |
| |
| def initialize_bluetooth_controller(self): |
| """ Just pass for Android as there is no concept of initializing |
| a Bluetooth controller. |
| """ |
| |
| def start_pairing_helper(self): |
| """ Starts the Android pairing helper. |
| """ |
| self.device.droid.bluetoothStartPairingHelper(True) |
| |
| def gatt_client_write_characteristic_without_response_by_handle( |
| self, peer_identifier, handle, value): |
| """ Perform a GATT Client write Characteristic without response to |
| remote peer GATT server database. |
| |
| Args: |
| peer_identifier: The mac address associated with the GATT connection |
| handle: The characteristic handle (or instance id). |
| value: The list of bytes to write. |
| Returns: |
| True if success, False if failure. |
| """ |
| peer_info = self.peer_mapping.get(peer_identifier) |
| if not peer_info: |
| self.log.error( |
| "Peer idenifier {} not currently connected or unknown.".format( |
| peer_identifier)) |
| return False |
| self._setup_discovered_services_index() |
| self.device.droid.gattClientWriteCharacteristicByInstanceId( |
| peer_info.get('bluetooth_gatt'), self.discovered_services_index, |
| handle, value) |
| try: |
| event = self._client_wait(gatt_event['char_write'], |
| peer_info.get('gatt_callback')) |
| except AssertionError as err: |
| self.log.error("Failed to write Characteristic: {}".format(err)) |
| return True |
| |
| def gatt_client_write_characteristic_by_handle(self, peer_identifier, |
| handle, offset, value): |
| """ Perform a GATT Client write Characteristic without response to |
| remote peer GATT server database. |
| |
| Args: |
| peer_identifier: The mac address associated with the GATT connection |
| handle: The characteristic handle (or instance id). |
| offset: Not used yet. |
| value: The list of bytes to write. |
| Returns: |
| True if success, False if failure. |
| """ |
| peer_info = self.peer_mapping.get(peer_identifier) |
| if not peer_info: |
| self.log.error( |
| "Peer idenifier {} not currently connected or unknown.".format( |
| peer_identifier)) |
| return False |
| self._setup_discovered_services_index() |
| self.device.droid.gattClientWriteCharacteristicByInstanceId( |
| peer_info.get('bluetooth_gatt'), self.discovered_services_index, |
| handle, value) |
| try: |
| event = self._client_wait(gatt_event['char_write'], |
| peer_info.get('gatt_callback')) |
| except AssertionError as err: |
| self.log.error("Failed to write Characteristic: {}".format(err)) |
| return True |
| |
| def gatt_client_read_characteristic_by_handle(self, peer_identifier, |
| handle): |
| """ Perform a GATT Client read Characteristic to remote peer GATT |
| server database. |
| |
| Args: |
| peer_identifier: The mac address associated with the GATT connection |
| handle: The characteristic handle (or instance id). |
| Returns: |
| Value of Characteristic if success, None if failure. |
| """ |
| peer_info = self.peer_mapping.get(peer_identifier) |
| if not peer_info: |
| self.log.error( |
| "Peer idenifier {} not currently connected or unknown.".format( |
| peer_identifier)) |
| return False |
| self._setup_discovered_services_index() |
| self.dut.droid.gattClientReadCharacteristicByInstanceId( |
| peer_info.get('bluetooth_gatt'), self.discovered_services_index, |
| handle) |
| try: |
| event = self._client_wait(gatt_event['char_read'], |
| peer_info.get('gatt_callback')) |
| except AssertionError as err: |
| self.log.error("Failed to read Characteristic: {}".format(err)) |
| |
| return event['data']['CharacteristicValue'] |
| |
| def gatt_client_read_long_characteristic_by_handle(self, peer_identifier, |
| handle, offset, |
| max_bytes): |
| """ Perform a GATT Client read Characteristic to remote peer GATT |
| server database. |
| |
| Args: |
| peer_identifier: The mac address associated with the GATT connection |
| offset: Not used yet. |
| handle: The characteristic handle (or instance id). |
| max_bytes: Not used yet. |
| Returns: |
| Value of Characteristic if success, None if failure. |
| """ |
| peer_info = self.peer_mapping.get(peer_identifier) |
| if not peer_info: |
| self.log.error( |
| "Peer idenifier {} not currently connected or unknown.".format( |
| peer_identifier)) |
| return False |
| self._setup_discovered_services_index() |
| self.dut.droid.gattClientReadCharacteristicByInstanceId( |
| peer_info.get('bluetooth_gatt'), self.discovered_services_index, |
| handle) |
| try: |
| event = self._client_wait(gatt_event['char_read'], |
| peer_info.get('gatt_callback')) |
| except AssertionError as err: |
| self.log.error("Failed to read Characteristic: {}".format(err)) |
| |
| return event['data']['CharacteristicValue'] |
| |
| def gatt_client_enable_notifiy_characteristic_by_handle( |
| self, peer_identifier, handle): |
| """ Perform a GATT Client enable Characteristic notification to remote |
| peer GATT server database. |
| |
| Args: |
| peer_identifier: The mac address associated with the GATT connection |
| handle: The characteristic handle. |
| Returns: |
| True is success, False if failure. |
| """ |
| raise NotImplementedError("{} not yet implemented.".format( |
| inspect.currentframe().f_code.co_name)) |
| |
| def gatt_client_disable_notifiy_characteristic_by_handle( |
| self, peer_identifier, handle): |
| """ Perform a GATT Client disable Characteristic notification to remote |
| peer GATT server database. |
| |
| Args: |
| peer_identifier: The mac address associated with the GATT connection |
| handle: The characteristic handle. |
| Returns: |
| True is success, False if failure. |
| """ |
| raise NotImplementedError("{} not yet implemented.".format( |
| inspect.currentframe().f_code.co_name)) |
| |
| def gatt_client_read_descriptor_by_handle(self, peer_identifier, handle): |
| """ Perform a GATT Client read Descriptor to remote peer GATT |
| server database. |
| |
| Args: |
| peer_identifier: The mac address associated with the GATT connection |
| handle: The Descriptor handle (or instance id). |
| Returns: |
| Value of Descriptor if success, None if failure. |
| """ |
| peer_info = self.peer_mapping.get(peer_identifier) |
| if not peer_info: |
| self.log.error( |
| "Peer idenifier {} not currently connected or unknown.".format( |
| peer_identifier)) |
| return False |
| self._setup_discovered_services_index() |
| self.dut.droid.gattClientReadDescriptorByInstanceId( |
| peer_info.get('bluetooth_gatt'), self.discovered_services_index, |
| handle) |
| try: |
| event = self._client_wait(gatt_event['desc_read'], |
| peer_info.get('gatt_callback')) |
| except AssertionError as err: |
| self.log.error("Failed to read Descriptor: {}".format(err)) |
| # TODO: Implement sending Descriptor value in SL4A such that the data |
| # can be represented by: event['data']['DescriptorValue'] |
| return "" |
| |
| def gatt_client_write_descriptor_by_handle(self, peer_identifier, handle, |
| offset, value): |
| """ Perform a GATT Client write Descriptor to the remote peer GATT |
| server database. |
| |
| Args: |
| peer_identifier: The mac address associated with the GATT connection |
| handle: The Descriptor handle (or instance id). |
| offset: Not used yet |
| value: The list of bytes to write. |
| Returns: |
| True if success, False if failure. |
| """ |
| peer_info = self.peer_mapping.get(peer_identifier) |
| if not peer_info: |
| self.log.error( |
| "Peer idenifier {} not currently connected or unknown.".format( |
| peer_identifier)) |
| return False |
| self._setup_discovered_services_index() |
| self.device.droid.gattClientWriteDescriptorByInstanceId( |
| peer_info.get('bluetooth_gatt'), self.discovered_services_index, |
| handle, value) |
| try: |
| event = self._client_wait(gatt_event['desc_write'], |
| peer_info.get('gatt_callback')) |
| except AssertionError as err: |
| self.log.error("Failed to write Characteristic: {}".format(err)) |
| return True |
| |
| def gatt_connect(self, peer_identifier, transport, autoconnect=False): |
| """ Perform a GATT connection to a perihperal. |
| |
| Args: |
| peer_identifier: The mac address to connect to. |
| transport: Which transport to use. |
| autoconnect: Set autocnnect to True or False. |
| Returns: |
| True if success, False if failure. |
| """ |
| try: |
| bluetooth_gatt, gatt_callback = setup_gatt_connection( |
| self.device, peer_identifier, autoconnect, transport) |
| self.peer_mapping[peer_identifier] = { |
| "bluetooth_gatt": bluetooth_gatt, |
| "gatt_callback": gatt_callback |
| } |
| except GattTestUtilsError as err: |
| self.log.error(err) |
| return False |
| return True |
| |
| def gatt_disconnect(self, peer_identifier): |
| """ Perform a GATT disconnect from a perihperal. |
| |
| Args: |
| peer_identifier: The peer to disconnect from. |
| Returns: |
| True if success, False if failure. |
| """ |
| peer_info = self.peer_mapping.get(peer_identifier) |
| if not peer_info: |
| self.log.error( |
| "No previous connections made to {}".format(peer_identifier)) |
| return False |
| |
| try: |
| disconnect_gatt_connection(self.device, |
| peer_info.get("bluetooth_gatt"), |
| peer_info.get("gatt_callback")) |
| self.device.droid.gattClientClose(peer_info.get("bluetooth_gatt")) |
| except GattTestUtilsError as err: |
| self.log.error(err) |
| return False |
| self.device.droid.gattClientClose(peer_info.get("bluetooth_gatt")) |
| |
| def gatt_client_refresh(self, peer_identifier): |
| """ Perform a GATT Client Refresh of a perihperal. |
| |
| Clears the internal cache and forces a refresh of the services from the |
| remote device. |
| |
| Args: |
| peer_identifier: The peer to refresh. |
| """ |
| peer_info = self.peer_mapping.get(peer_identifier) |
| if not peer_info: |
| self.log.error( |
| "No previous connections made to {}".format(peer_identifier)) |
| return False |
| self.device.droid.gattClientRefresh(peer_info["bluetooth_gatt"]) |
| |
| def le_scan_with_name_filter(self, name, timeout): |
| """ Scan over LE for a specific device name. |
| |
| Args: |
| name: The name filter to set. |
| timeout: The timeout to wait to find the advertisement. |
| Returns: |
| Discovered mac address or None |
| """ |
| self.device.droid.bleSetScanSettingsScanMode( |
| ble_scan_settings_modes['low_latency']) |
| filter_list = self.device.droid.bleGenFilterList() |
| scan_settings = self.device.droid.bleBuildScanSetting() |
| scan_callback = self.device.droid.bleGenScanCallback() |
| self.device.droid.bleSetScanFilterDeviceName(name) |
| self.device.droid.bleBuildScanFilter(filter_list) |
| self.device.droid.bleSetScanFilterDeviceName(self.name) |
| self.device.droid.bleStartBleScan(filter_list, scan_settings, |
| scan_callback) |
| try: |
| event = self.device.ed.pop_event(scan_result.format(scan_callback), |
| timeout) |
| return event['data']['Result']['deviceInfo']['address'] |
| except Empty as err: |
| self.log.info("Scanner did not find advertisement {}".format(err)) |
| return None |
| |
| def log_info(self, log): |
| """ Log directly onto the device. |
| |
| Args: |
| log: The informative log. |
| """ |
| self.device.droid.log.logI(log) |
| |
| def set_bluetooth_local_name(self, name): |
| """ Sets the Bluetooth controller's local name |
| Args: |
| name: The name to set. |
| """ |
| self.device.droid.bluetoothSetLocalName(name) |
| |
| def get_local_bluetooth_address(self): |
| """ Returns the Bluetooth local address. |
| """ |
| return self.device.droid.bluetoothGetLocalAddress() |
| |
| def reset_bluetooth(self): |
| """ Resets Bluetooth on the Android Device. |
| """ |
| bt_test_utils.reset_bluetooth([self.device]) |
| |
| def sdp_add_search(self, attribute_list, profile_id): |
| """Adds an SDP search record. |
| Args: |
| attribute_list: The list of attributes to set |
| profile_id: The profile ID to set. |
| """ |
| # Android devices currently have no hooks to modify the SDP record. |
| |
| def sdp_add_service(self, sdp_record): |
| """Adds an SDP service record. |
| Args: |
| sdp_record: The dictionary representing the search record to add. |
| Returns: |
| service_id: The service id to track the service record published. |
| None if failed. |
| """ |
| # Android devices currently have no hooks to modify the SDP record. |
| |
| def sdp_clean_up(self): |
| """Cleans up all objects related to SDP. |
| """ |
| self.device.sl4f.sdp_lib.cleanUp() |
| |
| def sdp_init(self): |
| """Initializes SDP on the device. |
| """ |
| # Android devices currently have no hooks to modify the SDP record. |
| |
| def sdp_remove_service(self, service_id): |
| """Removes a service based on an input id. |
| Args: |
| service_id: The service ID to remove. |
| """ |
| # Android devices currently have no hooks to modify the SDP record. |
| |
| def unbond_all_known_devices(self): |
| """ Unbond all known remote devices. |
| """ |
| self.device.droid.bluetoothFactoryReset() |
| |
| def unbond_device(self, peer_identifier): |
| """ Unbond peer identifier. |
| |
| Args: |
| peer_identifier: The mac address for the peer to unbond. |
| |
| """ |
| self.device.droid.bluetoothUnbond(peer_identifier) |
| |
| def init_pair(self, peer_identifier, security_level, non_bondable, |
| transport): |
| """ Send an outgoing pairing request the input peer_identifier. |
| |
| Android currently does not support setting various security levels or |
| bondable modes. Making them available for other bluetooth_device |
| variants. Depending on the Address type, Android will figure out the |
| transport to pair automatically. |
| |
| Args: |
| peer_identifier: A string representing the device id. |
| security_level: Not yet implemented. See Fuchsia device impl. |
| non_bondable: Not yet implemented. See Fuchsia device impl. |
| transport: Not yet implemented. See Fuchsia device impl. |
| |
| """ |
| self.dut.droid.bluetoothBond(self.peer_identifier) |
| |
| |
| class FuchsiaBluetoothDevice(BluetoothDevice): |
| """Class wrapper for an Fuchsia Bluetooth device. |
| |
| Each object of this class represents a generic luetooth device. |
| Android device and Fuchsia devices are the currently supported devices/ |
| |
| Attributes: |
| fuchsia_device: A Fuchsia Bluetooth device. |
| """ |
| |
| def __init__(self, fuchsia_device): |
| super().__init__(fuchsia_device) |
| |
| def a2dp_initiate_open_stream(self): |
| raise NotImplementedError("{} not yet implemented.".format( |
| inspect.currentframe().f_code.co_name)) |
| |
| def start_profile_a2dp_sink(self): |
| """ Starts the A2DP sink profile. |
| """ |
| self.device.start_v1_component("bt-a2dp-sink") |
| |
| def stop_profile_a2dp_sink(self): |
| """ Stops the A2DP sink profile. |
| """ |
| self.device.stop_v1_component("bt-a2dp-sink") |
| |
| def start_pairing_helper(self): |
| self.device.sl4f.bts_lib.acceptPairing() |
| |
| def bluetooth_toggle_state(self, state): |
| """Stub for Fuchsia implementation.""" |
| |
| def set_discoverable(self, is_discoverable): |
| """ Sets the device's discoverability. |
| |
| Args: |
| is_discoverable: True if discoverable, false if not discoverable |
| """ |
| self.device.sl4f.bts_lib.setDiscoverable(is_discoverable) |
| |
| def get_pairing_pin(self): |
| """ Get the pairing pin from the active pairing delegate. |
| """ |
| return self.device.sl4f.bts_lib.getPairingPin()['result'] |
| |
| def input_pairing_pin(self, pin): |
| """ Input pairing pin to active pairing delegate. |
| |
| Args: |
| pin: The pin to input. |
| """ |
| self.device.sl4f.bts_lib.inputPairingPin(pin) |
| |
| def initialize_bluetooth_controller(self): |
| """ Initialize Bluetooth controller for first time use. |
| """ |
| self.device.sl4f.bts_lib.initBluetoothSys() |
| |
| def get_local_bluetooth_address(self): |
| """ Returns the Bluetooth local address. |
| """ |
| return self.device.sl4f.bts_lib.getActiveAdapterAddress().get("result") |
| |
| def set_bluetooth_local_name(self, name): |
| """ Sets the Bluetooth controller's local name |
| Args: |
| name: The name to set. |
| """ |
| self.device.sl4f.bts_lib.setName(name) |
| |
| def gatt_client_write_characteristic_without_response_by_handle( |
| self, peer_identifier, handle, value): |
| """ Perform a GATT Client write Characteristic without response to |
| remote peer GATT server database. |
| |
| Args: |
| peer_identifier: The peer to connect to. |
| handle: The characteristic handle. |
| value: The list of bytes to write. |
| Returns: |
| True if success, False if failure. |
| """ |
| if (not self._find_service_id_and_connect_to_service_for_handle( |
| peer_identifier, handle)): |
| self.log.warn( |
| "Unable to find handle {} in GATT server db.".format(handle)) |
| result = self.device.sl4f.gattc_lib.writeCharByIdWithoutResponse( |
| handle, value) |
| if result.get("error") is not None: |
| self.log.error( |
| "Failed to write characteristic handle {} with err: {}".format( |
| handle, result.get("error"))) |
| return False |
| return True |
| |
| def gatt_client_write_characteristic_by_handle(self, peer_identifier, |
| handle, offset, value): |
| """ Perform a GATT Client write Characteristic to remote peer GATT |
| server database. |
| |
| Args: |
| peer_identifier: The peer to connect to. |
| handle: The characteristic handle. |
| offset: The offset to start writing to. |
| value: The list of bytes to write. |
| Returns: |
| True if success, False if failure. |
| """ |
| if (not self._find_service_id_and_connect_to_service_for_handle( |
| peer_identifier, handle)): |
| self.log.warn( |
| "Unable to find handle {} in GATT server db.".format(handle)) |
| result = self.device.sl4f.gattc_lib.writeCharById( |
| handle, offset, value) |
| if result.get("error") is not None: |
| self.log.error( |
| "Failed to write characteristic handle {} with err: {}".format( |
| handle, result.get("error"))) |
| return False |
| return True |
| |
| def gatt_client_write_long_characteristic_by_handle( |
| self, peer_identifier, handle, offset, value, reliable_mode=False): |
| """ Perform a GATT Client write long Characteristic to remote peer GATT |
| server database. |
| |
| Args: |
| peer_identifier: The peer to connect to. |
| handle: The characteristic handle. |
| offset: The offset to start writing to. |
| value: The list of bytes to write. |
| reliable_mode: A bool value representing a reliable write or not. |
| Returns: |
| True if success, False if failure. |
| """ |
| if (not self._find_service_id_and_connect_to_service_for_handle( |
| peer_identifier, handle)): |
| self.log.error( |
| "Unable to find handle {} in GATT server db.".format(handle)) |
| return False |
| result = self.device.sl4f.gattc_lib.writeLongCharById( |
| handle, offset, value, reliable_mode) |
| if result.get("error") is not None: |
| self.log.error( |
| "Failed to write long characteristic handle {} with err: {}". |
| format(peer_identifier, result.get("error"))) |
| return False |
| return True |
| |
| def gatt_client_write_long_descriptor_by_handle(self, peer_identifier, |
| handle, offset, value): |
| """ Perform a GATT Client write long Descriptor to remote peer GATT |
| server database. |
| |
| Args: |
| peer_identifier: The peer to connect to. |
| handle: The descriptor handle. |
| offset: The offset to start writing to. |
| value: The list of bytes to write. |
| Returns: |
| True if success, False if failure. |
| """ |
| if (not self._find_service_id_and_connect_to_service_for_handle( |
| peer_identifier, handle)): |
| self.log.error( |
| "Unable to find handle {} in GATT server db.".format(handle)) |
| return False |
| result = self.device.sl4f.gattc_lib.writeLongDescById( |
| handle, offset, value) |
| if result.get("error") is not None: |
| self.log.error( |
| "Failed to write long descriptor handle {} with err: {}". |
| format(peer_identifier, result.get("error"))) |
| return False |
| return True |
| |
| def gatt_client_read_characteristic_by_handle(self, peer_identifier, |
| handle): |
| """ Perform a GATT Client read Characteristic to remote peer GATT |
| server database. |
| |
| Args: |
| peer_identifier: The peer to connect to. |
| handle: The characteristic handle. |
| Returns: |
| Value of Characteristic if success, None if failure. |
| """ |
| if (not self._find_service_id_and_connect_to_service_for_handle( |
| peer_identifier, handle)): |
| self.log.warn( |
| "Unable to find handle {} in GATT server db.".format(handle)) |
| result = self.device.sl4f.gattc_lib.readCharacteristicById(handle) |
| if result.get("error") is not None: |
| self.log.error( |
| "Failed to read characteristic handle {} with err: {}".format( |
| handle, result.get("error"))) |
| return None |
| return result.get("result") |
| |
| def gatt_client_read_characteristic_by_uuid(self, peer_identifier, uuid): |
| """ Perform a GATT Client read Characteristic by uuid to remote peer GATT |
| server database. |
| |
| Args: |
| peer_identifier: The peer to connect to. |
| uuid: The characteristic uuid. |
| Returns: |
| Value of Characteristic if success, None if failure. |
| """ |
| if (not self._find_service_id_and_connect_to_service_for_handle( |
| peer_identifier, uuid, uuid=True)): |
| self.log.warn( |
| "Unable to find uuid {} in GATT server db.".format(uuid)) |
| result = self.device.sl4f.gattc_lib.readCharacteristicByType(uuid) |
| if result.get("error") is not None: |
| self.log.error( |
| "Failed to read characteristic uuid {} with err: {}".format( |
| uuid, result.get("error"))) |
| return None |
| return result.get("result") |
| |
| def gatt_client_read_long_characteristic_by_handle(self, peer_identifier, |
| handle, offset, |
| max_bytes): |
| """ Perform a GATT Client read Characteristic to remote peer GATT |
| server database. |
| |
| Args: |
| peer_identifier: The peer to connect to. |
| handle: The characteristic handle. |
| offset: The offset to start reading. |
| max_bytes: The max bytes to return for each read. |
| Returns: |
| Value of Characteristic if success, None if failure. |
| """ |
| if (not self._find_service_id_and_connect_to_service_for_handle( |
| peer_identifier, handle)): |
| self.log.warn( |
| "Unable to find handle {} in GATT server db.".format(handle)) |
| result = self.device.sl4f.gattc_lib.readLongCharacteristicById( |
| handle, offset, max_bytes) |
| if result.get("error") is not None: |
| self.log.error( |
| "Failed to read characteristic handle {} with err: {}".format( |
| handle, result.get("error"))) |
| return None |
| return result.get("result") |
| |
| def gatt_client_enable_notifiy_characteristic_by_handle( |
| self, peer_identifier, handle): |
| """ Perform a GATT Client enable Characteristic notification to remote |
| peer GATT server database. |
| |
| Args: |
| peer_identifier: The peer to connect to. |
| handle: The characteristic handle. |
| Returns: |
| True is success, False if failure. |
| """ |
| if (not self._find_service_id_and_connect_to_service_for_handle( |
| peer_identifier, handle)): |
| self.log.warn( |
| "Unable to find handle {} in GATT server db.".format(handle)) |
| result = self.device.sl4f.gattc_lib.enableNotifyCharacteristic(handle) |
| if result.get("error") is not None: |
| self.log.error( |
| "Failed to enable characteristic notifications for handle {} " |
| "with err: {}".format(handle, result.get("error"))) |
| return None |
| return result.get("result") |
| |
| def gatt_client_disable_notifiy_characteristic_by_handle( |
| self, peer_identifier, handle): |
| """ Perform a GATT Client disable Characteristic notification to remote |
| peer GATT server database. |
| |
| Args: |
| peer_identifier: The peer to connect to. |
| handle: The characteristic handle. |
| Returns: |
| True is success, False if failure. |
| """ |
| if (not self._find_service_id_and_connect_to_service_for_handle( |
| peer_identifier, handle)): |
| self.log.warn( |
| "Unable to find handle {} in GATT server db.".format(handle)) |
| result = self.device.sl4f.gattc_lib.disableNotifyCharacteristic(handle) |
| if result.get("error") is not None: |
| self.log.error( |
| "Failed to disable characteristic notifications for handle {} " |
| "with err: {}".format(peer_identifier, result.get("error"))) |
| return None |
| return result.get("result") |
| |
| def gatt_client_read_descriptor_by_handle(self, peer_identifier, handle): |
| """ Perform a GATT Client read Descriptor to remote peer GATT server |
| database. |
| |
| Args: |
| peer_identifier: The peer to connect to. |
| handle: The Descriptor handle. |
| Returns: |
| Value of Descriptor if success, None if failure. |
| """ |
| if (not self._find_service_id_and_connect_to_service_for_handle( |
| peer_identifier, handle)): |
| self.log.warn( |
| "Unable to find handle {} in GATT server db.".format(handle)) |
| result = self.device.sl4f.gattc_lib.readDescriptorById(handle) |
| if result.get("error") is not None: |
| self.log.error( |
| "Failed to read descriptor for handle {} with err: {}".format( |
| peer_identifier, result.get("error"))) |
| return None |
| return result.get("result") |
| |
| def gatt_client_write_descriptor_by_handle(self, peer_identifier, handle, |
| offset, value): |
| """ Perform a GATT Client write Descriptor to remote peer GATT server |
| database. |
| |
| Args: |
| peer_identifier: The peer to connect to. |
| handle: The Descriptor handle. |
| offset: The offset to start writing at. |
| value: The list of bytes to write. |
| Returns: |
| True if success, False if failure. |
| """ |
| if (not self._find_service_id_and_connect_to_service_for_handle( |
| peer_identifier, handle)): |
| self.log.warn( |
| "Unable to find handle {} in GATT server db.".format(handle)) |
| result = self.device.sl4f.gattc_lib.writeDescriptorById( |
| handle, offset, value) |
| if result.get("error") is not None: |
| self.log.error( |
| "Failed to write descriptor for handle {} with err: {}".format( |
| peer_identifier, result.get("error"))) |
| return None |
| return True |
| |
| def gatt_connect(self, peer_identifier, transport, autoconnect): |
| """ Perform a GATT connection to a perihperal. |
| |
| Args: |
| peer_identifier: The peer to connect to. |
| transport: Not implemented. |
| autoconnect: Not implemented. |
| Returns: |
| True if success, False if failure. |
| """ |
| connection_result = self.device.sl4f.gattc_lib.bleConnectToPeripheral( |
| peer_identifier) |
| if connection_result.get("error") is not None: |
| self.log.error("Failed to connect to peer id {}: {}".format( |
| peer_identifier, connection_result.get("error"))) |
| return False |
| return True |
| |
| def gatt_client_refresh(self, peer_identifier): |
| """ Perform a GATT Client Refresh of a perihperal. |
| |
| Clears the internal cache and forces a refresh of the services from the |
| remote device. In Fuchsia there is no FIDL api to automatically do this |
| yet. Therefore just read all Characteristics which satisfies the same |
| requirements. |
| |
| Args: |
| peer_identifier: The peer to refresh. |
| """ |
| self._read_all_characteristics(peer_identifier) |
| |
| def gatt_client_discover_characteristic_by_uuid(self, peer_identifier, |
| uuid): |
| """ Perform a GATT Client Refresh of a perihperal. |
| |
| Clears the internal cache and forces a refresh of the services from the |
| remote device. In Fuchsia there is no FIDL api to automatically do this |
| yet. Therefore just read all Characteristics which satisfies the same |
| requirements. |
| |
| Args: |
| peer_identifier: The peer to refresh. |
| """ |
| self._read_all_characteristics(peer_identifier, uuid) |
| |
| def gatt_disconnect(self, peer_identifier): |
| """ Perform a GATT disconnect from a perihperal. |
| |
| Args: |
| peer_identifier: The peer to disconnect from. |
| Returns: |
| True if success, False if failure. |
| """ |
| disconnect_result = self.device.sl4f.gattc_lib.bleDisconnectPeripheral( |
| peer_identifier) |
| if disconnect_result.get("error") is not None: |
| self.log.error("Failed to disconnect from peer id {}: {}".format( |
| peer_identifier, disconnect_result.get("error"))) |
| return False |
| return True |
| |
| def reset_bluetooth(self): |
| """Stub for Fuchsia implementation.""" |
| |
| def sdp_add_search(self, attribute_list, profile_id): |
| """Adds an SDP search record. |
| Args: |
| attribute_list: The list of attributes to set |
| profile_id: The profile ID to set. |
| """ |
| return self.device.sl4f.sdp_lib.addSearch(attribute_list, profile_id) |
| |
| def sdp_add_service(self, sdp_record): |
| """Adds an SDP service record. |
| Args: |
| sdp_record: The dictionary representing the search record to add. |
| """ |
| return self.device.sl4f.sdp_lib.addService(sdp_record) |
| |
| def sdp_clean_up(self): |
| """Cleans up all objects related to SDP. |
| """ |
| return self.device.sl4f.sdp_lib.cleanUp() |
| |
| def sdp_init(self): |
| """Initializes SDP on the device. |
| """ |
| return self.device.sl4f.sdp_lib.init() |
| |
| def sdp_remove_service(self, service_id): |
| """Removes a service based on an input id. |
| Args: |
| service_id: The service ID to remove. |
| """ |
| return self.device.sl4f.sdp_lib.init() |
| |
| def start_le_advertisement(self, adv_data, scan_response, adv_interval, |
| connectable): |
| """ Starts an LE advertisement |
| |
| Args: |
| adv_data: Advertisement data. |
| adv_interval: Advertisement interval. |
| """ |
| self.device.sl4f.ble_lib.bleStartBleAdvertising( |
| adv_data, scan_response, adv_interval, connectable) |
| |
| def stop_le_advertisement(self): |
| """ Stop active LE advertisement. |
| """ |
| self.device.sl4f.ble_lib.bleStopBleAdvertising() |
| |
| def setup_gatt_server(self, database): |
| """ Sets up an input GATT server. |
| |
| Args: |
| database: A dictionary representing the GATT database to setup. |
| """ |
| self.device.sl4f.gatts_lib.publishServer(database) |
| |
| def close_gatt_server(self): |
| """ Closes an existing GATT server. |
| """ |
| self.device.sl4f.gatts_lib.closeServer() |
| |
| def le_scan_with_name_filter(self, name, timeout): |
| """ Scan over LE for a specific device name. |
| |
| Args: |
| name: The name filter to set. |
| timeout: The timeout to wait to find the advertisement. |
| Returns: |
| Discovered device id or None |
| """ |
| partial_match = True |
| return le_scan_for_device_by_name(self.device, self.device.log, name, |
| timeout, partial_match) |
| |
| def log_info(self, log): |
| """ Log directly onto the device. |
| |
| Args: |
| log: The informative log. |
| """ |
| self.device.sl4f.logging_lib.logI(log) |
| |
| def unbond_all_known_devices(self): |
| """ Unbond all known remote devices. |
| """ |
| try: |
| device_list = self.device.sl4f.bts_lib.getKnownRemoteDevices( |
| )['result'] |
| for device_info in device_list: |
| device = device_list[device_info] |
| if device['bonded']: |
| self.device.sl4f.bts_lib.forgetDevice(device['id']) |
| except Exception as err: |
| self.log.err("Unable to unbond all devices: {}".format(err)) |
| |
| def unbond_device(self, peer_identifier): |
| """ Unbond peer identifier. |
| |
| Args: |
| peer_identifier: The peer identifier for the peer to unbond. |
| |
| """ |
| self.device.sl4f.bts_lib.forgetDevice(peer_identifier) |
| |
| def _find_service_id_and_connect_to_service_for_handle( |
| self, peer_identifier, handle, uuid=False): |
| fail_err = "Failed to find handle {} in Peer database." |
| if uuid: |
| handle = handle.lower() |
| try: |
| services = self.device.sl4f.gattc_lib.listServices(peer_identifier) |
| for service in services['result']: |
| service_id = service['id'] |
| self.device.sl4f.gattc_lib.connectToService( |
| peer_identifier, service_id) |
| chars = self.device.sl4f.gattc_lib.discoverCharacteristics() |
| |
| for char in chars['result']: |
| char_id = char['id'] |
| if uuid: |
| char_id = char['uuid_type'] |
| if handle == char_id: |
| return True |
| descriptors = char['descriptors'] |
| for desc in descriptors: |
| desc_id = desc["id"] |
| if uuid: |
| desc_id = desc['uuid_type'] |
| if handle == desc_id: |
| return True |
| except Exception as err: |
| self.log.error(fail_err.format(err)) |
| return False |
| |
| def _read_all_characteristics(self, peer_identifier, uuid=None): |
| fail_err = "Failed to read all characteristics with: {}" |
| try: |
| services = self.device.sl4f.gattc_lib.listServices(peer_identifier) |
| for service in services['result']: |
| service_id = service['id'] |
| service_uuid = service['uuid_type'] |
| self.device.sl4f.gattc_lib.connectToService( |
| peer_identifier, service_id) |
| chars = self.device.sl4f.gattc_lib.discoverCharacteristics() |
| self.log.info( |
| "Reading chars in service uuid: {}".format(service_uuid)) |
| |
| for char in chars['result']: |
| char_id = char['id'] |
| char_uuid = char['uuid_type'] |
| if uuid and uuid.lower() not in char_uuid.lower(): |
| continue |
| try: |
| read_val = \ |
| self.device.sl4f.gattc_lib.readCharacteristicById( |
| char_id) |
| self.log.info( |
| "\tCharacteristic uuid / Value: {} / {}".format( |
| char_uuid, read_val['result'])) |
| str_value = "" |
| for val in read_val['result']: |
| str_value += chr(val) |
| self.log.info("\t\tstr val: {}".format(str_value)) |
| except Exception as err: |
| self.log.error(err) |
| except Exception as err: |
| self.log.error(fail_err.forma(err)) |
| |
| def _perform_read_all_descriptors(self, peer_identifier): |
| fail_err = "Failed to read all characteristics with: {}" |
| try: |
| services = self.device.sl4f.gattc_lib.listServices(peer_identifier) |
| for service in services['result']: |
| service_id = service['id'] |
| service_uuid = service['uuid_type'] |
| self.device.sl4f.gattc_lib.connectToService( |
| peer_identifier, service_id) |
| chars = self.device.sl4f.gattc_lib.discoverCharacteristics() |
| self.log.info( |
| "Reading descs in service uuid: {}".format(service_uuid)) |
| |
| for char in chars['result']: |
| char_id = char['id'] |
| char_uuid = char['uuid_type'] |
| descriptors = char['descriptors'] |
| self.log.info( |
| "\tReading descs in char uuid: {}".format(char_uuid)) |
| for desc in descriptors: |
| desc_id = desc["id"] |
| desc_uuid = desc["uuid_type"] |
| try: |
| read_val = self.device.sl4f.gattc_lib.readDescriptorById( |
| desc_id) |
| self.log.info( |
| "\t\tDescriptor uuid / Value: {} / {}".format( |
| desc_uuid, read_val['result'])) |
| except Exception as err: |
| pass |
| except Exception as err: |
| self.log.error(fail_err.format(err)) |
| |
| def init_pair(self, peer_identifier, security_level, non_bondable, |
| transport): |
| """ Send an outgoing pairing request the input peer_identifier. |
| |
| Android currently does not support setting various security levels or |
| bondable modes. Making them available for other bluetooth_device |
| variants. Depending on the Address type, Android will figure out the |
| transport to pair automatically. |
| |
| Args: |
| peer_identifier: A string representing the device id. |
| security_level: The security level required for this pairing request |
| represented as a u64. (Only for LE pairing) |
| Available Values |
| 1 - ENCRYPTED: Encrypted without MITM protection |
| (unauthenticated) |
| 2 - AUTHENTICATED: Encrypted with MITM protection |
| (authenticated) |
| None: No pairing security level. |
| non_bondable: A bool representing whether the pairing mode is |
| bondable or not. None is also accepted. False if bondable, True |
| if non-bondable |
| transport: A u64 representing the transport type. |
| Available Values |
| 1 - BREDR: Classic BR/EDR transport |
| 2 - LE: Bluetooth Low Energy Transport |
| Returns: |
| True if successful, False if failed. |
| """ |
| try: |
| self.device.sl4f.bts_lib.pair(peer_identifier, security_level, |
| non_bondable, transport) |
| return True |
| except Exception as err: |
| fail_err = "Failed to pair to peer_identifier {} with: {}".format( |
| peer_identifier) |
| self.log.error(fail_err.format(err)) |
