Merge changes If1c9f6f2,I3ec69c86,Id5107d96,Icc5d86fc,Ie9a271f0, ... into main (a725833c) · Commits · e / os / android_packages_modules_Bluetooth

floss/pandora/floss/adapter_client.py

+6 −0

Original line number	Diff line number	Diff line
		@@ -830,6 +830,12 @@ class FlossAdapterClient(BluetoothCallbacks, BluetoothConnectionCallbacks):
		device = self._make_dbus_device(address, self.known_devices.get(address, {}).get('name', 'Test device'))
		return bool(self.proxy().DisconnectAllEnabledProfiles(device))

		@utils.glib_call(None)
		def get_connection_state(self, address):
		"""Gets connection state."""
		device = self._make_dbus_device(address, self.known_devices.get(address, {}).get('name', 'Test device'))
		return self.proxy().GetConnectionState(device)

		def wait_for_device_disconnected(self, address):
		"""Waits for the device become disconnected."""

floss/pandora/floss/floss_enums.py

+24 −9

Original line number	Diff line number	Diff line
		@@ -19,8 +19,9 @@ import enum
		class BtTransport(enum.IntEnum):
		"""Bluetooth transport type."""
		AUTO = 0
		BR_EDR = 1
		BREDR = 1
		LE = 2
		DUAL = 3


		class GattWriteRequestStatus(enum.IntEnum):
		@@ -142,14 +143,6 @@ class BondState(enum.IntEnum):
		BONDED = 2


		class Transport(enum.IntEnum):
		"""Bluetooth transport type."""
		AUTO = 0
		BREDR = 1
		LE = 2
		DUAL = 3


		class PairingVariant(enum.IntEnum):
		"""Bluetooth pairing variant type."""
		# SSP variants.
		@@ -178,6 +171,14 @@ class OwnAddressType(enum.IntEnum):
		RANDOM = 1


		class BtConnectionState(enum.IntEnum):
		NOT_CONNECTED = 0
		CONNECTED_ONLY = 1
		ENCRYPTED_BR_EDR = 3
		ENCRYPTED_LE = 5
		ENCRYPTED_BOTH = 7


		class CompanyIdentifiers(enum.IntEnum):
		"""Bluetooth SIG Company ID values.

		@@ -212,3 +213,17 @@ class AdvertisingDataType(enum.IntEnum):
		URI = 0x24
		LE_SUPPORTED_FEATURES = 0x27
		MANUFACTURER_SPECIFIC_DATA = 0xFF


		class BtDiscMode(enum.IntEnum):
		"""Bluetooth discoverable mode."""
		NON_DISCOVERABLE = 0
		LIMITED_DISCOVERABLE = 1
		GENERAL_DISCOVERABLE = 2


		class BthhReportType(enum.IntEnum):
		"""Bluetooth HID report type."""
		INPUT_REPORT = 1
		OUTPUT_REPORT = 2
		FEATURE_REPORT = 3

floss/pandora/floss/gatt_client.py

0 → 100644

+1037 −0

File added.

Preview size limit exceeded, changes collapsed.

floss/pandora/floss/qa_client.py

0 → 100644

+449 −0

Original line number	Diff line number	Diff line
		# Copyright 2023 Google LLC
		#
		# 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
		#
		# https://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.
		"""Client class to access the Floss QA interface."""

		import logging

		from floss.pandora.floss import observer_base
		from floss.pandora.floss import utils


		class BluetoothQACallbacks:
		"""Callbacks for the QA Interface.

		Implement this to observe these callbacks when exporting callbacks via register_callback.
		"""

		def on_fetch_discoverable_mode_completed(self, disc_mode):
		"""Called when fetch discoverable mode completed.

		Args:
		disc_mode: BtDiscMode.
		"""
		pass

		def on_fetch_connectable_completed(self, connectable):
		"""Called when fetch connectable completed.

		Args:
		connectable: A boolean value indicates whether connectable enabled or disabled.
		"""
		pass

		def on_set_connectable_completed(self, succeed):
		"""Called when set connectable completed.

		Args:
		succeed: A boolean value indicates whether the operation is succeeded.
		"""
		pass

		def on_fetch_alias_completed(self, alias):
		"""Called when fetch alias completed.

		Args:
		alias: Alias value as string.
		"""
		pass

		def on_get_hid_report_completed(self, status):
		"""Called when get hid report completed.

		Args:
		status: BtStatus.
		"""
		pass

		def on_set_hid_report_completed(self, status):
		"""Called when set hid report completed.

		Args:
		status: BtStatus.
		"""
		pass

		def on_send_hid_data_completed(self, status):
		"""Called when send hid data completed.

		Args:
		status: BtStatus.
		"""
		pass


		class FlossQAClient(BluetoothQACallbacks):
		"""Handles method calls to and callbacks from the QA interface."""

		QA_SERVICE = 'org.chromium.bluetooth'
		QA_INTERFACE = 'org.chromium.bluetooth.BluetoothQA'
		QA_OBJECT_PATTERN = '/org/chromium/bluetooth/hci{}/qa'
		QA_CB_INTF = 'org.chromium.bluetooth.QACallback'
		QA_CB_OBJ_NAME = 'test_qa_client'

		class ExportedQACallbacks(observer_base.ObserverBase):
		"""
		<node>
		<interface name="org.chromium.bluetooth.QACallback">
		<method name="OnFetchDiscoverableModeComplete">
		<arg type="u" name="disc_mode" direction="in" />
		</method>
		<method name="OnFetchConnectableComplete">
		<arg type="b" name="connectable" direction="in" />
		</method>
		<method name="OnSetConnectableComplete">
		<arg type="b" name="succeed" direction="in" />
		</method>
		<method name="OnFetchAliasComplete">
		<arg type="s" name="alias" direction="in" />
		</method>
		<method name="OnGetHIDReportComplete">
		<arg type="u" name="status" direction="in" />
		</method>
		<method name="OnSetHIDReportComplete">
		<arg type="u" name="status" direction="in" />
		</method>
		<method name="OnSendHIDDataComplete">
		<arg type="u" name="status" direction="in" />
		</method>
		</interface>
		</node>
		"""

		def __init__(self):
		"""Constructs exported callbacks object."""
		observer_base.ObserverBase.__init__(self)

		def OnFetchDiscoverableModeComplete(self, disc_mode):
		"""Handles fetch discoverable mode complete callback.

		Args:
		disc_mode: BtDiscMode.
		"""
		for observer in self.observers.values():
		observer.on_fetch_discoverable_mode_completed(disc_mode)

		def OnFetchConnectableComplete(self, connectable):
		"""Handles fetch connectable complete callback.

		Args:
		connectable: A boolean value indicates whether connectable enabled or disabled.
		"""
		for observer in self.observers.values():
		observer.on_fetch_connectable_completed(connectable)

		def OnSetConnectableComplete(self, succeed):
		"""Handles set connectable complete callback.

		Args:
		succeed: A boolean value indicates whether the operation is succeeded.
		"""
		for observer in self.observers.values():
		observer.on_set_connectable_completed(succeed)

		def OnFetchAliasComplete(self, alias):
		"""Handles fetch alias complete callback.

		Args:
		alias: Alias value as string.
		"""
		for observer in self.observers.values():
		observer.on_fetch_alias_completed(alias)

		def OnGetHIDReportComplete(self, status):
		"""Handles get HID report complete callback.

		Args:
		status: BtStatus.
		"""
		for observer in self.observers.values():
		observer.on_get_hid_report_completed(status)

		def OnSetHIDReportComplete(self, status):
		"""Handles set HID report complete callback.

		Args:
		status: BtStatus.
		"""
		for observer in self.observers.values():
		observer.on_set_hid_report_completed(status)

		def OnSendHIDDataComplete(self, status):
		"""Handles send HID data complete callback.

		Args:
		status: BtStatus.
		"""
		for observer in self.observers.values():
		observer.on_send_hid_data_completed(status)

		def __init__(self, bus, hci):
		"""Constructs the client.

		Args:
		bus: D-Bus bus over which we'll establish connections.
		hci: HCI adapter index. Get this value from `get_default_adapter` on FlossManagerClient.
		"""
		self.bus = bus
		self.hci = hci
		self.objpath = self.QA_OBJECT_PATTERN.format(hci)

		# We don't register callbacks by default.
		self.callbacks = None
		self.callback_id = None

		def __del__(self):
		"""Destructor."""
		del self.callbacks

		@utils.glib_callback()
		def on_fetch_discoverable_mode_completed(self, disc_mode):
		"""Handles fetch discoverable mode completed callback.

		Args:
		disc_mode: BtDiscMode.
		"""
		logging.debug('on_fetch_discoverable_mode_completed: disc_mode: %s', disc_mode)

		@utils.glib_callback()
		def on_fetch_connectable_completed(self, connectable):
		"""Handles fetch connectable completed callback.

		Args:
		connectable: A boolean value indicates whether connectable enabled or disabled.
		"""
		logging.debug('on_fetch_connectable_completed: connectable: %s', connectable)

		@utils.glib_callback()
		def on_set_connectable_completed(self, succeed):
		"""Handles set connectable completed callback.

		Args:
		succeed: A boolean value indicates whether the operation is succeeded.
		"""
		logging.debug('on_set_connectable_completed: succeed: %s', succeed)

		@utils.glib_callback()
		def on_fetch_alias_completed(self, alias):
		"""Handles fetch alias completed callback.

		Args:
		alias: Alias value as string.
		"""
		logging.debug('on_fetch_alias_completed: alias: %s', alias)

		@utils.glib_callback()
		def on_get_hid_report_completed(self, status):
		"""Handles get HID report completed callback.

		Args:
		status: BtStatus.
		"""
		logging.debug('on_get_hid_report_completed: status: %s', status)

		@utils.glib_callback()
		def on_set_hid_report_completed(self, status):
		"""Handles set HID report completed callback.

		Args:
		status: BtStatus.
		"""
		logging.debug('on_set_hid_report_completed: status: %s', status)

		@utils.glib_callback()
		def on_send_hid_data_completed(self, status):
		"""Handles send HID data completed callback.

		Args:
		status: BtStatus.
		"""
		logging.debug('on_send_hid_data_completed: status: %s', status)

		@utils.glib_call(False)
		def has_proxy(self):
		"""Checks whether QA proxy can be acquired."""
		return bool(self.proxy())

		def proxy(self):
		"""Gets proxy object to QA interface for method calls."""
		return self.bus.get(self.QA_SERVICE, self.objpath)[self.QA_INTERFACE]

		@utils.glib_call(False)
		def register_qa_callback(self):
		"""Registers QA callbacks if it doesn't exist."""

		if self.callbacks:
		return True

		# Create and publish callbacks
		self.callbacks = self.ExportedQACallbacks()
		self.callbacks.add_observer('QA_client', self)
		objpath = utils.generate_dbus_cb_objpath(self.QA_CB_OBJ_NAME, self.hci)
		self.bus.register_object(objpath, self.callbacks, None)

		# Register published callbacks with QA daemon
		self.callback_id = self.proxy().RegisterQACallback(objpath)
		return True

		@utils.glib_call(False)
		def unregister_qa_callback(self):
		"""Unregisters QA callbacks for this client.

		Returns:
		True on success, False otherwise.
		"""
		return self.proxy().UnregisterQACallback(self.callback_id)

		def register_callback_observer(self, name, observer):
		"""Add an observer for all callbacks.

		Args:
		name: Name of the observer.
		observer: Observer that implements all callback classes.
		"""
		if isinstance(observer, BluetoothQACallbacks):
		self.callbacks.add_observer(name, observer)

		def unregister_callback_observer(self, name, observer):
		"""Remove an observer for all callbacks.

		Args:
		name: Name of the observer.
		observer: Observer that implements all callback classes.
		"""
		if isinstance(observer, BluetoothQACallbacks):
		self.callbacks.remove_observer(name, observer)

		@utils.glib_call(False)
		def add_media_player(self, name, browsing_supported):
		"""Adds media player.

		Args:
		name: Media player name.
		browsing_supported: A boolean value indicates whether browsing_supported or not.

		Returns:
		True on success, False otherwise.
		"""
		self.proxy().AddMediaPlayer(name, browsing_supported)
		return True

		@utils.glib_call(False)
		def rfcomm_send_msc(self, dlci, addr):
		"""Sends MSC command over RFCOMM to the remote device.

		Args:
		dlci: The Data Link Control Identifier (DLCI) for the RFCOMM channel.
		addr: The Bluetooth address of the remote device.

		Returns:
		True on success, False otherwise.
		"""
		self.proxy().RfcommSendMsc(dlci, addr)
		return True

		@utils.glib_call(False)
		def fetch_discoverable_mode(self):
		"""Fetches discoverable mode.

		Returns:
		True on success, False otherwise.
		"""
		self.proxy().FetchDiscoverableMode()
		return True

		@utils.glib_call(False)
		def fetch_connectable(self):
		"""Fetches connectable.

		Returns:
		True on success, False otherwise.
		"""
		self.proxy().FetchConnectable()
		return True

		@utils.glib_call(False)
		def set_connectable(self, mode):
		"""Sets connectable mode.

		Args:
		mode: A boolean value indicates whether connectable mode enabled or disabled.

		Returns:
		True on success, False otherwise.
		"""
		self.proxy().SetConnectable(mode)
		return True

		@utils.glib_call(False)
		def fetch_alias(self):
		"""Fetches alias.

		Returns:
		True on success, False otherwise.
		"""
		self.proxy().FetchAlias()
		return True

		@utils.glib_call(None)
		def get_modalias(self):
		"""Gets modalias.

		Returns:
		Modalias value on success, None otherwise.
		"""
		return self.proxy().GetModalias()

		@utils.glib_call(False)
		def get_hid_report(self, addr, report_type, report_id):
		"""Gets HID report on the remote device.

		Args:
		addr: The Bluetooth address of the remote device.
		report_type: The type of HID report.
		report_id: The id of HID report.

		Returns:
		True on success, False otherwise.
		"""
		self.proxy().FetchHIDReport(addr, report_type, report_id)
		return True

		@utils.glib_call(False)
		def set_hid_report(self, addr, report_type, report):
		"""Sets HID report to the remote device.

		Args:
		addr: The Bluetooth address of the remote device.
		report_type: The type of HID report.
		report: The HID report to be set.

		Returns:
		True on success, False otherwise.
		"""
		self.proxy().SetHIDReport(addr, report_type, report)
		return True

		@utils.glib_call(False)
		def send_hid_data(self, addr, data):
		"""Sends HID report data to the remote device.

		Args:
		addr: The Bluetooth address of the remote device.
		data: The HID report data to be sent.

		Returns:
		True on success, False otherwise.
		"""
		self.proxy().SendHIDData(addr, data)
		return True

