Merge "a2dp: Implement first a2dp avatar test" into main am: 0dcdecf230 (a221516c) · Commits · e / os / android_packages_modules_Bluetooth

android/pandora/server/src/A2dp.kt

+7 −0

Original line number	Diff line number	Diff line
		@@ -161,6 +161,11 @@ class A2dp(val context: Context) : A2DPImplBase(), Closeable {
		throw RuntimeException("Device is not connected, cannot start")
		}

		// Configure the selected device as active device if it is not
		// already.
		bluetoothA2dp.setActiveDevice(device)

		// Play an audio track.
		audioTrack!!.play()

		// If A2dp is not already playing, wait for it
		@@ -281,6 +286,7 @@ class A2dp(val context: Context) : A2DPImplBase(), Closeable {
		)
		}
		}

		override fun onError(t: Throwable) {
		t.printStackTrace()
		val sw = StringWriter()
		@@ -289,6 +295,7 @@ class A2dp(val context: Context) : A2DPImplBase(), Closeable {
		Status.UNKNOWN.withCause(t).withDescription(sw.toString()).asException()
		)
		}

		override fun onCompleted() {
		responseObserver.onNext(PlaybackAudioResponse.getDefaultInstance())
		responseObserver.onCompleted()

android/pandora/test/a2dp_test.py

0 → 100644

+257 −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.

		import asyncio
		import avatar
		import itertools
		import logging
		import math
		import numpy as np
		import os

		from avatar import BumblePandoraDevice, PandoraDevice, PandoraDevices
		from bumble import avdtp
		from bumble.a2dp import (
		A2DP_SBC_CODEC_TYPE,
		SBC_DUAL_CHANNEL_MODE,
		SBC_JOINT_STEREO_CHANNEL_MODE,
		SBC_LOUDNESS_ALLOCATION_METHOD,
		SBC_MONO_CHANNEL_MODE,
		SBC_SNR_ALLOCATION_METHOD,
		SBC_STEREO_CHANNEL_MODE,
		SbcMediaCodecInformation,
		make_audio_sink_service_sdp_records,
		)
		from bumble.avdtp import (
		AVDTP_AUDIO_MEDIA_TYPE,
		AVDTP_OPEN_STATE,
		AVDTP_STREAMING_STATE,
		Listener,
		MediaCodecCapabilities,
		)
		from bumble.pairing import PairingDelegate
		from mobly import base_test, test_runner
		from mobly.asserts import assert_equal # type: ignore
		from mobly.asserts import assert_in # type: ignore
		from mobly.asserts import assert_is_none # type: ignore
		from mobly.asserts import assert_is_not_none # type: ignore
		from mobly.asserts import fail # type: ignore
		from pandora.a2dp_grpc_aio import A2DP
		from pandora.a2dp_pb2 import PlaybackAudioRequest, Sink, Source
		from pandora.host_pb2 import Connection
		from pandora.security_pb2 import LEVEL2
		from threading import Thread
		from typing import Optional


		async def initiate_pairing(device, address) -> Connection:
		"""Connect and pair a remote device."""

		result = await device.aio.host.Connect(address=address)
		connection = result.connection
		assert connection

		bond = await device.aio.security.Secure(connection=connection, classic=LEVEL2)
		assert bond.success

		return connection


		async def accept_pairing(device, address) -> Connection:
		"""Accept connection and pairing from a remote device."""

		result = await device.aio.host.WaitConnection(address=address)
		connection = result.connection
		assert connection

		bond = await device.aio.security.WaitSecurity(connection=connection, classic=LEVEL2)
		assert bond.success

		return connection


		async def open_source(device, connection) -> Source:
		"""Initiate AVDTP connection from Android device."""

		result = await device.a2dp.OpenSource(connection=connection)
		source = result.source
		assert source

		return source


		def codec_capabilities():
		"""Codec capabilities for the Bumble sink devices."""

		return MediaCodecCapabilities(
		media_type=AVDTP_AUDIO_MEDIA_TYPE,
		media_codec_type=A2DP_SBC_CODEC_TYPE,
		media_codec_information=SbcMediaCodecInformation.from_lists(
		sampling_frequencies=[48000, 44100, 32000, 16000],
		channel_modes=[
		SBC_MONO_CHANNEL_MODE,
		SBC_DUAL_CHANNEL_MODE,
		SBC_STEREO_CHANNEL_MODE,
		SBC_JOINT_STEREO_CHANNEL_MODE,
		],
		block_lengths=[4, 8, 12, 16],
		subbands=[4, 8],
		allocation_methods=[
		SBC_LOUDNESS_ALLOCATION_METHOD,
		SBC_SNR_ALLOCATION_METHOD,
		],
		minimum_bitpool_value=2,
		maximum_bitpool_value=53,
		),
		)


		class AudioSignal:
		"""Audio signal generator and verifier."""

		SINE_FREQUENCY = 440
		SINE_DURATION = 0.1

		def __init__(self, a2dp: A2DP, source: Source, amplitude, fs):
		"""Init AudioSignal class.

		Args:
		a2dp: A2DP profile interface.
		source: Source connection object to send the data to.
		amplitude: amplitude of the signal to generate.
		fs: sampling rate of the signal to generate.
		"""
		self.a2dp = a2dp
		self.source = source
		self.amplitude = amplitude
		self.fs = fs
		self.task = None

		def start(self):
		"""Generates the audio signal and send it to the transport."""
		self.task = asyncio.create_task(self._run())

		async def _run(self):
		sine = self._generate_sine(self.SINE_FREQUENCY, self.SINE_DURATION)

		# Interleaved audio.
		stereo = np.zeros(sine.size * 2, dtype=sine.dtype)
		stereo[0::2] = sine

		# Send 4 second of audio.
		audio = itertools.repeat(stereo.tobytes(), int(4 / self.SINE_DURATION))

		for frame in audio:
		await self.a2dp.PlaybackAudio(PlaybackAudioRequest(data=frame, source=self.source))

		def _generate_sine(self, f, duration):
		sine = self.amplitude * np.sin(2 * np.pi * np.arange(self.fs * duration) * (f / self.fs))
		s16le = (sine * 32767).astype('<i2')
		return s16le


		class A2dpTest(base_test.BaseTestClass): # type: ignore[misc]
		"""A2DP test suite."""

		devices: Optional[PandoraDevices] = None

		# pandora devices.
		dut: PandoraDevice
		ref1: PandoraDevice
		ref2: PandoraDevice

		@avatar.asynchronous
		async def setup_class(self) -> None:
		self.devices = PandoraDevices(self)
		self.dut, self.ref1, self.ref2, *_ = self.devices

		if not isinstance(self.ref1, BumblePandoraDevice):
		raise signals.TestAbortClass('Test require Bumble as reference device(s)')
		if not isinstance(self.ref2, BumblePandoraDevice):
		raise signals.TestAbortClass('Test require Bumble as reference device(s)')

		# Enable BR/EDR mode and SSP for Bumble devices.
		for device in self.devices:
		if isinstance(device, BumblePandoraDevice):
		device.config.setdefault('classic_enabled', True)
		device.config.setdefault('classic_ssp_enabled', True)
		device.config.setdefault('classic_smp_enabled', False)
		device.server_config.io_capability = PairingDelegate.NO_OUTPUT_NO_INPUT

		await asyncio.gather(self.dut.reset(), self.ref1.reset(), self.ref2.reset())

		self.dut.a2dp = A2DP(channel=self.dut.aio.channel)

		handle = 0x00010001
		self.ref1.device.sdp_service_records = {handle: make_audio_sink_service_sdp_records(handle)}
		self.ref2.device.sdp_service_records = {handle: make_audio_sink_service_sdp_records(handle)}

		self.ref1.a2dp = Listener.for_device(self.ref1.device)
		self.ref2.a2dp = Listener.for_device(self.ref2.device)
		self.ref1.a2dp_sink = None
		self.ref2.a2dp_sink = None

		def on_ref1_avdtp_connection(server):
		self.ref1.a2dp_sink = server.add_sink(codec_capabilities())

		def on_ref2_avdtp_connection(server):
		self.ref2.a2dp_sink = server.add_sink(codec_capabilities())

		self.ref1.a2dp.on('connection', on_ref1_avdtp_connection)
		self.ref2.a2dp.on('connection', on_ref2_avdtp_connection)

		def teardown_class(self) -> None:
		if self.devices:
		self.devices.stop_all()

		@avatar.asynchronous
		async def setup_test(self) -> None:
		pass

		@avatar.asynchronous
		async def test_connect_and_stream(self) -> None:
		"""Basic A2DP connection and streaming test.
		This test wants to be a template to be reused for other tests.

		1. Pair and Connect RD1
		2. Start streaming
		3. Check AVDTP status on RD1
		4. Stop streaming
		5. Check AVDTP status on RD1
		"""
		# Connect and pair RD1.
		dut_ref1, ref1_dut = await asyncio.gather(
		initiate_pairing(self.dut, self.ref1.address),
		accept_pairing(self.ref1, self.dut.address),
		)

		# Connect AVDTP to RD1.
		dut_ref1_source = await open_source(self.dut, dut_ref1)
		assert_is_not_none(self.ref1.a2dp_sink)
		assert_is_not_none(self.ref1.a2dp_sink.stream)
		assert_in(self.ref1.a2dp_sink.stream.state, [AVDTP_OPEN_STATE, AVDTP_STREAMING_STATE])

		# Start streaming to RD1.
		await self.dut.a2dp.Start(source=dut_ref1_source)
		audio = AudioSignal(self.dut.a2dp, dut_ref1_source, 0.8, 44100)
		assert_equal(self.ref1.a2dp_sink.stream.state, AVDTP_STREAMING_STATE)

		# Stop streaming to RD1.
		await self.dut.a2dp.Suspend(source=dut_ref1_source)
		assert_equal(self.ref1.a2dp_sink.stream.state, AVDTP_OPEN_STATE)


		if __name__ == '__main__':
		logging.basicConfig(level=logging.DEBUG)
		test_runner.main() # type: ignore

android/pandora/test/main.py

+5 −2

Original line number	Diff line number	Diff line
		@@ -13,6 +13,8 @@ from typing import List, Tuple

		_BUMBLE_BTSNOOP_FMT = 'bumble_btsnoop_{pid}_{instance}.log'

		import a2dp_test

		# Import test cases modules.
		import asha_test
		import avatar.cases.host_test
		@@ -21,14 +23,15 @@ import avatar.cases.le_security_test
		import avatar.cases.security_test
		import gatt_test
		import hfpclient_test
		import sdp_test
		import pairing.smp_test as smp_test
		import sdp_test

		_TEST_CLASSES_LIST = [
		avatar.cases.host_test.HostTest,
		avatar.cases.le_host_test.LeHostTest,
		avatar.cases.security_test.SecurityTest,
		avatar.cases.le_security_test.LeSecurityTest,
		a2dp_test.A2dpTest,
		sdp_test.SdpTest,
		smp_test.SmpTest,
		gatt_test.GattTest,