Floss: Implement Pandora A2DP profile (f28adda7) · Commits · e / os / android_packages_modules_Bluetooth

floss/pandora/floss/audio_test_data.py

0 → 100644

+50 −0

Original line number	Diff line number	Diff line
		# Copyright 2024 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.
		"""This module provides audio test data."""

		import os


		class AudioTestDataException(Exception):
		"""Exception for audio test data."""
		pass


		class AudioTestData(object):
		"""Class to represent audio test data."""

		def __init__(self, data_format=None, path=None, frequencies=None, duration_secs=None):
		"""Initializes an audio test file.

		Args:
		data_format: A dict containing data format including
		file_type, sample_format, channel, and rate.
		file_type: file type e.g. 'raw' or 'wav'.
		sample_format: One of the keys in audio_utils.SAMPLE_FORMAT.
		channel: number of channels.
		rate: sampling rate.
		path: The path to the file.
		frequencies: A list containing the frequency of each channel in this file.
		Only applicable to data of sine tone.
		duration_secs: Duration of test file in seconds.

		Raises:
		AudioTestDataException if the path does not exist.
		"""
		self.data_format = data_format
		if not os.path.exists(path):
		raise AudioTestDataException('Can not find path %s' % path)
		self.path = path
		self.frequencies = frequencies
		self.duration_secs = duration_secs

floss/pandora/floss/audio_utils.py

0 → 100644

+187 −0

Original line number	Diff line number	Diff line
		# Copyright 2024 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 logging
		import os
		import subprocess
		import wave

		from floss.pandora.floss import audio_test_data
		from floss.pandora.floss import cras_utils
		from floss.pandora.floss import sox_utils
		from floss.pandora.floss import utils

		CRAS_BLUETOOTH_OUTPUT_NODE_TYPE = 'BLUETOOTH'

		AUDIO_TEST_DIR = '/tmp/audio'

		A2DP_TEST_DATA = {
		'rate': 48000,
		'channels': 2,
		'frequencies': (440, 20000),
		'file': os.path.join(AUDIO_TEST_DIR, 'binaural_sine_440hz_20000hz_rate48000_5secs.wav'),
		'recorded_by_sink': os.path.join(AUDIO_TEST_DIR, 'a2dp_recorded_by_sink.wav'),
		'chunk_in_secs': 5,
		'bit_width': 16,
		'format': 'S16_LE',
		'duration': 5,
		}

		A2DP_PLAYBACK_DATA = {
		'rate': 44100,
		'channels': 2,
		'file': os.path.join(AUDIO_TEST_DIR, 'audio_playback.wav'),
		'sample_width': 2
		}

		SAMPLE_FORMATS = dict(S32_LE=dict(message='Signed 32-bit integer, little-endian', dtype_str='<i', size_bytes=4),
		S16_LE=dict(message='Signed 16-bit integer, little-endian', dtype_str='<i', size_bytes=2))


		@utils.dbus_safe(None)
		def get_selected_output_device_type():
		"""Gets the selected audio output node type.

		Returns:
		The node type of the selected output device.
		"""
		return str(cras_utils.get_selected_output_device_type())


		@utils.dbus_safe(None)
		def select_output_node(node_type):
		"""Selects the audio output node.

		Args:
		node_type: The node type of the Bluetooth peer device.

		Returns:
		True if the operation succeeds.
		"""
		return cras_utils.set_single_selected_output_node(node_type)


		def select_audio_output_node():
		"""Selects the audio output node through cras."""

		def bluetooth_type_selected(node_type):
		"""Checks if the bluetooth node type is selected."""
		selected = get_selected_output_device_type()
		logging.debug('active output node type: %s, expected %s', selected, node_type)
		return selected == node_type

		node_type = CRAS_BLUETOOTH_OUTPUT_NODE_TYPE
		if not select_output_node(node_type):
		return False

		desc = 'waiting for %s as active cras audio output node type' % node_type
		try:
		utils.poll_for_condition(condition=lambda: bluetooth_type_selected(node_type),
		timeout=20,
		sleep_interval=1,
		desc=desc)
		except TimeoutError:
		return False
		return True


		def generate_audio_test_data(path, data_format=None, frequencies=None, duration_secs=None, volume_scale=None):
		"""Generates audio test data with specified format and frequencies.

		Args:
		path: The path to the file.
		data_format: A dict containing data format including
		file_type, sample_format, channel, and rate.
		file_type: file type e.g. 'raw' or 'wav'.
		sample_format: One of the keys in audio_data.SAMPLE_FORMAT.
		channel: number of channels.
		ate: sampling rate.
		frequencies: A list containing the frequency of each channel in this file.
		Only applicable to data of sine tone.
		duration_secs: Duration of test file in seconds.
		volume_scale: A float for volume scale used in sox command.
		E.g. 0.5 to scale volume by half. -1.0 to invert.

		Returns:
		An AudioTestData object.
		"""
		sox_file_path = path

		if data_format is None:
		data_format = dict(file_type='wav', sample_format='S16_LE', channel=2, rate=48000)

		sample_format = SAMPLE_FORMATS[data_format['sample_format']]
		bits = sample_format['size_bytes'] * 8

		command = sox_utils.generate_sine_tone_cmd(filename=sox_file_path,
		channels=data_format['channel'],
		bits=bits,
		rate=data_format['rate'],
		duration=duration_secs,
		frequencies=frequencies,
		vol=volume_scale,
		raw=(data_format['file_type'] == 'raw'))

		logging.info(' '.join(command))
		subprocess.check_call(command)

		test_data = audio_test_data.AudioTestData(data_format=data_format,
		path=sox_file_path,
		frequencies=frequencies,
		duration_secs=duration_secs)

		return test_data


		def generate_playback_file(audio_data):
		"""Generates the playback file if it does not exist yet.

		Some audio test files may be large. Generate them on the fly to save the storage of the source tree.

		Args:
		audio_data: The audio test data.
		"""
		directory = os.path.dirname(audio_data['file'])
		if not os.path.exists(directory):
		os.makedirs(directory)

		if not os.path.exists(audio_data['file']):
		data_format = dict(file_type='wav',
		sample_format='S16_LE',
		channel=audio_data['channels'],
		rate=audio_data['rate'])
		generate_audio_test_data(data_format=data_format,
		path=audio_data['file'],
		duration_secs=audio_data['duration'],
		frequencies=audio_data['frequencies'])
		logging.debug('Audio file generated: %s', audio_data['file'])


		def generate_playback_file_from_binary_data(audio_data):
		"""Generates wav audio file from binary audio data.

		Args:
		audio_data: The binary audio data.
		"""
		directory = os.path.dirname(A2DP_PLAYBACK_DATA['file'])
		if not os.path.exists(directory):
		os.makedirs(directory)

		with wave.open(A2DP_PLAYBACK_DATA['file'], 'wb') as wav_file:
		wav_file.setnchannels(A2DP_PLAYBACK_DATA['channels'])
		wav_file.setframerate(A2DP_PLAYBACK_DATA['rate'])
		wav_file.setsampwidth(A2DP_PLAYBACK_DATA['sample_width'])
		wav_file.writeframes(audio_data)

		logging.debug('wav file generated from binary data: %s', A2DP_PLAYBACK_DATA['file'])