floss/pandora/floss/utils.py

+30 −5

Original line number	Diff line number	Diff line
		@@ -13,16 +13,17 @@
		# limitations under the License.
		"""All floss utils functions."""

		from typing import List, Optional

		import functools
		import logging
		import threading
		import time
		from typing import List, Optional

		from floss.pandora.floss import floss_enums
		from gi.repository import GLib
		from google.protobuf import any_pb2
		from pandora import host_pb2
		from pandora_experimental import os_pb2

		# All GLIB method calls should wait this many seconds by default
		GLIB_METHOD_CALL_TIMEOUT = 2
		@@ -343,15 +344,39 @@ class PropertySet:
		return setter(*args)


		class Connection:
		"""A Bluetooth connection."""

		def __init__(self, address: str, transport: floss_enums.BtTransport):
		self.address = address
		self.transport = transport


		def connection_to(connection: Connection):
		"""Converts Connection from Floss format to gRPC format."""
		internal_connection_ref = os_pb2.InternalConnectionRef(address=address_to(connection.address),
		transport=connection.transport)
		cookie = any_pb2.Any(value=internal_connection_ref.SerializeToString())
		return host_pb2.Connection(cookie=cookie)


		def connection_from(connection: host_pb2.Connection):
		"""Converts Connection from gRPC format to Floss format."""
		internal_connection_ref = os_pb2.InternalConnectionRef()
		internal_connection_ref.ParseFromString(connection.cookie.value)
		return Connection(address=address_from(internal_connection_ref.address),
		transport=internal_connection_ref.transport)


		def address_from(request_address: bytes):
		"""Converts address from grpc server format to floss format."""
		"""Converts address from gRPC format to Floss format."""
		address = request_address.hex()
		address = f'{address[:2]}:{address[2:4]}:{address[4:6]}:{address[6:8]}:{address[8:10]}:{address[10:12]}'
		return address.upper()


		def address_to(address: str):
		"""Converts address from floss format to grpc server format."""
		"""Converts address from Floss format to gRPC format."""
		request_address = bytes.fromhex(address.replace(':', ''))
		return request_address

		@@ -367,7 +392,7 @@ def uuid32_to_uuid128(uuid32: str):
		def advertise_data_from(request_data: host_pb2.DataTypes):
		"""Mapping DataTypes to a dict.

		The dict content follows the format of floss AdvertiseData.
		The dict content follows the format of Floss AdvertiseData.

		Args:
		request_data : advertising data.