pstruebi/bumble_fork
1
0
Fork 0
forked from auracaster/bumble_mirror
Code Pull Requests Activity

improvements

Browse Source
This commit is contained in:
Gilles Boccon-Gibod
2025-02-03 17:58:09 -05:00
parent 55d3fd90f5
commit 70141c0439
22 changed files with 495 additions and 649 deletions
Download Patch File Download Diff File Expand all files Collapse all files

45
bumble/device.py
View File

@@ -17,6 +17,7 @@
# -----------------------------------------------------------------------------
from __future__ import annotations
import asyncio
import collections
from collections.abc import Iterable, Sequence
from contextlib import (
asynccontextmanager,
@@ -36,6 +37,7 @@ from typing import (
Any,
Callable,
ClassVar,
Deque,
Dict,
Optional,
Type,
@@ -1433,6 +1435,49 @@ class BisLink(_IsoLink):
self.device = self.big.device
# -----------------------------------------------------------------------------
class IsoPacketStream:
"""Async stream that can write SDUs to a CIS or BIS, with a maximum queue size."""
iso_link: _IsoLink
data_packet_queue: DataPacketQueue
def __init__(self, iso_link: _IsoLink, max_queue_size: int) -> None:
if iso_link.data_packet_queue is None:
raise ValueError('link has no data packet queue')
self.iso_link = iso_link
self.data_packet_queue = iso_link.data_packet_queue
self.data_packet_queue.on('flow', self._on_flow)
self._thresholds: Deque[int] = collections.deque()
self._semaphore = asyncio.Semaphore(max_queue_size)
def _on_flow(self) -> None:
# Release the semaphore once for each completed packet.
while (
self._thresholds and self.data_packet_queue.completed >= self._thresholds[0]
):
self._thresholds.popleft()
self._semaphore.release()
async def write(self, sdu: bytes) -> None:
"""
Write an SDU to the queue.
This method blocks until there are fewer than max_queue_size packets queued
but not yet completed.
"""
# Wait until there's space in the queue.
await self._semaphore.acquire()
# Queue the packet.
self.iso_link.write(sdu)
# Remember the position of the packet so we can know when it is completed.
self._thresholds.append(self.data_packet_queue.queued)
# -----------------------------------------------------------------------------
class Connection(CompositeEventEmitter):
device: Device

2
bumble/gatt.py
View File

@@ -323,8 +323,6 @@ GATT_SERVER_SUPPORTED_FEATURES_CHARACTERISTIC = UUID.from_16_bi
# -----------------------------------------------------------------------------
# Utils
# -----------------------------------------------------------------------------
def show_services(services: Iterable[Service]) -> None:
for service in services:
print(color(str(service), 'cyan'))

16
bumble/gatt_client.py
View File

@@ -78,6 +78,7 @@ from .gatt import (
GATT_INCLUDE_ATTRIBUTE_TYPE,
Characteristic,
ClientCharacteristicConfigurationBits,
InvalidServiceError,
TemplateService,
)
@@ -162,12 +163,23 @@ class ServiceProxy(AttributeProxy):
self.uuid = uuid
self.characteristics = []
async def discover_characteristics(self, uuids=()):
async def discover_characteristics(self, uuids=()) -> list[CharacteristicProxy]:
return await self.client.discover_characteristics(uuids, self)
def get_characteristics_by_uuid(self, uuid):
def get_characteristics_by_uuid(self, uuid: UUID) -> list[CharacteristicProxy]:
"""Get all the characteristics with a specified UUID."""
return self.client.get_characteristics_by_uuid(uuid, self)
def get_required_characteristic_by_uuid(self, uuid: UUID) -> CharacteristicProxy:
"""
Get the first characteristic with a specified UUID.
If no characteristic with that UUID is found, an InvalidServiceError is raised.
"""
if not (characteristics := self.get_characteristics_by_uuid(uuid)):
raise InvalidServiceError(f'{uuid} characteristic not found')
return characteristics[0]
def __str__(self) -> str:
return f'Service(handle=0x{self.handle:04X}, uuid={self.uuid})'

57
bumble/profiles/aics.py
View File

@@ -451,54 +451,35 @@ class AICSServiceProxy(ProfileServiceProxy):
def __init__(self, service_proxy: ServiceProxy) -> None:
self.service_proxy = service_proxy
if not (
characteristics := service_proxy.get_characteristics_by_uuid(
GATT_AUDIO_INPUT_STATE_CHARACTERISTIC
)
):
raise gatt.InvalidServiceError("Audio Input State Characteristic not found")
self.audio_input_state = SerializableCharacteristicAdapter(
characteristics[0], AudioInputState
service_proxy.get_required_characteristic_by_uuid(
GATT_AUDIO_INPUT_STATE_CHARACTERISTIC
),
AudioInputState,
)
if not (
characteristics := service_proxy.get_characteristics_by_uuid(
GATT_GAIN_SETTINGS_ATTRIBUTE_CHARACTERISTIC
)
):
raise gatt.InvalidServiceError(
"Gain Settings Attribute Characteristic not found"
)
self.gain_settings_properties = SerializableCharacteristicAdapter(
characteristics[0], GainSettingsProperties
service_proxy.get_required_characteristic_by_uuid(
GATT_GAIN_SETTINGS_ATTRIBUTE_CHARACTERISTIC
),
GainSettingsProperties,
)
if not (
characteristics := service_proxy.get_characteristics_by_uuid(
self.audio_input_status = PackedCharacteristicAdapter(
service_proxy.get_required_characteristic_by_uuid(
GATT_AUDIO_INPUT_STATUS_CHARACTERISTIC
)
):
raise gatt.InvalidServiceError(
"Audio Input Status Characteristic not found"
)
self.audio_input_status = PackedCharacteristicAdapter(characteristics[0], 'B')
),
'B',
)
if not (
characteristics := service_proxy.get_characteristics_by_uuid(
self.audio_input_control_point = (
service_proxy.get_required_characteristic_by_uuid(
GATT_AUDIO_INPUT_CONTROL_POINT_CHARACTERISTIC
)
):
raise gatt.InvalidServiceError(
"Audio Input Control Point Characteristic not found"
)
self.audio_input_control_point = characteristics[0]
)
if not (
characteristics := service_proxy.get_characteristics_by_uuid(
self.audio_input_description = UTF8CharacteristicAdapter(
service_proxy.get_required_characteristic_by_uuid(
GATT_AUDIO_INPUT_DESCRIPTION_CHARACTERISTIC
)
):
raise gatt.InvalidServiceError(
"Audio Input Description Characteristic not found"
)
self.audio_input_description = UTF8CharacteristicAdapter(characteristics[0])
)

10
bumble/profiles/asha.py
View File

@@ -288,8 +288,8 @@ class AshaServiceProxy(gatt_client.ProfileServiceProxy):
'psm_characteristic',
),
):
if not (
characteristics := self.service_proxy.get_characteristics_by_uuid(uuid)
):
raise gatt.InvalidServiceError(f"Missing {uuid} Characteristic")
setattr(self, attribute_name, characteristics[0])
setattr(
self,
attribute_name,
self.service_proxy.get_required_characteristic_by_uuid(uuid),
)

29
bumble/profiles/bass.py
View File

@@ -354,34 +354,25 @@ class BroadcastAudioScanServiceProxy(gatt_client.ProfileServiceProxy):
SERVICE_CLASS = BroadcastAudioScanService
broadcast_audio_scan_control_point: gatt_client.CharacteristicProxy
broadcast_receive_states: List[gatt.SerializableCharacteristicAdapter]
broadcast_receive_states: List[gatt.DelegatedCharacteristicAdapter]
def __init__(self, service_proxy: gatt_client.ServiceProxy):
self.service_proxy = service_proxy
if not (
characteristics := service_proxy.get_characteristics_by_uuid(
self.broadcast_audio_scan_control_point = (
service_proxy.get_required_characteristic_by_uuid(
gatt.GATT_BROADCAST_AUDIO_SCAN_CONTROL_POINT_CHARACTERISTIC
)
):
raise gatt.InvalidServiceError(
"Broadcast Audio Scan Control Point characteristic not found"
)
self.broadcast_audio_scan_control_point = characteristics[0]
)
if not (
characteristics := service_proxy.get_characteristics_by_uuid(
self.broadcast_receive_states = [
gatt.DelegatedCharacteristicAdapter(
characteristic,
decode=lambda x: BroadcastReceiveState.from_bytes(x) if x else None,
)
for characteristic in service_proxy.get_characteristics_by_uuid(
gatt.GATT_BROADCAST_RECEIVE_STATE_CHARACTERISTIC
)
):
raise gatt.InvalidServiceError(
"Broadcast Receive State characteristic not found"
)
self.broadcast_receive_states = [
gatt.SerializableCharacteristicAdapter(
characteristic, BroadcastReceiveState
)
for characteristic in characteristics
]
async def send_control_point_operation(

11
bumble/profiles/gmap.py
View File

@@ -30,7 +30,6 @@ from bumble.gatt import (
GATT_UGT_FEATURES_CHARACTERISTIC,
GATT_BGS_FEATURES_CHARACTERISTIC,
GATT_BGR_FEATURES_CHARACTERISTIC,
InvalidServiceError,
)
from bumble.gatt_client import ProfileServiceProxy, ServiceProxy
from enum import IntFlag
@@ -154,14 +153,10 @@ class GamingAudioServiceProxy(ProfileServiceProxy):
def __init__(self, service_proxy: ServiceProxy) -> None:
self.service_proxy = service_proxy
if not (
characteristics := service_proxy.get_characteristics_by_uuid(
GATT_GMAP_ROLE_CHARACTERISTIC
)
):
raise InvalidServiceError("GMAP Role Characteristic not found")
self.gmap_role = DelegatedCharacteristicAdapter(
characteristic=characteristics[0],
service_proxy.get_required_characteristic_by_uuid(
GATT_GMAP_ROLE_CHARACTERISTIC
),
decode=lambda value: GmapRole(value[0]),
)

31
bumble/profiles/le_audio.py
View File

@@ -17,6 +17,7 @@
# -----------------------------------------------------------------------------
from __future__ import annotations
import dataclasses
import enum
import struct
from typing import Any, List, Type
from typing_extensions import Self
@@ -41,9 +42,10 @@ class AssistedListeningStream(utils.OpenIntEnum):
class Metadata:
'''Bluetooth Assigned Numbers, Section 6.12.6 - Metadata LTV structures.
As Metadata fields may extend, and Spec doesn't forbid duplication, we don't
automatically parse the Metadata data into specific classes. Callers may decode
the data by themselves, or use the Entry.decode method.
As Metadata fields may extend, and the spec may not guarantee the uniqueness of
tags, we don't automatically parse the Metadata data into specific classes.
Users of this class may decode the data by themselves, or use the Entry.decode
method.
'''
class Tag(utils.OpenIntEnum):
@@ -114,6 +116,29 @@ class Metadata:
entries: List[Entry] = dataclasses.field(default_factory=list)
def pretty_print(self, indent: str) -> str:
"""Convenience method to generate a string with one key-value pair per line."""
max_key_length = 0
keys = []
values = []
for entry in self.entries:
key = entry.tag.name
max_key_length = max(max_key_length, len(key))
keys.append(key)
decoded = entry.decode()
if isinstance(decoded, enum.Enum):
values.append(decoded.name)
elif isinstance(decoded, bytes):
values.append(decoded.hex())
else:
values.append(str(decoded))
return '\n'.join(
f'{indent}{key}: {" " * (max_key_length-len(key))}{value}'
for key, value in zip(keys, values)
)
@classmethod
def from_bytes(cls: Type[Self], data: bytes) -> Self:
entries = []

64
bumble/profiles/pacs.py
View File

@@ -72,6 +72,19 @@ class PacRecord:
metadata=metadata,
)
@classmethod
def list_from_bytes(cls, data: bytes) -> list[PacRecord]:
"""Parse a serialized list of records preceded by a one byte list length."""
record_count = data[0]
records = []
offset = 1
for _ in range(record_count):
record = PacRecord.from_bytes(data[offset:])
offset += len(bytes(record))
records.append(record)
return records
def __bytes__(self) -> bytes:
capabilities_bytes = bytes(self.codec_specific_capabilities)
metadata_bytes = bytes(self.metadata)
@@ -172,39 +185,58 @@ class PublishedAudioCapabilitiesService(gatt.TemplateService):
class PublishedAudioCapabilitiesServiceProxy(gatt_client.ProfileServiceProxy):
SERVICE_CLASS = PublishedAudioCapabilitiesService
sink_pac: Optional[gatt_client.CharacteristicProxy] = None
sink_audio_locations: Optional[gatt_client.CharacteristicProxy] = None
source_pac: Optional[gatt_client.CharacteristicProxy] = None
source_audio_locations: Optional[gatt_client.CharacteristicProxy] = None
available_audio_contexts: gatt_client.CharacteristicProxy
supported_audio_contexts: gatt_client.CharacteristicProxy
sink_pac: Optional[gatt.DelegatedCharacteristicAdapter] = None
sink_audio_locations: Optional[gatt.DelegatedCharacteristicAdapter] = None
source_pac: Optional[gatt.DelegatedCharacteristicAdapter] = None
source_audio_locations: Optional[gatt.DelegatedCharacteristicAdapter] = None
available_audio_contexts: gatt.DelegatedCharacteristicAdapter
supported_audio_contexts: gatt.DelegatedCharacteristicAdapter
def __init__(self, service_proxy: gatt_client.ServiceProxy):
self.service_proxy = service_proxy
self.available_audio_contexts = service_proxy.get_characteristics_by_uuid(
gatt.GATT_AVAILABLE_AUDIO_CONTEXTS_CHARACTERISTIC
)[0]
self.supported_audio_contexts = service_proxy.get_characteristics_by_uuid(
gatt.GATT_SUPPORTED_AUDIO_CONTEXTS_CHARACTERISTIC
)[0]
self.available_audio_contexts = gatt.DelegatedCharacteristicAdapter(
service_proxy.get_required_characteristic_by_uuid(
gatt.GATT_AVAILABLE_AUDIO_CONTEXTS_CHARACTERISTIC
),
decode=lambda x: tuple(map(ContextType, struct.unpack('<HH', x))),
)
self.supported_audio_contexts = gatt.DelegatedCharacteristicAdapter(
service_proxy.get_required_characteristic_by_uuid(
gatt.GATT_SUPPORTED_AUDIO_CONTEXTS_CHARACTERISTIC
),
decode=lambda x: tuple(map(ContextType, struct.unpack('<HH', x))),
)
if characteristics := service_proxy.get_characteristics_by_uuid(
gatt.GATT_SINK_PAC_CHARACTERISTIC
):
self.sink_pac = characteristics[0]
self.sink_pac = gatt.DelegatedCharacteristicAdapter(
characteristics[0],
decode=PacRecord.list_from_bytes,
)
if characteristics := service_proxy.get_characteristics_by_uuid(
gatt.GATT_SOURCE_PAC_CHARACTERISTIC
):
self.source_pac = characteristics[0]
self.source_pac = gatt.DelegatedCharacteristicAdapter(
characteristics[0],
decode=PacRecord.list_from_bytes,
)
if characteristics := service_proxy.get_characteristics_by_uuid(
gatt.GATT_SINK_AUDIO_LOCATION_CHARACTERISTIC
):
self.sink_audio_locations = characteristics[0]
self.sink_audio_locations = gatt.DelegatedCharacteristicAdapter(
characteristics[0],
decode=lambda x: AudioLocation(struct.unpack('<I', x)[0]),
)
if characteristics := service_proxy.get_characteristics_by_uuid(
gatt.GATT_SOURCE_AUDIO_LOCATION_CHARACTERISTIC
):
self.source_audio_locations = characteristics[0]
self.source_audio_locations = gatt.DelegatedCharacteristicAdapter(
characteristics[0],
decode=lambda x: AudioLocation(struct.unpack('<I', x)[0]),
)

12
bumble/profiles/tmap.py
View File

@@ -25,7 +25,6 @@ from bumble.gatt import (
TemplateService,
Characteristic,
DelegatedCharacteristicAdapter,
InvalidServiceError,
GATT_TELEPHONY_AND_MEDIA_AUDIO_SERVICE,
GATT_TMAP_ROLE_CHARACTERISTIC,
)
@@ -74,15 +73,10 @@ class TelephonyAndMediaAudioServiceProxy(ProfileServiceProxy):
def __init__(self, service_proxy: ServiceProxy):
self.service_proxy = service_proxy
if not (
characteristics := service_proxy.get_characteristics_by_uuid(
GATT_TMAP_ROLE_CHARACTERISTIC
)
):
raise InvalidServiceError('TMAP Role characteristic not found')
self.role = DelegatedCharacteristicAdapter(
characteristics[0],
service_proxy.get_required_characteristic_by_uuid(
GATT_TMAP_ROLE_CHARACTERISTIC
),
decode=lambda value: Role(
struct.unpack_from('<H', value, 0)[0],
),

65
bumble/profiles/vcp.py → bumble/profiles/vcs.py
View File

@@ -1,4 +1,4 @@
# Copyright 2021-2024 Google LLC
# Copyright 2021-2025 Google LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
@@ -17,14 +17,16 @@
# Imports
# -----------------------------------------------------------------------------
from __future__ import annotations
import dataclasses
import enum
from typing import Optional, Sequence
from bumble import att
from bumble import device
from bumble import gatt
from bumble import gatt_client
from typing import Optional, Sequence
# -----------------------------------------------------------------------------
# Constants
@@ -67,6 +69,20 @@ class VolumeControlPointOpcode(enum.IntEnum):
MUTE = 0x06
@dataclasses.dataclass
class VolumeState:
volume_setting: int
mute: int
change_counter: int
@classmethod
def from_bytes(cls, data: bytes) -> VolumeState:
return cls(data[0], data[1], data[2])
def __bytes__(self) -> bytes:
return bytes([self.volume_setting, self.mute, self.change_counter])
# -----------------------------------------------------------------------------
# Server
# -----------------------------------------------------------------------------
@@ -126,16 +142,8 @@ class VolumeControlService(gatt.TemplateService):
included_services=list(included_services),
)
@property
def volume_state_bytes(self) -> bytes:
return bytes([self.volume_setting, self.muted, self.change_counter])
@volume_state_bytes.setter
def volume_state_bytes(self, new_value: bytes) -> None:
self.volume_setting, self.muted, self.change_counter = new_value
def _on_read_volume_state(self, _connection: Optional[device.Connection]) -> bytes:
return self.volume_state_bytes
return bytes(VolumeState(self.volume_setting, self.muted, self.change_counter))
def _on_write_volume_control_point(
self, connection: Optional[device.Connection], value: bytes
@@ -153,14 +161,9 @@ class VolumeControlService(gatt.TemplateService):
self.change_counter = (self.change_counter + 1) % 256
connection.abort_on(
'disconnection',
connection.device.notify_subscribers(
attribute=self.volume_state,
value=self.volume_state_bytes,
),
)
self.emit(
'volume_state', self.volume_setting, self.muted, self.change_counter
connection.device.notify_subscribers(attribute=self.volume_state),
)
self.emit('volume_state_change')
def _on_relative_volume_down(self) -> bool:
old_volume = self.volume_setting
@@ -207,24 +210,26 @@ class VolumeControlServiceProxy(gatt_client.ProfileServiceProxy):
SERVICE_CLASS = VolumeControlService
volume_control_point: gatt_client.CharacteristicProxy
volume_state: gatt.SerializableCharacteristicAdapter
volume_flags: gatt.DelegatedCharacteristicAdapter
def __init__(self, service_proxy: gatt_client.ServiceProxy) -> None:
self.service_proxy = service_proxy
self.volume_state = gatt.PackedCharacteristicAdapter(
service_proxy.get_characteristics_by_uuid(
self.volume_state = gatt.SerializableCharacteristicAdapter(
service_proxy.get_required_characteristic_by_uuid(
gatt.GATT_VOLUME_STATE_CHARACTERISTIC
)[0],
'BBB',
),
VolumeState,
)
self.volume_control_point = service_proxy.get_characteristics_by_uuid(
self.volume_control_point = service_proxy.get_required_characteristic_by_uuid(
gatt.GATT_VOLUME_CONTROL_POINT_CHARACTERISTIC
)[0]
self.volume_flags = gatt.PackedCharacteristicAdapter(
service_proxy.get_characteristics_by_uuid(
gatt.GATT_VOLUME_FLAGS_CHARACTERISTIC
)[0],
'B',
)
self.volume_flags = gatt.DelegatedCharacteristicAdapter(
service_proxy.get_required_characteristic_by_uuid(
gatt.GATT_VOLUME_FLAGS_CHARACTERISTIC
),
decode=lambda data: VolumeFlags(data[0]),
)

46
bumble/profiles/vocs.py
View File

@@ -27,8 +27,8 @@ from bumble.gatt import (
DelegatedCharacteristicAdapter,
TemplateService,
CharacteristicValue,
SerializableCharacteristicAdapter,
UTF8CharacteristicAdapter,
InvalidServiceError,
GATT_VOLUME_OFFSET_CONTROL_SERVICE,
GATT_VOLUME_OFFSET_STATE_CHARACTERISTIC,
GATT_AUDIO_LOCATION_CHARACTERISTIC,
@@ -287,44 +287,28 @@ class VolumeOffsetControlServiceProxy(ProfileServiceProxy):
def __init__(self, service_proxy: ServiceProxy) -> None:
self.service_proxy = service_proxy
if not (
characteristics := service_proxy.get_characteristics_by_uuid(
GATT_VOLUME_OFFSET_STATE_CHARACTERISTIC
)
):
raise InvalidServiceError("Volume Offset State characteristic not found")
self.volume_offset_state = DelegatedCharacteristicAdapter(
characteristics[0], decode=VolumeOffsetState.from_bytes
service_proxy.get_required_characteristic_by_uuid(
GATT_VOLUME_OFFSET_STATE_CHARACTERISTIC
),
decode=VolumeOffsetState.from_bytes,
)
if not (
characteristics := service_proxy.get_characteristics_by_uuid(
self.audio_location = SerializableCharacteristicAdapter(
service_proxy.get_required_characteristic_by_uuid(
GATT_AUDIO_LOCATION_CHARACTERISTIC
)
):
raise InvalidServiceError("Audio Location characteristic not found")
self.audio_location = DelegatedCharacteristicAdapter(
characteristics[0],
encode=lambda value: bytes(value),
decode=VocsAudioLocation.from_bytes,
),
VocsAudioLocation,
)
if not (
characteristics := service_proxy.get_characteristics_by_uuid(
self.volume_offset_control_point = (
service_proxy.get_required_characteristic_by_uuid(
GATT_VOLUME_OFFSET_CONTROL_POINT_CHARACTERISTIC
)
):
raise InvalidServiceError(
"Volume Offset Control Point characteristic not found"
)
self.volume_offset_control_point = characteristics[0]
)
if not (
characteristics := service_proxy.get_characteristics_by_uuid(
self.audio_output_description = UTF8CharacteristicAdapter(
service_proxy.get_required_characteristic_by_uuid(
GATT_AUDIO_OUTPUT_DESCRIPTION_CHARACTERISTIC
)
):
raise InvalidServiceError(
"Audio Output Description characteristic not found"
)
self.audio_output_description = UTF8CharacteristicAdapter(characteristics[0])
)