floss/pandora/floss/cmd_utils.py

0 → 100644

+228 −0

Original line number	Diff line number	Diff line
		# Copyright 2024 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 atexit
		import itertools
		import logging
		import os
		import pipes
		import pwd
		import select
		import subprocess
		import threading

		TEE_TO_LOGS = object()
		_popen_lock = threading.Lock()
		_logging_service = None
		_command_serial_number = itertools.count(1)

		_LOG_BUFSIZE = 4096
		_PIPE_CLOSED = -1


		class _LoggerProxy(object):

		def __init__(self, logger):
		self._logger = logger

		def fileno(self):
		"""Returns the fileno of the logger pipe."""
		return self._logger._pipe[1]

		def __del__(self):
		self._logger.close()


		class _PipeLogger(object):

		def __init__(self, level, prefix):
		self._pipe = list(os.pipe())
		self._level = level
		self._prefix = prefix

		def close(self):
		"""Closes the logger."""
		if self._pipe[1] != _PIPE_CLOSED:
		os.close(self._pipe[1])
		self._pipe[1] = _PIPE_CLOSED


		class _LoggingService(object):

		def __init__(self):
		# Python's list is thread safe
		self._loggers = []

		# Change tuple to list so that we can change the value when
		# closing the pipe.
		self._pipe = list(os.pipe())
		self._thread = threading.Thread(target=self._service_run)
		self._thread.daemon = True
		self._thread.start()

		def _service_run(self):
		terminate_loop = False
		while not terminate_loop:
		rlist = [l._pipe[0] for l in self._loggers]
		rlist.append(self._pipe[0])
		for r in select.select(rlist, [], [])[0]:
		data = os.read(r, _LOG_BUFSIZE)
		if r != self._pipe[0]:
		self._output_logger_message(r, data)
		elif len(data) == 0:
		terminate_loop = True
		# Release resources.
		os.close(self._pipe[0])
		for logger in self._loggers:
		os.close(logger._pipe[0])

		def _output_logger_message(self, r, data):
		logger = next(l for l in self._loggers if l._pipe[0] == r)

		if len(data) == 0:
		os.close(logger._pipe[0])
		self._loggers.remove(logger)
		return

		for line in data.split('\n'):
		logging.log(logger._level, '%s%s', logger._prefix, line)

		def create_logger(self, level=logging.DEBUG, prefix=''):
		"""Creates a new logger.

		Args:
		level: The desired logging level.
		prefix: The prefix to add to each log entry.
		"""
		logger = _PipeLogger(level=level, prefix=prefix)
		self._loggers.append(logger)
		os.write(self._pipe[1], '\0')
		return _LoggerProxy(logger)

		def shutdown(self):
		"""Shuts down the logger."""
		if self._pipe[1] != _PIPE_CLOSED:
		os.close(self._pipe[1])
		self._pipe[1] = _PIPE_CLOSED
		self._thread.join()


		def create_logger(level=logging.DEBUG, prefix=''):
		"""Creates a new logger.

		Args:
		level: The desired logging level.
		prefix: The prefix to add to each log entry.
		"""
		global _logging_service
		if _logging_service is None:
		_logging_service = _LoggingService()
		atexit.register(_logging_service.shutdown)
		return _logging_service.create_logger(level=level, prefix=prefix)


		def wait_and_check_returncode(*popens):
		"""Waits for all the Popens and check the return code is 0.

		Args:
		popens: The Popens to be checked.

		Raises:
		RuntimeError if the return code is not 0.
		"""
		error_message = None
		for p in popens:
		if p.wait() != 0:
		error_message = ('Command failed(%d, %d): %s' % (p.pid, p.returncode, p.command))
		logging.error(error_message)
		if error_message:
		raise RuntimeError(error_message)


		def execute(args, stdin=None, stdout=TEE_TO_LOGS, stderr=TEE_TO_LOGS, run_as=None):
		"""Executes a child command and wait for it.

		Args:
		args: The command to be executed.
		stdin: The executed program's standard input.
		stdout: The executed program's standard output.
		stderr: The executed program's standard error.
		run_as: If not None, run the command as the given user.

		Returns:
		The output from standard output if 'stdout' is subprocess.PIPE.

		Raises:
		RuntimeException if the return code of the child command is not 0.
		"""
		ps = popen(args, stdin=stdin, stdout=stdout, stderr=stderr, run_as=run_as)
		out = ps.communicate()[0] if stdout == subprocess.PIPE else None
		wait_and_check_returncode(ps)
		return out


		def _run_as(user):
		"""Changes the uid and gid of the running process to be that of the
		given user and configures its supplementary groups.

		Don't call this function directly, instead wrap it in a lambda and
		pass that to the preexec_fn argument of subprocess.Popen.

		Example usage:
		subprocess.Popen(..., preexec_fn=lambda: _run_as('chronos'))

		Args:
		user: The user to run as.
		"""
		pw = pwd.getpwnam(user)
		os.setgid(pw.pw_gid)
		os.initgroups(user, pw.pw_gid)
		os.setuid(pw.pw_uid)


		def popen(args, stdin=None, stdout=TEE_TO_LOGS, stderr=TEE_TO_LOGS, env=None, run_as=None):
		"""Returns a Popen object just as subprocess.Popen does but with the
		executed command stored in Popen.command.

		Args:
		args: The command to be executed.
		stdin: The executed program's standard input.
		stdout: The executed program's standard output.
		stderr: The executed program's standard error.
		env: The executed program's environment.
		run_as: If not None, run the command as the given user.
		"""
		command_id = next(_command_serial_number)
		prefix = '[%04d] ' % command_id

		if stdout is TEE_TO_LOGS:
		stdout = create_logger(level=logging.DEBUG, prefix=prefix)
		if stderr is TEE_TO_LOGS:
		stderr = create_logger(level=logging.ERROR, prefix=prefix)

		command = ' '.join(pipes.quote(x) for x in args)
		logging.info('%sRunning: %s', prefix, command)

		preexec_fn = None
		if run_as is not None:
		preexec_fn = lambda: _run_as(run_as)

		# The lock is required for http://crbug.com/323843.
		with _popen_lock:
		ps = subprocess.Popen(args, stdin=stdin, stdout=stdout, stderr=stderr, env=env, preexec_fn=preexec_fn)

		logging.info('%spid is %d', prefix, ps.pid)
		ps.command_id = command_id
		ps.command = command
		return ps

