Floss: Implements HID of floss bluetooth test server. (dded461e) · Commits · e / os / android_packages_modules_Bluetooth

floss/pandora/server/bluetooth.py

+11 −1

Original line number	Diff line number	Diff line
		@@ -20,6 +20,7 @@ import traceback
		from floss.pandora.floss import adapter_client
		from floss.pandora.floss import advertising_client
		from floss.pandora.floss import manager_client
		from floss.pandora.floss import qa_client
		from floss.pandora.floss import scanner_client
		from floss.pandora.floss import utils
		from gi.repository import GLib
		@@ -58,6 +59,7 @@ class Bluetooth(object):
		self.adapter_client = adapter_client.FlossAdapterClient(self.bus, self.DEFAULT_ADAPTER)
		self.advertising_client = advertising_client.FlossAdvertisingClient(self.bus, self.DEFAULT_ADAPTER)
		self.scanner_client = scanner_client.FlossScannerClient(self.bus, self.DEFAULT_ADAPTER)
		self.qa_client = qa_client.FlossQAClient(self.bus, self.DEFAULT_ADAPTER)

		def __del__(self):
		if not self.is_clean:
		@@ -119,6 +121,9 @@ class Bluetooth(object):
		if not self.scanner_client.register_scanner_callback():
		logging.error('scanner_client: Failed to register callbacks')
		return False
		if not self.qa_client.register_qa_callback():
		logging.error('qa_client: Failed to register callbacks')
		return False
		return True

		def is_bluetoothd_proxy_valid(self):
		@@ -128,7 +133,8 @@ class Bluetooth(object):
		self.manager_client.has_proxy(),
		self.adapter_client.has_proxy(),
		self.advertising_client.has_proxy(),
		self.scanner_client.has_proxy()
		self.scanner_client.has_proxy(),
		self.qa_client.has_proxy()
		])

		if not proxy_ready:
		@@ -161,6 +167,7 @@ class Bluetooth(object):
		self.adapter_client = adapter_client.FlossAdapterClient(self.bus, default_adapter)
		self.advertising_client = advertising_client.FlossAdvertisingClient(self.bus, default_adapter)
		self.scanner_client = scanner_client.FlossScannerClient(self.bus, default_adapter)
		self.qa_client = qa_client.FlossQAClient(self.bus, default_adapter)

		try:
		utils.poll_for_condition(
		@@ -277,3 +284,6 @@ class Bluetooth(object):
		logging.error('Failed to stop scanning.')
		return False
		return True

		def set_hid_report(self, addr, report_type, report):
		return self.qa_client.set_hid_report(addr, report_type, report)

floss/pandora/server/hid.py

0 → 100644

+76 −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.
		"""HID grpc interface."""

		import asyncio
		import logging

		from floss.pandora.floss import floss_enums
		from floss.pandora.floss import qa_client
		from floss.pandora.floss import utils
		from floss.pandora.server import bluetooth as bluetooth_module
		from google.protobuf import empty_pb2
		import grpc
		from pandora_experimental import hid_grpc_aio
		from pandora_experimental import hid_pb2


		class HIDService(hid_grpc_aio.HIDServicer):
		"""Service to trigger Bluetooth HID procedures.

		This class implements the Pandora bluetooth test interfaces,
		where the meta class definition is automatically generated by the protobuf.
		The interface definition can be found in:
		https://cs.android.com/android/platform/superproject/+/main:packages/modules/Bluetooth/pandora/interfaces/pandora_experimental/hid.proto
		"""

		def __init__(self, bluetooth: bluetooth_module.Bluetooth):
		self.bluetooth = bluetooth

		async def SendHostReport(self, request: hid_pb2.SendHostReportRequest,
		context: grpc.ServicerContext) -> hid_pb2.SendHostReportResponse:

		class HIDReportObserver(qa_client.BluetoothQACallbacks):
		"""Observer to observe the set HID report state."""

		def __init__(self, task):
		self.task = task

		@utils.glib_callback()
		def on_set_hid_report_completed(self, status):
		if floss_enums.BtStatus(status) != floss_enums.BtStatus.SUCCESS:
		logging.error('Failed to set HID report. Status: %s', status)
		future = self.task['set_hid_report']
		future.get_loop().call_soon_threadsafe(future.set_result, None)

		if request.address is None:
		raise ValueError('Request address field must be set.')
		address = utils.address_from(request.address)

		try:
		set_hid_report = asyncio.get_running_loop().create_future()
		observer = HIDReportObserver({'set_hid_report': set_hid_report})
		name = utils.create_observer_name(observer)
		self.bluetooth.qa_client.register_callback_observer(name, observer)
		if request.report_type not in iter(floss_enums.BthhReportType):
		raise ValueError('Invalid report type.')
		if not self.bluetooth.set_hid_report(address, request.report_type, request.report):
		raise RuntimeError('Failed to call set_hid_report.')

		await asyncio.wait_for(set_hid_report, timeout=5)

		finally:
		self.bluetooth.qa_client.unregister_callback_observer(name, observer)

		return empty_pb2.Empty()

floss/pandora/server/server.py

+5 −0

Original line number	Diff line number	Diff line
		@@ -17,11 +17,13 @@ import asyncio
		import logging

		from floss.pandora.server import bluetooth as bluetooth_module
		from floss.pandora.server import hid
		from floss.pandora.server import host
		from floss.pandora.server import security
		import grpc
		from pandora import host_grpc_aio
		from pandora import security_grpc_aio
		from pandora_experimental import hid_grpc_aio


		async def serve(port):
		@@ -45,6 +47,9 @@ async def serve(port):
		security_storage_service = security.SecurityStorageService(server, bluetooth)
		security_grpc_aio.add_SecurityStorageServicer_to_server(security_storage_service, server)

		hid_service = hid.HIDService(bluetooth)
		hid_grpc_aio.add_HIDServicer_to_server(hid_service, server)

		server.add_insecure_port(f'[::]:{port}')

		await server.start()