floss/pandora/floss/cras_utils.py

0 → 100644

+852 −0

File added.

Preview size limit exceeded, changes collapsed.

floss/pandora/floss/media_client.py

+26 −0

Original line number	Diff line number	Diff line
		@@ -182,6 +182,7 @@ class FlossMediaClient(BluetoothMediaCallbacks):
		self.bus = bus
		self.hci = hci
		self.objpath = self.MEDIA_OBJECT_PATTERN.format(hci)
		self.devices = []

		# We don't register callbacks by default.
		self.callbacks = None
		@@ -198,6 +199,9 @@ class FlossMediaClient(BluetoothMediaCallbacks):
		device: The struct of BluetoothAudioDevice.
		"""
		logging.debug('on_bluetooth_audio_device_added: device: %s', device)
		if device['address'] in self.devices:
		logging.debug("Device already added")
		self.devices.append(device['address'])

		@utils.glib_callback()
		def on_bluetooth_audio_device_removed(self, addr):
		@@ -207,6 +211,8 @@ class FlossMediaClient(BluetoothMediaCallbacks):
		addr: Address of device to be removed.
		"""
		logging.debug('on_bluetooth_audio_device_removed: address: %s', addr)
		if addr in self.devices:
		self.devices.remove(addr)

		@utils.glib_callback()
		def on_absolute_volume_supported_changed(self, supported):
		@@ -523,3 +529,23 @@ class FlossMediaClient(BluetoothMediaCallbacks):
		"""
		self.proxy().SetPlayerMetadata(metadata)
		return True

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

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

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

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