Compare commits

...

52 Commits

Author SHA1 Message Date
Gilles Boccon-Gibod
69c6643bb8 Merge pull request #452 from marshallpierce/mp/rust-0.2.0
Bumble crate 0.2.0
2024-03-21 17:15:43 -07:00
Marshall Pierce
b8214bf948 Bumble crate 0.2.0 2024-03-21 12:36:32 -06:00
Charlie Boutier
a9c62c44b3 pandora host: change AdvertisingType
change advertising type from high duty to low duty

Test: python le_host_test.py -c config.yml --test_bed android.bumbles --tests "test_scan('connectable','non_scannable','directed',0)" -v
2024-03-20 11:17:50 -07:00
Charlie Boutier
7d0b4ef4e0 pandora_server: Parse FLAGS into advertising data
Bug: 328089785
2024-03-18 09:20:55 -07:00
Charlie Boutier
313340f1c6 intel driver: check the vendorId and productId 2024-03-15 10:53:33 -07:00
Charlie Boutier
e8ed69fb09 pyusb: Collect vendorId and productId as metadata 2024-03-15 10:53:33 -07:00
David Duarte
16d5cf6770 usb: Add usb path moniker
Add a new moniker for usb and pyusb driver allowing
to select the usb device using its bus id and port
path like `usb:3-3.4.1`.
2024-03-15 09:17:39 -07:00
Gilles Boccon-Gibod
a2caf1deb2 Merge pull request #448 from BenjaminLawson/bump-avatar
Bump pandora-avatar to 0.0.8
2024-03-14 20:49:28 -07:00
Ben Lawson
01bfdd2c98 Bump pandora-avater to 0.0.8 2024-03-14 14:13:27 -07:00
Gilles Boccon-Gibod
4a60df108a Merge pull request #447 from BenjaminLawson/bump-rootcanal
Bump rootcanal to 1.9.0
2024-03-14 14:00:36 -07:00
Ben Lawson
ad48109748 Bump rootcanal to 1.9.0 2024-03-14 13:15:02 -07:00
Gilles Boccon-Gibod
44c51c13ac Merge pull request #445 from google/gbg/driver-probe-fix
fix intel driver probe
2024-03-12 12:51:08 -07:00
Gilles Boccon-Gibod
7507be1eab update metadata when setting the host controller directly 2024-03-12 11:50:47 -07:00
Gilles Boccon-Gibod
cbe9446dcf fix intel driver probe 2024-03-12 09:54:20 -07:00
Charlie Boutier
174930399a intel: send vsc INTEL_DDC_CONFIG_WRITE
This VSC enable host-initiated role-switching after connection.

Implement this VSC in a driver fashion.

Test: avatar security_test with the Bluetooth Dongle Intel BE200
2024-03-11 09:15:18 -07:00
Gilles Boccon-Gibod
1f3aee5566 Merge pull request #438 from BenjaminLawson/pandora-extended-advertising
Implement Pandora extended advertising
2024-03-07 20:36:56 -08:00
Ben Lawson
256044a789 Implement Pandora extended advertising
Support setting the PHY of Pandora scans.
2024-03-07 16:18:49 -08:00
Gilles Boccon-Gibod
e554bd1033 Merge pull request #434 from google/gbg/show-timestamps
show timestamps from snoop logs
2024-02-29 11:44:23 -08:00
Gilles Boccon-Gibod
38981cefa1 pad index field 2024-02-28 11:46:35 -08:00
Gilles Boccon-Gibod
f2d601f411 show timestamps from snoop logs 2024-02-27 16:40:37 -08:00
zxzxwu
6e7c64c1de Merge pull request #431 from zxzxwu/rust
Bump Rust to 1.76.0
2024-02-23 15:14:30 +08:00
Josh Wu
565d51f4db Bump Rust to 1.76.0
```
error: failed to compile `cargo-all-features v1.10.0`, intermediate artifacts can be found at `/tmp/cargo-installshCmAG`

Caused by:
  package `clap v4.5.1` cannot be built because it requires rustc 1.74 or newer, while the currently active rustc version is 1.70.0
  Try re-running cargo install with `--locked`

```
2024-02-22 15:22:20 +08:00
Gilles Boccon-Gibod
de8f3d9c1e Merge pull request #426 from akuker/patch-1
Add clarification to short circuit list feature
2024-02-12 21:22:14 -08:00
Tony Kuker
cde6d48690 Add clarification to short circuit list feature 2024-02-12 12:22:36 -06:00
zxzxwu
02180088b3 Merge pull request #425 from zxzxwu/command
Refactor command supporting list
2024-02-07 21:45:52 +08:00
zxzxwu
90f49267d1 Merge pull request #424 from zxzxwu/adv
Fix double-disable legacy advertising set
2024-02-06 16:13:51 +08:00
Josh Wu
0e6d69cd7b Refactor command supporting list 2024-02-06 12:06:00 +08:00
Josh Wu
9eccc583d5 Fix double-disable legacy advertising set
When legacy advertising set is disabled passively(by set termination),
the legacy advertising set won't be released, and the next
stop_advertising() call will try to disable it again and cause an error.
2024-02-06 12:00:30 +08:00
Gilles Boccon-Gibod
f4aeaa6eb3 Merge pull request #422 from google/gbg/bench-rfcomm-params
add rfcomm options and fix l2cap mtu negotiation
2024-02-05 09:14:16 -08:00
Gilles Boccon-Gibod
d7489a644a update websockets version (for better typecheck) 2024-02-05 09:07:39 -08:00
Gilles Boccon-Gibod
a877283360 add rfcomm options and fix l2cap mtu negotiation 2024-02-05 08:56:59 -08:00
zxzxwu
6d91e7e79b Merge pull request #423 from zxzxwu/vcp
Fix Lint error in VCP example
2024-02-06 00:40:05 +08:00
Josh Wu
567146b143 Fix Lint error in VCP example 2024-02-04 21:23:22 +08:00
zxzxwu
1a3272d7ca Merge pull request #412 from zxzxwu/vcp
Add Volume Control Service
2024-02-04 00:42:51 +08:00
zxzxwu
1ee1ff0b62 Merge pull request #420 from zxzxwu/rfc
Add RFCOMM and SDP context manager and search helper
2024-02-04 00:42:24 +08:00
zxzxwu
729fd97748 Merge pull request #419 from zxzxwu/feat
Add local LMP feature reader
2024-02-03 13:51:19 +08:00
Josh Wu
e308051885 Add LMP feature reader 2024-02-03 13:29:25 +08:00
Josh Wu
10e53553d7 Add RFCOMM and SDP helpers 2024-02-03 13:13:35 +08:00
Gilles Boccon-Gibod
ef0b30d059 Merge pull request #382 from google/gbg/extended-advertising-v2
extended advertising v2
2024-02-02 20:43:28 -08:00
Gilles Boccon-Gibod
e7e9f9509a update rootcanal version 2024-02-02 20:33:19 -08:00
zxzxwu
c6cfd101df Merge pull request #415 from zxzxwu/hfp
HFP: State memory and event emission
2024-02-02 11:36:53 +08:00
Josh Wu
d2dcf063ee HFP: State memory and event emit 2024-02-01 12:08:43 +08:00
Michael Mogenson
d15bc7d664 Merge pull request #417 from mogenson/controller-loopback-cid-range
controller_loopback: LE support and max packet count
2024-01-31 21:13:21 -05:00
Michael Mogenson
2a764fd6bb controller_loopback: LE support and max packet count
Bound the packet count CLI option. We're using the L2CAP header CID for
a paket ID, so the max packet count value has to fit into this 16-bit
field.

Add support for controllers that are LE only by checking the
le_acl_packet_queue.max_size.

Tested with 65535 max packet count. Took 138 seconds at 481 kB/s with a
USB BT dongle.
2024-01-31 10:26:51 -05:00
Josh Wu
3e8ce38eba Add Volume Control Service 2024-01-31 10:04:30 +08:00
Gilles Boccon-Gibod
8d2f37aa7a inclusive language 2024-01-28 19:09:39 -08:00
Gilles Boccon-Gibod
b7b70ebcbb address PR comments 2024-01-28 19:09:37 -08:00
Gilles Boccon-Gibod
8ba91f4986 fix assert 2024-01-28 19:02:32 -08:00
Gilles Boccon-Gibod
79a5e953bc comply with limits for certain advertising event types 2024-01-28 19:02:32 -08:00
Gilles Boccon-Gibod
20de5ea250 format 2024-01-28 19:02:32 -08:00
Gilles Boccon-Gibod
bad9ce272c add doc 2024-01-28 19:02:32 -08:00
Gilles Boccon-Gibod
d3273ffa8c format (+3 squashed commits)
Squashed commits:
[60e610f] wip
[eeab73d] wip
[3cdd5b8] basic first pass
2024-01-28 19:02:30 -08:00
59 changed files with 3467 additions and 1503 deletions

View File

@@ -47,7 +47,7 @@ jobs:
strategy:
matrix:
python-version: [ "3.8", "3.9", "3.10", "3.11" ]
rust-version: [ "1.70.0", "stable" ]
rust-version: [ "1.76.0", "stable" ]
fail-fast: false
steps:
- name: Check out from Git

3
.gitignore vendored
View File

@@ -9,6 +9,9 @@ __pycache__
# generated by setuptools_scm
bumble/_version.py
.vscode/launch.json
.vscode/settings.json
/.idea
venv/
.venv/
# snoop logs
out/

View File

@@ -74,6 +74,8 @@
"substates",
"tobytes",
"tsep",
"UNMUTE",
"unmuted",
"usbmodem",
"vhci",
"websockets",

View File

@@ -50,10 +50,8 @@ from bumble.sdp import (
SDP_PUBLIC_BROWSE_ROOT,
SDP_SERVICE_CLASS_ID_LIST_ATTRIBUTE_ID,
SDP_SERVICE_RECORD_HANDLE_ATTRIBUTE_ID,
SDP_BLUETOOTH_PROFILE_DESCRIPTOR_LIST_ATTRIBUTE_ID,
DataElement,
ServiceAttribute,
Client as SdpClient,
)
from bumble.transport import open_transport_or_link
import bumble.rfcomm
@@ -89,6 +87,7 @@ DEFAULT_LINGER_TIME = 1.0
DEFAULT_POST_CONNECTION_WAIT_TIME = 1.0
DEFAULT_RFCOMM_CHANNEL = 8
DEFAULT_RFCOMM_MTU = 2048
# -----------------------------------------------------------------------------
@@ -198,48 +197,6 @@ def make_sdp_records(channel):
}
async def find_rfcomm_channel_with_uuid(connection: Connection, uuid: str) -> int:
# Connect to the SDP Server
sdp_client = SdpClient(connection)
await sdp_client.connect()
# Search for services with an L2CAP service attribute
search_result = await sdp_client.search_attributes(
[BT_L2CAP_PROTOCOL_ID],
[
SDP_PROTOCOL_DESCRIPTOR_LIST_ATTRIBUTE_ID,
SDP_BLUETOOTH_PROFILE_DESCRIPTOR_LIST_ATTRIBUTE_ID,
SDP_SERVICE_CLASS_ID_LIST_ATTRIBUTE_ID,
],
)
for attribute_list in search_result:
service_uuid = None
service_class_id_list = ServiceAttribute.find_attribute_in_list(
attribute_list, SDP_SERVICE_CLASS_ID_LIST_ATTRIBUTE_ID
)
if service_class_id_list:
if service_class_id_list.value:
for service_class_id in service_class_id_list.value:
service_uuid = service_class_id.value
if str(service_uuid) != uuid:
# This service doesn't have a UUID or isn't the right one.
continue
# Look for the RFCOMM Channel number
protocol_descriptor_list = ServiceAttribute.find_attribute_in_list(
attribute_list, SDP_PROTOCOL_DESCRIPTOR_LIST_ATTRIBUTE_ID
)
if protocol_descriptor_list:
for protocol_descriptor in protocol_descriptor_list.value:
if len(protocol_descriptor.value) >= 2:
if protocol_descriptor.value[0].value == BT_RFCOMM_PROTOCOL_ID:
await sdp_client.disconnect()
return protocol_descriptor.value[1].value
await sdp_client.disconnect()
return 0
def log_stats(title, stats):
stats_min = min(stats)
stats_max = max(stats)
@@ -940,11 +897,14 @@ class L2capServer(StreamedPacketIO):
# RfcommClient
# -----------------------------------------------------------------------------
class RfcommClient(StreamedPacketIO):
def __init__(self, device, channel, uuid):
def __init__(self, device, channel, uuid, l2cap_mtu, max_frame_size, window_size):
super().__init__()
self.device = device
self.channel = channel
self.uuid = uuid
self.l2cap_mtu = l2cap_mtu
self.max_frame_size = max_frame_size
self.window_size = window_size
self.rfcomm_session = None
self.ready = asyncio.Event()
@@ -957,7 +917,9 @@ class RfcommClient(StreamedPacketIO):
logging.info(
color(f'@@@ Discovering channel number from UUID {self.uuid}', 'cyan')
)
channel = await find_rfcomm_channel_with_uuid(connection, self.uuid)
channel = await bumble.rfcomm.find_rfcomm_channel_with_uuid(
connection, self.uuid
)
logging.info(color(f'@@@ Channel number = {channel}', 'cyan'))
if channel == 0:
logging.info(color('!!! No RFComm service with this UUID found', 'red'))
@@ -966,13 +928,21 @@ class RfcommClient(StreamedPacketIO):
# Create a client and start it
logging.info(color('*** Starting RFCOMM client...', 'blue'))
rfcomm_client = bumble.rfcomm.Client(connection)
rfcomm_options = {}
if self.l2cap_mtu:
rfcomm_options['l2cap_mtu'] = self.l2cap_mtu
rfcomm_client = bumble.rfcomm.Client(connection, **rfcomm_options)
rfcomm_mux = await rfcomm_client.start()
logging.info(color('*** Started', 'blue'))
logging.info(color(f'### Opening session for channel {channel}...', 'yellow'))
try:
rfcomm_session = await rfcomm_mux.open_dlc(channel)
dlc_options = {}
if self.max_frame_size:
dlc_options['max_frame_size'] = self.max_frame_size
if self.window_size:
dlc_options['window_size'] = self.window_size
rfcomm_session = await rfcomm_mux.open_dlc(channel, **dlc_options)
logging.info(color(f'### Session open: {rfcomm_session}', 'yellow'))
except bumble.core.ConnectionError as error:
logging.info(color(f'!!! Session open failed: {error}', 'red'))
@@ -997,13 +967,16 @@ class RfcommClient(StreamedPacketIO):
# RfcommServer
# -----------------------------------------------------------------------------
class RfcommServer(StreamedPacketIO):
def __init__(self, device, channel):
def __init__(self, device, channel, l2cap_mtu):
super().__init__()
self.dlc = None
self.ready = asyncio.Event()
# Create and register a server
rfcomm_server = bumble.rfcomm.Server(device)
server_options = {}
if l2cap_mtu:
server_options['l2cap_mtu'] = l2cap_mtu
rfcomm_server = bumble.rfcomm.Server(device, **server_options)
# Listen for incoming DLC connections
channel_number = rfcomm_server.listen(self.on_dlc, channel)
@@ -1340,11 +1313,20 @@ def create_mode_factory(ctx, default_mode):
if mode == 'rfcomm-client':
return RfcommClient(
device, channel=ctx.obj['rfcomm_channel'], uuid=ctx.obj['rfcomm_uuid']
device,
channel=ctx.obj['rfcomm_channel'],
uuid=ctx.obj['rfcomm_uuid'],
l2cap_mtu=ctx.obj['rfcomm_l2cap_mtu'],
max_frame_size=ctx.obj['rfcomm_max_frame_size'],
window_size=ctx.obj['rfcomm_window_size'],
)
if mode == 'rfcomm-server':
return RfcommServer(device, channel=ctx.obj['rfcomm_channel'])
return RfcommServer(
device,
channel=ctx.obj['rfcomm_channel'],
l2cap_mtu=ctx.obj['rfcomm_l2cap_mtu'],
)
raise ValueError('invalid mode')
@@ -1431,6 +1413,21 @@ def create_role_factory(ctx, default_role):
default=DEFAULT_RFCOMM_UUID,
help='RFComm service UUID to use (ignored if --rfcomm-channel is not 0)',
)
@click.option(
'--rfcomm-l2cap-mtu',
type=int,
help='RFComm L2CAP MTU',
)
@click.option(
'--rfcomm-max-frame-size',
type=int,
help='RFComm maximum frame size',
)
@click.option(
'--rfcomm-window-size',
type=int,
help='RFComm window size',
)
@click.option(
'--l2cap-psm',
type=int,
@@ -1528,6 +1525,9 @@ def bench(
linger,
rfcomm_channel,
rfcomm_uuid,
rfcomm_l2cap_mtu,
rfcomm_max_frame_size,
rfcomm_window_size,
l2cap_psm,
l2cap_mtu,
l2cap_mps,
@@ -1540,6 +1540,9 @@ def bench(
ctx.obj['att_mtu'] = att_mtu
ctx.obj['rfcomm_channel'] = rfcomm_channel
ctx.obj['rfcomm_uuid'] = rfcomm_uuid
ctx.obj['rfcomm_l2cap_mtu'] = rfcomm_l2cap_mtu
ctx.obj['rfcomm_max_frame_size'] = rfcomm_max_frame_size
ctx.obj['rfcomm_window_size'] = rfcomm_window_size
ctx.obj['l2cap_psm'] = l2cap_psm
ctx.obj['l2cap_mtu'] = l2cap_mtu
ctx.obj['l2cap_mps'] = l2cap_mps

View File

@@ -99,7 +99,12 @@ class Loopback:
# make sure data can fit in one l2cap pdu
l2cap_header_size = 4
max_packet_size = host.acl_packet_queue.max_packet_size - l2cap_header_size
max_packet_size = (
host.acl_packet_queue
if host.acl_packet_queue
else host.le_acl_packet_queue
).max_packet_size - l2cap_header_size
if self.packet_size > max_packet_size:
print(
color(
@@ -183,7 +188,7 @@ class Loopback:
'--packet-count',
'-c',
metavar='COUNT',
type=int,
type=click.IntRange(1, 65535),
default=10,
help='Packet count',
)

View File

@@ -15,7 +15,11 @@
# -----------------------------------------------------------------------------
# Imports
# -----------------------------------------------------------------------------
import datetime
import logging
import os
import struct
import click
from bumble.colors import color
@@ -24,6 +28,14 @@ from bumble.transport.common import PacketReader
from bumble.helpers import PacketTracer
# -----------------------------------------------------------------------------
# Logging
# -----------------------------------------------------------------------------
logger = logging.getLogger(__name__)
# -----------------------------------------------------------------------------
# Classes
# -----------------------------------------------------------------------------
class SnoopPacketReader:
'''
@@ -36,12 +48,18 @@ class SnoopPacketReader:
DATALINK_BSCP = 1003
DATALINK_H5 = 1004
IDENTIFICATION_PATTERN = b'btsnoop\0'
TIMESTAMP_ANCHOR = datetime.datetime(2000, 1, 1)
TIMESTAMP_DELTA = 0x00E03AB44A676000
ONE_MICROSECOND = datetime.timedelta(microseconds=1)
def __init__(self, source):
self.source = source
self.at_end = False
# Read the header
identification_pattern = source.read(8)
if identification_pattern.hex().lower() != '6274736e6f6f7000':
if identification_pattern != self.IDENTIFICATION_PATTERN:
raise ValueError(
'not a valid snoop file, unexpected identification pattern'
)
@@ -55,19 +73,32 @@ class SnoopPacketReader:
# Read the record header
header = self.source.read(24)
if len(header) < 24:
return (0, None)
self.at_end = True
return (None, 0, None)
# Parse the header
(
original_length,
included_length,
packet_flags,
_cumulative_drops,
_timestamp_seconds,
_timestamp_microsecond,
) = struct.unpack('>IIIIII', header)
timestamp,
) = struct.unpack('>IIIIQ', header)
# Abort on truncated packets
# Skip truncated packets
if original_length != included_length:
return (0, None)
print(
color(
f"!!! truncated packet ({included_length}/{original_length})", "red"
)
)
self.source.read(included_length)
return (None, 0, None)
# Convert the timestamp to a datetime object.
ts_dt = self.TIMESTAMP_ANCHOR + datetime.timedelta(
microseconds=timestamp - self.TIMESTAMP_DELTA
)
if self.data_link_type == self.DATALINK_H1:
# The packet is un-encapsulated, look at the flags to figure out its type
@@ -89,7 +120,17 @@ class SnoopPacketReader:
bytes([packet_type]) + self.source.read(included_length),
)
return (packet_flags & 1, self.source.read(included_length))
return (ts_dt, packet_flags & 1, self.source.read(included_length))
# -----------------------------------------------------------------------------
class Printer:
def __init__(self):
self.index = 0
def print(self, message: str) -> None:
self.index += 1
print(f"[{self.index:8}]{message}")
# -----------------------------------------------------------------------------
@@ -122,24 +163,28 @@ def main(format, vendors, filename):
packet_reader = PacketReader(input)
def read_next_packet():
return (0, packet_reader.next_packet())
return (None, 0, packet_reader.next_packet())
else:
packet_reader = SnoopPacketReader(input)
read_next_packet = packet_reader.next_packet
tracer = PacketTracer(emit_message=print)
printer = Printer()
tracer = PacketTracer(emit_message=printer.print)
while True:
while not packet_reader.at_end:
try:
(direction, packet) = read_next_packet()
if packet is None:
break
tracer.trace(hci.HCI_Packet.from_bytes(packet), direction)
(timestamp, direction, packet) = read_next_packet()
if packet:
tracer.trace(hci.HCI_Packet.from_bytes(packet), direction, timestamp)
else:
printer.print(color("[TRUNCATED]", "red"))
except Exception as error:
logger.exception()
print(color(f'!!! {error}', 'red'))
# -----------------------------------------------------------------------------
if __name__ == '__main__':
logging.basicConfig(level=os.environ.get('BUMBLE_LOGLEVEL', 'WARNING').upper())
main() # pylint: disable=no-value-for-parameter

View File

@@ -1470,10 +1470,10 @@ class Protocol(EventEmitter):
f'[{transaction_label}] {message}'
)
max_fragment_size = (
self.l2cap_channel.mtu - 3
self.l2cap_channel.peer_mtu - 3
) # Enough space for a 3-byte start packet header
payload = message.payload
if len(payload) + 2 <= self.l2cap_channel.mtu:
if len(payload) + 2 <= self.l2cap_channel.peer_mtu:
# Fits in a single packet
packet_type = self.PacketType.SINGLE_PACKET
else:

View File

@@ -1151,7 +1151,28 @@ class Controller:
'''
See Bluetooth spec Vol 4, Part E - 7.4.3 Read Local Supported Features Command
'''
return bytes([HCI_SUCCESS]) + self.lmp_features
return bytes([HCI_SUCCESS]) + self.lmp_features[:8]
def on_hci_read_local_extended_features_command(self, command):
'''
See Bluetooth spec Vol 4, Part E - 7.4.4 Read Local Extended Features Command
'''
if command.page_number * 8 > len(self.lmp_features):
return bytes([HCI_INVALID_HCI_COMMAND_PARAMETERS_ERROR])
return (
bytes(
[
# Status
HCI_SUCCESS,
# Page number
command.page_number,
# Max page number
len(self.lmp_features) // 8 - 1,
]
)
# Features of the current page
+ self.lmp_features[command.page_number * 8 : (command.page_number + 1) * 8]
)
def on_hci_read_buffer_size_command(self, _command):
'''
@@ -1522,6 +1543,20 @@ class Controller:
}
return bytes([HCI_SUCCESS])
def on_hci_le_read_maximum_advertising_data_length_command(self, _command):
'''
See Bluetooth spec Vol 4, Part E - 7.8.57 LE Read Maximum Advertising Data
Length Command
'''
return struct.pack('<BH', HCI_SUCCESS, 0x0672)
def on_hci_le_read_number_of_supported_advertising_sets_command(self, _command):
'''
See Bluetooth spec Vol 4, Part E - 7.8.58 LE Read Number of Supported
Advertising Set Command
'''
return struct.pack('<BB', HCI_SUCCESS, 0xF0)
def on_hci_le_read_transmit_power_command(self, _command):
'''
See Bluetooth spec Vol 4, Part E - 7.8.74 LE Read Transmit Power Command

File diff suppressed because it is too large Load Diff

View File

@@ -25,7 +25,7 @@ import pathlib
import platform
from typing import Dict, Iterable, Optional, Type, TYPE_CHECKING
from . import rtk
from . import rtk, intel
from .common import Driver
if TYPE_CHECKING:
@@ -45,7 +45,7 @@ async def get_driver_for_host(host: Host) -> Optional[Driver]:
found.
If a "driver" HCI metadata entry is present, only that driver class will be probed.
"""
driver_classes: Dict[str, Type[Driver]] = {"rtk": rtk.Driver}
driver_classes: Dict[str, Type[Driver]] = {"rtk": rtk.Driver, "intel": intel.Driver}
probe_list: Iterable[str]
if driver_name := host.hci_metadata.get("driver"):
# Only probe a single driver

102
bumble/drivers/intel.py Normal file
View File

@@ -0,0 +1,102 @@
# 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.
# -----------------------------------------------------------------------------
# Imports
# -----------------------------------------------------------------------------
import logging
from bumble.drivers import common
from bumble.hci import (
hci_vendor_command_op_code, # type: ignore
HCI_Command,
HCI_Reset_Command,
)
# -----------------------------------------------------------------------------
# Logging
# -----------------------------------------------------------------------------
logger = logging.getLogger(__name__)
# -----------------------------------------------------------------------------
# Constant
# -----------------------------------------------------------------------------
INTEL_USB_PRODUCTS = {
# Intel AX210
(0x8087, 0x0032),
# Intel BE200
(0x8087, 0x0036),
}
# -----------------------------------------------------------------------------
# HCI Commands
# -----------------------------------------------------------------------------
HCI_INTEL_DDC_CONFIG_WRITE_COMMAND = hci_vendor_command_op_code(0xFC8B) # type: ignore
HCI_INTEL_DDC_CONFIG_WRITE_PAYLOAD = [0x03, 0xE4, 0x02, 0x00]
HCI_Command.register_commands(globals())
@HCI_Command.command( # type: ignore
fields=[("params", "*")],
return_parameters_fields=[
("params", "*"),
],
)
class Hci_Intel_DDC_Config_Write_Command(HCI_Command):
pass
class Driver(common.Driver):
def __init__(self, host):
self.host = host
@staticmethod
def check(host):
driver = host.hci_metadata.get("driver")
if driver == "intel":
return True
vendor_id = host.hci_metadata.get("vendor_id")
product_id = host.hci_metadata.get("product_id")
if vendor_id is None or product_id is None:
logger.debug("USB metadata not sufficient")
return False
if (vendor_id, product_id) not in INTEL_USB_PRODUCTS:
logger.debug(
f"USB device ({vendor_id:04X}, {product_id:04X}) " "not in known list"
)
return False
return True
@classmethod
async def for_host(cls, host, force=False): # type: ignore
# Only instantiate this driver if explicitly selected
if not force and not cls.check(host):
return None
return cls(host)
async def init_controller(self):
self.host.ready = True
await self.host.send_command(HCI_Reset_Command(), check_result=True)
await self.host.send_command(
Hci_Intel_DDC_Config_Write_Command(
params=HCI_INTEL_DDC_CONFIG_WRITE_PAYLOAD
)
)

View File

@@ -1068,7 +1068,7 @@ class Client:
logger.warning('!!! unexpected response, there is no pending request')
return
# Sanity check: the response should match the pending request unless it is
# The response should match the pending request unless it is
# an error response
if att_pdu.op_code != ATT_ERROR_RESPONSE:
expected_response_name = self.pending_request.name.replace(

View File

@@ -328,7 +328,7 @@ class Server(EventEmitter):
f'handle=0x{characteristic.handle:04X}: {value.hex()}'
)
# Sanity check
# Check parameters
if len(value) != 2:
logger.warning('CCCD value not 2 bytes long')
return

File diff suppressed because it is too large Load Diff

View File

@@ -18,6 +18,7 @@
from __future__ import annotations
from collections.abc import Callable, MutableMapping
import datetime
from typing import cast, Any, Optional
import logging
@@ -66,12 +67,13 @@ PSM_NAMES = {
rfcomm.RFCOMM_PSM: 'RFCOMM',
sdp.SDP_PSM: 'SDP',
avdtp.AVDTP_PSM: 'AVDTP',
avctp.AVCTP_PSM: 'AVCTP'
avctp.AVCTP_PSM: 'AVCTP',
# TODO: add more PSM values
}
AVCTP_PID_NAMES = {avrcp.AVRCP_PID: 'AVRCP'}
# -----------------------------------------------------------------------------
class PacketTracer:
class AclStream:
@@ -207,6 +209,7 @@ class PacketTracer:
self.label = label
self.emit_message = emit_message
self.acl_streams = {} # ACL streams, by connection handle
self.packet_timestamp: Optional[datetime.datetime] = None
def start_acl_stream(self, connection_handle: int) -> PacketTracer.AclStream:
logger.info(
@@ -234,7 +237,10 @@ class PacketTracer:
# Let the other forwarder know so it can cleanup its stream as well
self.peer.end_acl_stream(connection_handle)
def on_packet(self, packet: HCI_Packet) -> None:
def on_packet(
self, timestamp: Optional[datetime.datetime], packet: HCI_Packet
) -> None:
self.packet_timestamp = timestamp
self.emit(packet)
if packet.hci_packet_type == HCI_ACL_DATA_PACKET:
@@ -254,13 +260,22 @@ class PacketTracer:
)
def emit(self, message: Any) -> None:
self.emit_message(f'[{self.label}] {message}')
if self.packet_timestamp:
prefix = f"[{self.packet_timestamp.strftime('%Y-%m-%d %H:%M:%S.%f')}]"
else:
prefix = ""
self.emit_message(f'{prefix}[{self.label}] {message}')
def trace(self, packet: HCI_Packet, direction: int = 0) -> None:
def trace(
self,
packet: HCI_Packet,
direction: int = 0,
timestamp: Optional[datetime.datetime] = None,
) -> None:
if direction == 0:
self.host_to_controller_analyzer.on_packet(packet)
self.host_to_controller_analyzer.on_packet(timestamp, packet)
else:
self.controller_to_host_analyzer.on_packet(packet)
self.controller_to_host_analyzer.on_packet(timestamp, packet)
def __init__(
self,

View File

@@ -21,12 +21,11 @@ import asyncio
import dataclasses
import enum
import traceback
import warnings
from typing import Dict, List, Union, Set, Any, TYPE_CHECKING
from . import at
from . import rfcomm
import pyee
from typing import Dict, List, Union, Set, Any, Optional, TYPE_CHECKING
from bumble import at
from bumble import rfcomm
from bumble.colors import color
from bumble.core import (
ProtocolError,
@@ -79,7 +78,6 @@ class HfpProtocol:
lines_available: asyncio.Event
def __init__(self, dlc: rfcomm.DLC) -> None:
warnings.warn("See HfProtocol", DeprecationWarning)
self.dlc = dlc
self.buffer = ''
self.lines = collections.deque()
@@ -128,10 +126,13 @@ class HfpProtocol:
# -----------------------------------------------------------------------------
# HF supported features (AT+BRSF=) (normative).
# Hands-Free Profile v1.8, 4.34.2, AT Capabilities Re-Used from GSM 07.07
# and 3GPP 27.007
class HfFeature(enum.IntFlag):
"""
HF supported features (AT+BRSF=) (normative).
Hands-Free Profile v1.8, 4.34.2, AT Capabilities Re-Used from GSM 07.07 and 3GPP 27.007.
"""
EC_NR = 0x001 # Echo Cancel & Noise reduction
THREE_WAY_CALLING = 0x002
CLI_PRESENTATION_CAPABILITY = 0x004
@@ -146,10 +147,13 @@ class HfFeature(enum.IntFlag):
VOICE_RECOGNITION_TEST = 0x800
# AG supported features (+BRSF:) (normative).
# Hands-Free Profile v1.8, 4.34.2, AT Capabilities Re-Used from GSM 07.07
# and 3GPP 27.007
class AgFeature(enum.IntFlag):
"""
AG supported features (+BRSF:) (normative).
Hands-Free Profile v1.8, 4.34.2, AT Capabilities Re-Used from GSM 07.07 and 3GPP 27.007.
"""
THREE_WAY_CALLING = 0x001
EC_NR = 0x002 # Echo Cancel & Noise reduction
VOICE_RECOGNITION_FUNCTION = 0x004
@@ -166,52 +170,90 @@ class AgFeature(enum.IntFlag):
VOICE_RECOGNITION_TEST = 0x2000
# Audio Codec IDs (normative).
# Hands-Free Profile v1.8, 10 Appendix B
class AudioCodec(enum.IntEnum):
"""
Audio Codec IDs (normative).
Hands-Free Profile v1.9, 11 Appendix B
"""
CVSD = 0x01 # Support for CVSD audio codec
MSBC = 0x02 # Support for mSBC audio codec
LC3_SWB = 0x03 # Support for LC3-SWB audio codec
# HF Indicators (normative).
# Bluetooth Assigned Numbers, 6.10.1 HF Indicators
class HfIndicator(enum.IntEnum):
"""
HF Indicators (normative).
Bluetooth Assigned Numbers, 6.10.1 HF Indicators.
"""
ENHANCED_SAFETY = 0x01 # Enhanced safety feature
BATTERY_LEVEL = 0x02 # Battery level feature
# Call Hold supported operations (normative).
# AT Commands Reference Guide, 3.5.2.3.12 +CHLD - Call Holding Services
class CallHoldOperation(enum.IntEnum):
"""
Call Hold supported operations (normative).
AT Commands Reference Guide, 3.5.2.3.12 +CHLD - Call Holding Services.
"""
RELEASE_ALL_HELD_CALLS = 0 # Release all held calls
RELEASE_ALL_ACTIVE_CALLS = 1 # Release all active calls, accept other
HOLD_ALL_ACTIVE_CALLS = 2 # Place all active calls on hold, accept other
ADD_HELD_CALL = 3 # Adds a held call to conversation
# Response Hold status (normative).
# Hands-Free Profile v1.8, 4.34.2, AT Capabilities Re-Used from GSM 07.07
# and 3GPP 27.007
class ResponseHoldStatus(enum.IntEnum):
"""
Response Hold status (normative).
Hands-Free Profile v1.8, 4.34.2, AT Capabilities Re-Used from GSM 07.07 and 3GPP 27.007.
"""
INC_CALL_HELD = 0 # Put incoming call on hold
HELD_CALL_ACC = 1 # Accept a held incoming call
HELD_CALL_REJ = 2 # Reject a held incoming call
# Values for the Call Setup AG indicator (normative).
# Hands-Free Profile v1.8, 4.34.2, AT Capabilities Re-Used from GSM 07.07
# and 3GPP 27.007
class AgIndicator(enum.Enum):
"""
Values for the AG indicator (normative).
Hands-Free Profile v1.8, 4.34.2, AT Capabilities Re-Used from GSM 07.07 and 3GPP 27.007.
"""
SERVICE = 'service'
CALL = 'call'
CALL_SETUP = 'callsetup'
CALL_HELD = 'callheld'
SIGNAL = 'signal'
ROAM = 'roam'
BATTERY_CHARGE = 'battchg'
class CallSetupAgIndicator(enum.IntEnum):
"""
Values for the Call Setup AG indicator (normative).
Hands-Free Profile v1.8, 4.34.2, AT Capabilities Re-Used from GSM 07.07 and 3GPP 27.007.
"""
NOT_IN_CALL_SETUP = 0
INCOMING_CALL_PROCESS = 1
OUTGOING_CALL_SETUP = 2
REMOTE_ALERTED = 3 # Remote party alerted in an outgoing call
# Values for the Call Held AG indicator (normative).
# Hands-Free Profile v1.8, 4.34.2, AT Capabilities Re-Used from GSM 07.07
# and 3GPP 27.007
class CallHeldAgIndicator(enum.IntEnum):
"""
Values for the Call Held AG indicator (normative).
Hands-Free Profile v1.8, 4.34.2, AT Capabilities Re-Used from GSM 07.07 and 3GPP 27.007.
"""
NO_CALLS_HELD = 0
# Call is placed on hold or active/held calls swapped
# (The AG has both an active AND a held call)
@@ -219,16 +261,24 @@ class CallHeldAgIndicator(enum.IntEnum):
CALL_ON_HOLD_NO_ACTIVE_CALL = 2 # Call on hold, no active call
# Call Info direction (normative).
# AT Commands Reference Guide, 3.5.2.3.15 +CLCC - List Current Calls
class CallInfoDirection(enum.IntEnum):
"""
Call Info direction (normative).
AT Commands Reference Guide, 3.5.2.3.15 +CLCC - List Current Calls.
"""
MOBILE_ORIGINATED_CALL = 0
MOBILE_TERMINATED_CALL = 1
# Call Info status (normative).
# AT Commands Reference Guide, 3.5.2.3.15 +CLCC - List Current Calls
class CallInfoStatus(enum.IntEnum):
"""
Call Info status (normative).
AT Commands Reference Guide, 3.5.2.3.15 +CLCC - List Current Calls.
"""
ACTIVE = 0
HELD = 1
DIALING = 2
@@ -237,15 +287,47 @@ class CallInfoStatus(enum.IntEnum):
WAITING = 5
# Call Info mode (normative).
# AT Commands Reference Guide, 3.5.2.3.15 +CLCC - List Current Calls
class CallInfoMode(enum.IntEnum):
"""
Call Info mode (normative).
AT Commands Reference Guide, 3.5.2.3.15 +CLCC - List Current Calls.
"""
VOICE = 0
DATA = 1
FAX = 2
UNKNOWN = 9
class CallInfoMultiParty(enum.IntEnum):
"""
Call Info Multi-Party state (normative).
AT Commands Reference Guide, 3.5.2.3.15 +CLCC - List Current Calls.
"""
NOT_IN_CONFERENCE = 0
IN_CONFERENCE = 1
@dataclasses.dataclass
class CallInfo:
"""
Enhanced call status.
AT Commands Reference Guide, 3.5.2.3.15 +CLCC - List Current Calls.
"""
index: int
direction: CallInfoDirection
status: CallInfoStatus
mode: CallInfoMode
multi_party: CallInfoMultiParty
number: Optional[int] = None
type: Optional[int] = None
# -----------------------------------------------------------------------------
# Hands-Free Control Interoperability Requirements
# -----------------------------------------------------------------------------
@@ -326,8 +408,9 @@ class Configuration:
class AtResponseType(enum.Enum):
"""Indicate if a response is expected from an AT command, and if multiple
responses are accepted."""
"""
Indicates if a response is expected from an AT command, and if multiple responses are accepted.
"""
NONE = 0
SINGLE = 1
@@ -361,9 +444,20 @@ class HfIndicatorState:
enabled: bool = False
class HfProtocol:
"""Implementation for the Hands-Free side of the Hands-Free profile.
Reference specification Hands-Free Profile v1.8"""
class HfProtocol(pyee.EventEmitter):
"""
Implementation for the Hands-Free side of the Hands-Free profile.
Reference specification Hands-Free Profile v1.8.
Emitted events:
codec_negotiation: When codec is renegotiated, notify the new codec.
Args:
active_codec: AudioCodec
ag_indicator: When AG update their indicators, notify the new state.
Args:
ag_indicator: AgIndicator
"""
supported_hf_features: int
supported_audio_codecs: List[AudioCodec]
@@ -383,14 +477,18 @@ class HfProtocol:
response_queue: asyncio.Queue
unsolicited_queue: asyncio.Queue
read_buffer: bytearray
active_codec: AudioCodec
def __init__(self, dlc: rfcomm.DLC, configuration: Configuration) -> None:
super().__init__()
def __init__(self, dlc: rfcomm.DLC, configuration: Configuration):
# Configure internal state.
self.dlc = dlc
self.command_lock = asyncio.Lock()
self.response_queue = asyncio.Queue()
self.unsolicited_queue = asyncio.Queue()
self.read_buffer = bytearray()
self.active_codec = AudioCodec.CVSD
# Build local features.
self.supported_hf_features = sum(configuration.supported_hf_features)
@@ -415,10 +513,12 @@ class HfProtocol:
def supports_ag_feature(self, feature: AgFeature) -> bool:
return (self.supported_ag_features & feature) != 0
# Read AT messages from the RFCOMM channel.
# Enqueue AT commands, responses, unsolicited responses to their
# respective queues, and set the corresponding event.
def _read_at(self, data: bytes):
"""
Reads AT messages from the RFCOMM channel.
Enqueues AT commands, responses, unsolicited responses to their respective queues, and set the corresponding event.
"""
# Append to the read buffer.
self.read_buffer.extend(data)
@@ -446,17 +546,25 @@ class HfProtocol:
else:
logger.warning(f"dropping unexpected response with code '{response.code}'")
# Send an AT command and wait for the peer response.
# Wait for the AT responses sent by the peer, to the status code.
# Raises asyncio.TimeoutError if the status is not received
# after a timeout (default 1 second).
# Raises ProtocolError if the status is not OK.
async def execute_command(
self,
cmd: str,
timeout: float = 1.0,
response_type: AtResponseType = AtResponseType.NONE,
) -> Union[None, AtResponse, List[AtResponse]]:
"""
Sends an AT command and wait for the peer response.
Wait for the AT responses sent by the peer, to the status code.
Args:
cmd: the AT command in string to execute.
timeout: timeout in float seconds.
response_type: type of response.
Raises:
asyncio.TimeoutError: the status is not received after a timeout (default 1 second).
ProtocolError: the status is not OK.
"""
async with self.command_lock:
logger.debug(f">>> {cmd}")
self.dlc.write(cmd + '\r')
@@ -479,8 +587,9 @@ class HfProtocol:
raise HfpProtocolError(result.code)
responses.append(result)
# 4.2.1 Service Level Connection Initialization.
async def initiate_slc(self):
"""4.2.1 Service Level Connection Initialization."""
# 4.2.1.1 Supported features exchange
# First, in the initialization procedure, the HF shall send the
# AT+BRSF=<HF supported features> command to the AG to both notify
@@ -620,16 +729,17 @@ class HfProtocol:
logger.info("SLC setup completed")
# 4.11.2 Audio Connection Setup by HF
async def setup_audio_connection(self):
"""4.11.2 Audio Connection Setup by HF."""
# When the HF triggers the establishment of the Codec Connection it
# shall send the AT command AT+BCC to the AG. The AG shall respond with
# OK if it will start the Codec Connection procedure, and with ERROR
# if it cannot start the Codec Connection procedure.
await self.execute_command("AT+BCC")
# 4.11.3 Codec Connection Setup
async def setup_codec_connection(self, codec_id: int):
"""4.11.3 Codec Connection Setup."""
# The AG shall send a +BCS=<Codec ID> unsolicited response to the HF.
# The HF shall then respond to the incoming unsolicited response with
# the AT command AT+BCS=<Codec ID>. The ID shall be the same as in the
@@ -647,27 +757,29 @@ class HfProtocol:
# Synchronous Connection with the settings that are determined by the
# ID. The HF shall be ready to accept the synchronous connection
# establishment as soon as it has sent the AT commands AT+BCS=<Codec ID>.
self.active_codec = AudioCodec(codec_id)
self.emit('codec_negotiation', self.active_codec)
logger.info("codec connection setup completed")
# 4.13.1 Answer Incoming Call from the HF In-Band Ringing
async def answer_incoming_call(self):
"""4.13.1 Answer Incoming Call from the HF - In-Band Ringing."""
# The user accepts the incoming voice call by using the proper means
# provided by the HF. The HF shall then send the ATA command
# (see Section 4.34) to the AG. The AG shall then begin the procedure for
# accepting the incoming call.
await self.execute_command("ATA")
# 4.14.1 Reject an Incoming Call from the HF
async def reject_incoming_call(self):
"""4.14.1 Reject an Incoming Call from the HF."""
# The user rejects the incoming call by using the User Interface on the
# Hands-Free unit. The HF shall then send the AT+CHUP command
# (see Section 4.34) to the AG. This may happen at any time during the
# procedures described in Sections 4.13.1 and 4.13.2.
await self.execute_command("AT+CHUP")
# 4.15.1 Terminate a Call Process from the HF
async def terminate_call(self):
"""4.15.1 Terminate a Call Process from the HF."""
# The user may abort the ongoing call process using whatever means
# provided by the Hands-Free unit. The HF shall send AT+CHUP command
# (see Section 4.34) to the AG, and the AG shall then start the
@@ -676,8 +788,35 @@ class HfProtocol:
# code, with the value indicating (call=0).
await self.execute_command("AT+CHUP")
async def query_current_calls(self) -> List[CallInfo]:
"""4.32.1 Query List of Current Calls in AG.
Return:
List of current calls in AG.
"""
responses = await self.execute_command(
"AT+CLCC", response_type=AtResponseType.MULTIPLE
)
assert isinstance(responses, list)
calls = []
for response in responses:
call_info = CallInfo(
index=int(response.parameters[0]),
direction=CallInfoDirection(int(response.parameters[1])),
status=CallInfoStatus(int(response.parameters[2])),
mode=CallInfoMode(int(response.parameters[3])),
multi_party=CallInfoMultiParty(int(response.parameters[4])),
)
if len(response.parameters) >= 7:
call_info.number = int(response.parameters[5])
call_info.type = int(response.parameters[6])
calls.append(call_info)
return calls
async def update_ag_indicator(self, index: int, value: int):
self.ag_indicators[index].current_status = value
self.emit('ag_indicator', self.ag_indicators[index])
logger.info(
f"AG indicator updated: {self.ag_indicators[index].description}, {value}"
)
@@ -695,9 +834,11 @@ class HfProtocol:
logging.info(f"unhandled unsolicited response {result.code}")
async def run(self):
"""Main rountine for the Hands-Free side of the HFP protocol.
Initiates the service level connection then loops handling
unsolicited AG responses."""
"""
Main routine for the Hands-Free side of the HFP protocol.
Initiates the service level connection then loops handling unsolicited AG responses.
"""
try:
await self.initiate_slc()
@@ -713,9 +854,13 @@ class HfProtocol:
# -----------------------------------------------------------------------------
# Profile version (normative).
# Hands-Free Profile v1.8, 5.3 SDP Interoperability Requirements
class ProfileVersion(enum.IntEnum):
"""
Profile version (normative).
Hands-Free Profile v1.8, 5.3 SDP Interoperability Requirements.
"""
V1_5 = 0x0105
V1_6 = 0x0106
V1_7 = 0x0107
@@ -723,9 +868,13 @@ class ProfileVersion(enum.IntEnum):
V1_9 = 0x0109
# HF supported features (normative).
# Hands-Free Profile v1.8, 5.3 SDP Interoperability Requirements
class HfSdpFeature(enum.IntFlag):
"""
HF supported features (normative).
Hands-Free Profile v1.8, 5.3 SDP Interoperability Requirements.
"""
EC_NR = 0x01 # Echo Cancel & Noise reduction
THREE_WAY_CALLING = 0x02
CLI_PRESENTATION_CAPABILITY = 0x04
@@ -736,9 +885,13 @@ class HfSdpFeature(enum.IntFlag):
VOICE_RECOGNITION_TEST = 0x80
# AG supported features (normative).
# Hands-Free Profile v1.8, 5.3 SDP Interoperability Requirements
class AgSdpFeature(enum.IntFlag):
"""
AG supported features (normative).
Hands-Free Profile v1.8, 5.3 SDP Interoperability Requirements.
"""
THREE_WAY_CALLING = 0x01
EC_NR = 0x02 # Echo Cancel & Noise reduction
VOICE_RECOGNITION_FUNCTION = 0x04
@@ -752,9 +905,12 @@ class AgSdpFeature(enum.IntFlag):
def sdp_records(
service_record_handle: int, rfcomm_channel: int, configuration: Configuration
) -> List[ServiceAttribute]:
"""Generate the SDP record for HFP Hands-Free support.
"""
Generates the SDP record for HFP Hands-Free support.
The record exposes the features supported in the input configuration,
and the allocated RFCOMM channel."""
and the allocated RFCOMM channel.
"""
hf_supported_features = 0

View File

@@ -416,7 +416,7 @@ class Device(HID):
data = bytearray()
data.append(report_id)
data.extend(ret.data)
if len(data) < self.l2cap_ctrl_channel.mtu: # type: ignore[union-attr]
if len(data) < self.l2cap_ctrl_channel.peer_mtu: # type: ignore[union-attr]
self.send_control_data(report_type=report_type, data=data)
else:
self.send_handshake_message(Message.Handshake.ERR_INVALID_PARAMETER)

View File

@@ -22,65 +22,31 @@ import dataclasses
import logging
import struct
from typing import Any, Awaitable, Callable, Deque, Dict, Optional, cast, TYPE_CHECKING
from typing import (
Any,
Awaitable,
Callable,
Deque,
Dict,
Optional,
Set,
cast,
TYPE_CHECKING,
)
from bumble.colors import color
from bumble.l2cap import L2CAP_PDU
from bumble.snoop import Snooper
from bumble import drivers
from .hci import (
Address,
HCI_ACL_DATA_PACKET,
HCI_COMMAND_PACKET,
HCI_EVENT_PACKET,
HCI_ISO_DATA_PACKET,
HCI_LE_READ_BUFFER_SIZE_COMMAND,
HCI_LE_READ_LOCAL_SUPPORTED_FEATURES_COMMAND,
HCI_LE_READ_SUGGESTED_DEFAULT_DATA_LENGTH_COMMAND,
HCI_LE_WRITE_SUGGESTED_DEFAULT_DATA_LENGTH_COMMAND,
HCI_READ_BUFFER_SIZE_COMMAND,
HCI_READ_LOCAL_VERSION_INFORMATION_COMMAND,
HCI_RESET_COMMAND,
HCI_SUCCESS,
HCI_SUPPORTED_COMMANDS_FLAGS,
HCI_SYNCHRONOUS_DATA_PACKET,
HCI_VERSION_BLUETOOTH_CORE_4_0,
HCI_AclDataPacket,
HCI_AclDataPacketAssembler,
HCI_Command,
HCI_Command_Complete_Event,
HCI_Constant,
HCI_Error,
HCI_Event,
HCI_IsoDataPacket,
HCI_LE_Long_Term_Key_Request_Negative_Reply_Command,
HCI_LE_Long_Term_Key_Request_Reply_Command,
HCI_LE_Read_Buffer_Size_Command,
HCI_LE_Read_Local_Supported_Features_Command,
HCI_LE_Read_Suggested_Default_Data_Length_Command,
HCI_LE_Remote_Connection_Parameter_Request_Reply_Command,
HCI_LE_Set_Event_Mask_Command,
HCI_LE_Write_Suggested_Default_Data_Length_Command,
HCI_Link_Key_Request_Negative_Reply_Command,
HCI_Link_Key_Request_Reply_Command,
HCI_Packet,
HCI_Read_Buffer_Size_Command,
HCI_Read_Local_Supported_Commands_Command,
HCI_Read_Local_Version_Information_Command,
HCI_Reset_Command,
HCI_Set_Event_Mask_Command,
HCI_SynchronousDataPacket,
LeFeatureMask,
)
from .core import (
from bumble import hci
from bumble.core import (
BT_BR_EDR_TRANSPORT,
BT_LE_TRANSPORT,
ConnectionPHY,
ConnectionParameters,
)
from .utils import AbortableEventEmitter
from .transport.common import TransportLostError
from bumble.utils import AbortableEventEmitter
from bumble.transport.common import TransportLostError
if TYPE_CHECKING:
from .transport.common import TransportSink, TransportSource
@@ -100,15 +66,15 @@ class AclPacketQueue:
self,
max_packet_size: int,
max_in_flight: int,
send: Callable[[HCI_Packet], None],
send: Callable[[hci.HCI_Packet], None],
) -> None:
self.max_packet_size = max_packet_size
self.max_in_flight = max_in_flight
self.in_flight = 0
self.send = send
self.packets: Deque[HCI_AclDataPacket] = collections.deque()
self.packets: Deque[hci.HCI_AclDataPacket] = collections.deque()
def enqueue(self, packet: HCI_AclDataPacket) -> None:
def enqueue(self, packet: hci.HCI_AclDataPacket) -> None:
self.packets.appendleft(packet)
self.check_queue()
@@ -140,11 +106,13 @@ class AclPacketQueue:
# -----------------------------------------------------------------------------
class Connection:
def __init__(self, host: Host, handle: int, peer_address: Address, transport: int):
def __init__(
self, host: Host, handle: int, peer_address: hci.Address, transport: int
):
self.host = host
self.handle = handle
self.peer_address = peer_address
self.assembler = HCI_AclDataPacketAssembler(self.on_acl_pdu)
self.assembler = hci.HCI_AclDataPacketAssembler(self.on_acl_pdu)
self.transport = transport
acl_packet_queue: Optional[AclPacketQueue] = (
host.le_acl_packet_queue
@@ -154,7 +122,7 @@ class Connection:
assert acl_packet_queue
self.acl_packet_queue = acl_packet_queue
def on_hci_acl_data_packet(self, packet: HCI_AclDataPacket) -> None:
def on_hci_acl_data_packet(self, packet: hci.HCI_AclDataPacket) -> None:
self.assembler.feed_packet(packet)
def on_acl_pdu(self, pdu: bytes) -> None:
@@ -165,14 +133,14 @@ class Connection:
# -----------------------------------------------------------------------------
@dataclasses.dataclass
class ScoLink:
peer_address: Address
peer_address: hci.Address
handle: int
# -----------------------------------------------------------------------------
@dataclasses.dataclass
class CisLink:
peer_address: Address
peer_address: hci.Address
handle: int
@@ -188,7 +156,7 @@ class Host(AbortableEventEmitter):
long_term_key_provider: Optional[
Callable[[int, bytes, int], Awaitable[Optional[bytes]]]
]
link_key_provider: Optional[Callable[[Address], Awaitable[Optional[bytes]]]]
link_key_provider: Optional[Callable[[hci.Address], Awaitable[Optional[bytes]]]]
def __init__(
self,
@@ -204,9 +172,12 @@ class Host(AbortableEventEmitter):
self.sco_links = {} # SCO links, by connection handle
self.pending_command = None
self.pending_response = None
self.number_of_supported_advertising_sets = 0
self.maximum_advertising_data_length = 31
self.local_version = None
self.local_supported_commands = bytes(64)
self.local_supported_commands = 0
self.local_le_features = 0
self.local_lmp_features = hci.LmpFeatureMask(0) # Classic LMP features
self.suggested_max_tx_octets = 251 # Max allowed
self.suggested_max_tx_time = 2120 # Max allowed
self.command_semaphore = asyncio.Semaphore(1)
@@ -223,7 +194,7 @@ class Host(AbortableEventEmitter):
def find_connection_by_bd_addr(
self,
bd_addr: Address,
bd_addr: hci.Address,
transport: Optional[int] = None,
check_address_type: bool = False,
) -> Optional[Connection]:
@@ -265,49 +236,167 @@ class Host(AbortableEventEmitter):
# Send a reset command unless a driver has already done so.
if reset_needed:
await self.send_command(HCI_Reset_Command(), check_result=True)
await self.send_command(hci.HCI_Reset_Command(), check_result=True)
self.ready = True
response = await self.send_command(
HCI_Read_Local_Supported_Commands_Command(), check_result=True
hci.HCI_Read_Local_Supported_Commands_Command(), check_result=True
)
self.local_supported_commands = int.from_bytes(
response.return_parameters.supported_commands, 'little'
)
self.local_supported_commands = response.return_parameters.supported_commands
if self.supports_command(HCI_LE_READ_LOCAL_SUPPORTED_FEATURES_COMMAND):
if self.supports_command(hci.HCI_LE_READ_LOCAL_SUPPORTED_FEATURES_COMMAND):
response = await self.send_command(
HCI_LE_Read_Local_Supported_Features_Command(), check_result=True
hci.HCI_LE_Read_Local_Supported_Features_Command(), check_result=True
)
self.local_le_features = struct.unpack(
'<Q', response.return_parameters.le_features
)[0]
if self.supports_command(HCI_READ_LOCAL_VERSION_INFORMATION_COMMAND):
if self.supports_command(hci.HCI_READ_LOCAL_VERSION_INFORMATION_COMMAND):
response = await self.send_command(
HCI_Read_Local_Version_Information_Command(), check_result=True
hci.HCI_Read_Local_Version_Information_Command(), check_result=True
)
self.local_version = response.return_parameters
if self.supports_command(hci.HCI_READ_LOCAL_EXTENDED_FEATURES_COMMAND):
max_page_number = 0
page_number = 0
lmp_features = 0
while page_number <= max_page_number:
response = await self.send_command(
hci.HCI_Read_Local_Extended_Features_Command(
page_number=page_number
),
check_result=True,
)
lmp_features |= int.from_bytes(
response.return_parameters.extended_lmp_features, 'little'
) << (64 * page_number)
max_page_number = response.return_parameters.maximum_page_number
page_number += 1
self.local_lmp_features = hci.LmpFeatureMask(lmp_features)
elif self.supports_command(hci.HCI_READ_LOCAL_SUPPORTED_FEATURES_COMMAND):
response = await self.send_command(
hci.HCI_Read_Local_Supported_Features_Command(), check_result=True
)
self.local_lmp_features = hci.LmpFeatureMask(
int.from_bytes(response.return_parameters.lmp_features, 'little')
)
await self.send_command(
HCI_Set_Event_Mask_Command(event_mask=bytes.fromhex('FFFFFFFFFFFFFF3F'))
hci.HCI_Set_Event_Mask_Command(
event_mask=hci.HCI_Set_Event_Mask_Command.mask(
[
hci.HCI_INQUIRY_COMPLETE_EVENT,
hci.HCI_INQUIRY_RESULT_EVENT,
hci.HCI_CONNECTION_COMPLETE_EVENT,
hci.HCI_CONNECTION_REQUEST_EVENT,
hci.HCI_DISCONNECTION_COMPLETE_EVENT,
hci.HCI_AUTHENTICATION_COMPLETE_EVENT,
hci.HCI_REMOTE_NAME_REQUEST_COMPLETE_EVENT,
hci.HCI_ENCRYPTION_CHANGE_EVENT,
hci.HCI_CHANGE_CONNECTION_LINK_KEY_COMPLETE_EVENT,
hci.HCI_LINK_KEY_TYPE_CHANGED_EVENT,
hci.HCI_READ_REMOTE_SUPPORTED_FEATURES_COMPLETE_EVENT,
hci.HCI_READ_REMOTE_VERSION_INFORMATION_COMPLETE_EVENT,
hci.HCI_QOS_SETUP_COMPLETE_EVENT,
hci.HCI_HARDWARE_ERROR_EVENT,
hci.HCI_FLUSH_OCCURRED_EVENT,
hci.HCI_ROLE_CHANGE_EVENT,
hci.HCI_MODE_CHANGE_EVENT,
hci.HCI_RETURN_LINK_KEYS_EVENT,
hci.HCI_PIN_CODE_REQUEST_EVENT,
hci.HCI_LINK_KEY_REQUEST_EVENT,
hci.HCI_LINK_KEY_NOTIFICATION_EVENT,
hci.HCI_LOOPBACK_COMMAND_EVENT,
hci.HCI_DATA_BUFFER_OVERFLOW_EVENT,
hci.HCI_MAX_SLOTS_CHANGE_EVENT,
hci.HCI_READ_CLOCK_OFFSET_COMPLETE_EVENT,
hci.HCI_CONNECTION_PACKET_TYPE_CHANGED_EVENT,
hci.HCI_QOS_VIOLATION_EVENT,
hci.HCI_PAGE_SCAN_REPETITION_MODE_CHANGE_EVENT,
hci.HCI_FLOW_SPECIFICATION_COMPLETE_EVENT,
hci.HCI_INQUIRY_RESULT_WITH_RSSI_EVENT,
hci.HCI_READ_REMOTE_EXTENDED_FEATURES_COMPLETE_EVENT,
hci.HCI_SYNCHRONOUS_CONNECTION_COMPLETE_EVENT,
hci.HCI_SYNCHRONOUS_CONNECTION_CHANGED_EVENT,
hci.HCI_SNIFF_SUBRATING_EVENT,
hci.HCI_EXTENDED_INQUIRY_RESULT_EVENT,
hci.HCI_ENCRYPTION_KEY_REFRESH_COMPLETE_EVENT,
hci.HCI_IO_CAPABILITY_REQUEST_EVENT,
hci.HCI_IO_CAPABILITY_RESPONSE_EVENT,
hci.HCI_USER_CONFIRMATION_REQUEST_EVENT,
hci.HCI_USER_PASSKEY_REQUEST_EVENT,
hci.HCI_REMOTE_OOB_DATA_REQUEST_EVENT,
hci.HCI_SIMPLE_PAIRING_COMPLETE_EVENT,
hci.HCI_LINK_SUPERVISION_TIMEOUT_CHANGED_EVENT,
hci.HCI_ENHANCED_FLUSH_COMPLETE_EVENT,
hci.HCI_USER_PASSKEY_NOTIFICATION_EVENT,
hci.HCI_KEYPRESS_NOTIFICATION_EVENT,
hci.HCI_REMOTE_HOST_SUPPORTED_FEATURES_NOTIFICATION_EVENT,
hci.HCI_LE_META_EVENT,
]
)
)
)
if (
self.local_version is not None
and self.local_version.hci_version <= HCI_VERSION_BLUETOOTH_CORE_4_0
and self.local_version.hci_version <= hci.HCI_VERSION_BLUETOOTH_CORE_4_0
):
# Some older controllers don't like event masks with bits they don't
# understand
le_event_mask = bytes.fromhex('1F00000000000000')
else:
le_event_mask = bytes.fromhex('FFFFFFFF00000000')
le_event_mask = hci.HCI_LE_Set_Event_Mask_Command.mask(
[
hci.HCI_LE_CONNECTION_COMPLETE_EVENT,
hci.HCI_LE_ADVERTISING_REPORT_EVENT,
hci.HCI_LE_CONNECTION_UPDATE_COMPLETE_EVENT,
hci.HCI_LE_READ_REMOTE_FEATURES_COMPLETE_EVENT,
hci.HCI_LE_LONG_TERM_KEY_REQUEST_EVENT,
hci.HCI_LE_REMOTE_CONNECTION_PARAMETER_REQUEST_EVENT,
hci.HCI_LE_DATA_LENGTH_CHANGE_EVENT,
hci.HCI_LE_READ_LOCAL_P_256_PUBLIC_KEY_COMPLETE_EVENT,
hci.HCI_LE_GENERATE_DHKEY_COMPLETE_EVENT,
hci.HCI_LE_ENHANCED_CONNECTION_COMPLETE_EVENT,
hci.HCI_LE_DIRECTED_ADVERTISING_REPORT_EVENT,
hci.HCI_LE_PHY_UPDATE_COMPLETE_EVENT,
hci.HCI_LE_EXTENDED_ADVERTISING_REPORT_EVENT,
hci.HCI_LE_PERIODIC_ADVERTISING_SYNC_ESTABLISHED_EVENT,
hci.HCI_LE_PERIODIC_ADVERTISING_REPORT_EVENT,
hci.HCI_LE_PERIODIC_ADVERTISING_SYNC_LOST_EVENT,
hci.HCI_LE_SCAN_TIMEOUT_EVENT,
hci.HCI_LE_ADVERTISING_SET_TERMINATED_EVENT,
hci.HCI_LE_SCAN_REQUEST_RECEIVED_EVENT,
hci.HCI_LE_CONNECTIONLESS_IQ_REPORT_EVENT,
hci.HCI_LE_CONNECTION_IQ_REPORT_EVENT,
hci.HCI_LE_CTE_REQUEST_FAILED_EVENT,
hci.HCI_LE_PERIODIC_ADVERTISING_SYNC_TRANSFER_RECEIVED_EVENT,
hci.HCI_LE_CIS_ESTABLISHED_EVENT,
hci.HCI_LE_CIS_REQUEST_EVENT,
hci.HCI_LE_CREATE_BIG_COMPLETE_EVENT,
hci.HCI_LE_TERMINATE_BIG_COMPLETE_EVENT,
hci.HCI_LE_BIG_SYNC_ESTABLISHED_EVENT,
hci.HCI_LE_BIG_SYNC_LOST_EVENT,
hci.HCI_LE_REQUEST_PEER_SCA_COMPLETE_EVENT,
hci.HCI_LE_PATH_LOSS_THRESHOLD_EVENT,
hci.HCI_LE_TRANSMIT_POWER_REPORTING_EVENT,
hci.HCI_LE_BIGINFO_ADVERTISING_REPORT_EVENT,
hci.HCI_LE_SUBRATE_CHANGE_EVENT,
]
)
await self.send_command(
HCI_LE_Set_Event_Mask_Command(le_event_mask=le_event_mask)
hci.HCI_LE_Set_Event_Mask_Command(le_event_mask=le_event_mask)
)
if self.supports_command(HCI_READ_BUFFER_SIZE_COMMAND):
if self.supports_command(hci.HCI_READ_BUFFER_SIZE_COMMAND):
response = await self.send_command(
HCI_Read_Buffer_Size_Command(), check_result=True
hci.HCI_Read_Buffer_Size_Command(), check_result=True
)
hc_acl_data_packet_length = (
response.return_parameters.hc_acl_data_packet_length
@@ -330,9 +419,9 @@ class Host(AbortableEventEmitter):
hc_le_acl_data_packet_length = 0
hc_total_num_le_acl_data_packets = 0
if self.supports_command(HCI_LE_READ_BUFFER_SIZE_COMMAND):
if self.supports_command(hci.HCI_LE_READ_BUFFER_SIZE_COMMAND):
response = await self.send_command(
HCI_LE_Read_Buffer_Size_Command(), check_result=True
hci.HCI_LE_Read_Buffer_Size_Command(), check_result=True
)
hc_le_acl_data_packet_length = (
response.return_parameters.hc_le_acl_data_packet_length
@@ -359,10 +448,12 @@ class Host(AbortableEventEmitter):
)
if self.supports_command(
HCI_LE_READ_SUGGESTED_DEFAULT_DATA_LENGTH_COMMAND
) and self.supports_command(HCI_LE_WRITE_SUGGESTED_DEFAULT_DATA_LENGTH_COMMAND):
hci.HCI_LE_READ_SUGGESTED_DEFAULT_DATA_LENGTH_COMMAND
) and self.supports_command(
hci.HCI_LE_WRITE_SUGGESTED_DEFAULT_DATA_LENGTH_COMMAND
):
response = await self.send_command(
HCI_LE_Read_Suggested_Default_Data_Length_Command()
hci.HCI_LE_Read_Suggested_Default_Data_Length_Command()
)
suggested_max_tx_octets = response.return_parameters.suggested_max_tx_octets
suggested_max_tx_time = response.return_parameters.suggested_max_tx_time
@@ -371,12 +462,34 @@ class Host(AbortableEventEmitter):
or suggested_max_tx_time != self.suggested_max_tx_time
):
await self.send_command(
HCI_LE_Write_Suggested_Default_Data_Length_Command(
hci.HCI_LE_Write_Suggested_Default_Data_Length_Command(
suggested_max_tx_octets=self.suggested_max_tx_octets,
suggested_max_tx_time=self.suggested_max_tx_time,
)
)
if self.supports_command(
hci.HCI_LE_READ_NUMBER_OF_SUPPORTED_ADVERTISING_SETS_COMMAND
):
response = await self.send_command(
hci.HCI_LE_Read_Number_Of_Supported_Advertising_Sets_Command(),
check_result=True,
)
self.number_of_supported_advertising_sets = (
response.return_parameters.num_supported_advertising_sets
)
if self.supports_command(
hci.HCI_LE_READ_MAXIMUM_ADVERTISING_DATA_LENGTH_COMMAND
):
response = await self.send_command(
hci.HCI_LE_Read_Maximum_Advertising_Data_Length_Command(),
check_result=True,
)
self.maximum_advertising_data_length = (
response.return_parameters.max_advertising_data_length
)
@property
def controller(self) -> Optional[TransportSink]:
return self.hci_sink
@@ -385,7 +498,7 @@ class Host(AbortableEventEmitter):
def controller(self, controller) -> None:
self.set_packet_sink(controller)
if controller:
controller.set_packet_sink(self)
self.set_packet_source(controller)
def set_packet_sink(self, sink: Optional[TransportSink]) -> None:
self.hci_sink = sink
@@ -394,7 +507,7 @@ class Host(AbortableEventEmitter):
source.set_packet_sink(self)
self.hci_metadata = getattr(source, 'metadata', self.hci_metadata)
def send_hci_packet(self, packet: HCI_Packet) -> None:
def send_hci_packet(self, packet: hci.HCI_Packet) -> None:
logger.debug(f'{color("### HOST -> CONTROLLER", "blue")}: {packet}')
if self.snooper:
self.snooper.snoop(bytes(packet), Snooper.Direction.HOST_TO_CONTROLLER)
@@ -425,11 +538,12 @@ class Host(AbortableEventEmitter):
else:
status = response.return_parameters.status
if status != HCI_SUCCESS:
if status != hci.HCI_SUCCESS:
logger.warning(
f'{command.name} failed ({HCI_Constant.error_name(status)})'
f'{command.name} failed '
f'({hci.HCI_Constant.error_name(status)})'
)
raise HCI_Error(status)
raise hci.HCI_Error(status)
return response
except Exception as error:
@@ -442,8 +556,8 @@ class Host(AbortableEventEmitter):
self.pending_response = None
# Use this method to send a command from a task
def send_command_sync(self, command: HCI_Command) -> None:
async def send_command(command: HCI_Command) -> None:
def send_command_sync(self, command: hci.HCI_Command) -> None:
async def send_command(command: hci.HCI_Command) -> None:
await self.send_command(command)
asyncio.create_task(send_command(command))
@@ -468,7 +582,7 @@ class Host(AbortableEventEmitter):
pb_flag = 0
while bytes_remaining:
data_total_length = min(bytes_remaining, packet_queue.max_packet_size)
acl_packet = HCI_AclDataPacket(
acl_packet = hci.HCI_AclDataPacket(
connection_handle=connection_handle,
pb_flag=pb_flag,
bc_flag=0,
@@ -481,35 +595,26 @@ class Host(AbortableEventEmitter):
offset += data_total_length
bytes_remaining -= data_total_length
def supports_command(self, command):
# Find the support flag position for this command
for octet, flags in enumerate(HCI_SUPPORTED_COMMANDS_FLAGS):
for flag_position, value in enumerate(flags):
if value == command:
# Check if the flag is set
if octet < len(self.local_supported_commands) and flag_position < 8:
return (
self.local_supported_commands[octet] & (1 << flag_position)
) != 0
return False
def supports_command(self, op_code: int) -> bool:
return (
self.local_supported_commands
& hci.HCI_SUPPORTED_COMMANDS_MASKS.get(op_code, 0)
) != 0
@property
def supported_commands(self):
commands = []
for octet, flags in enumerate(self.local_supported_commands):
if octet < len(HCI_SUPPORTED_COMMANDS_FLAGS):
for flag in range(8):
if flags & (1 << flag) != 0:
command = HCI_SUPPORTED_COMMANDS_FLAGS[octet][flag]
if command is not None:
commands.append(command)
def supported_commands(self) -> Set[int]:
return set(
op_code
for op_code, mask in hci.HCI_SUPPORTED_COMMANDS_MASKS.items()
if self.local_supported_commands & mask
)
return commands
def supports_le_features(self, feature: LeFeatureMask) -> bool:
def supports_le_features(self, feature: hci.LeFeatureMask) -> bool:
return (self.local_le_features & feature) == feature
def supports_lmp_features(self, feature: hci.LmpFeatureMask) -> bool:
return self.local_lmp_features & (feature) == feature
@property
def supported_le_features(self):
return [
@@ -518,10 +623,10 @@ class Host(AbortableEventEmitter):
# Packet Sink protocol (packets coming from the controller via HCI)
def on_packet(self, packet: bytes) -> None:
hci_packet = HCI_Packet.from_bytes(packet)
hci_packet = hci.HCI_Packet.from_bytes(packet)
if self.ready or (
isinstance(hci_packet, HCI_Command_Complete_Event)
and hci_packet.command_opcode == HCI_RESET_COMMAND
isinstance(hci_packet, hci.HCI_Command_Complete_Event)
and hci_packet.command_opcode == hci.HCI_RESET_COMMAND
):
self.on_hci_packet(hci_packet)
else:
@@ -534,44 +639,44 @@ class Host(AbortableEventEmitter):
self.emit('flush')
def on_hci_packet(self, packet: HCI_Packet) -> None:
def on_hci_packet(self, packet: hci.HCI_Packet) -> None:
logger.debug(f'{color("### CONTROLLER -> HOST", "green")}: {packet}')
if self.snooper:
self.snooper.snoop(bytes(packet), Snooper.Direction.CONTROLLER_TO_HOST)
# If the packet is a command, invoke the handler for this packet
if packet.hci_packet_type == HCI_COMMAND_PACKET:
self.on_hci_command_packet(cast(HCI_Command, packet))
elif packet.hci_packet_type == HCI_EVENT_PACKET:
self.on_hci_event_packet(cast(HCI_Event, packet))
elif packet.hci_packet_type == HCI_ACL_DATA_PACKET:
self.on_hci_acl_data_packet(cast(HCI_AclDataPacket, packet))
elif packet.hci_packet_type == HCI_SYNCHRONOUS_DATA_PACKET:
self.on_hci_sco_data_packet(cast(HCI_SynchronousDataPacket, packet))
elif packet.hci_packet_type == HCI_ISO_DATA_PACKET:
self.on_hci_iso_data_packet(cast(HCI_IsoDataPacket, packet))
if packet.hci_packet_type == hci.HCI_COMMAND_PACKET:
self.on_hci_command_packet(cast(hci.HCI_Command, packet))
elif packet.hci_packet_type == hci.HCI_EVENT_PACKET:
self.on_hci_event_packet(cast(hci.HCI_Event, packet))
elif packet.hci_packet_type == hci.HCI_ACL_DATA_PACKET:
self.on_hci_acl_data_packet(cast(hci.HCI_AclDataPacket, packet))
elif packet.hci_packet_type == hci.HCI_SYNCHRONOUS_DATA_PACKET:
self.on_hci_sco_data_packet(cast(hci.HCI_SynchronousDataPacket, packet))
elif packet.hci_packet_type == hci.HCI_ISO_DATA_PACKET:
self.on_hci_iso_data_packet(cast(hci.HCI_IsoDataPacket, packet))
else:
logger.warning(f'!!! unknown packet type {packet.hci_packet_type}')
def on_hci_command_packet(self, command: HCI_Command) -> None:
def on_hci_command_packet(self, command: hci.HCI_Command) -> None:
logger.warning(f'!!! unexpected command packet: {command}')
def on_hci_event_packet(self, event: HCI_Event) -> None:
def on_hci_event_packet(self, event: hci.HCI_Event) -> None:
handler_name = f'on_{event.name.lower()}'
handler = getattr(self, handler_name, self.on_hci_event)
handler(event)
def on_hci_acl_data_packet(self, packet: HCI_AclDataPacket) -> None:
def on_hci_acl_data_packet(self, packet: hci.HCI_AclDataPacket) -> None:
# Look for the connection to which this data belongs
if connection := self.connections.get(packet.connection_handle):
connection.on_hci_acl_data_packet(packet)
def on_hci_sco_data_packet(self, packet: HCI_SynchronousDataPacket) -> None:
def on_hci_sco_data_packet(self, packet: hci.HCI_SynchronousDataPacket) -> None:
# Experimental
self.emit('sco_packet', packet.connection_handle, packet)
def on_hci_iso_data_packet(self, packet: HCI_IsoDataPacket) -> None:
def on_hci_iso_data_packet(self, packet: hci.HCI_IsoDataPacket) -> None:
# Experimental
self.emit('iso_packet', packet.connection_handle, packet)
@@ -635,11 +740,11 @@ class Host(AbortableEventEmitter):
def on_hci_le_connection_complete_event(self, event):
# Check if this is a cancellation
if event.status == HCI_SUCCESS:
if event.status == hci.HCI_SUCCESS:
# Create/update the connection
logger.debug(
f'### LE CONNECTION: [0x{event.connection_handle:04X}] '
f'{event.peer_address} as {HCI_Constant.role_name(event.role)}'
f'{event.peer_address} as {hci.HCI_Constant.role_name(event.role)}'
)
connection = self.connections.get(event.connection_handle)
@@ -679,7 +784,7 @@ class Host(AbortableEventEmitter):
self.on_hci_le_connection_complete_event(event)
def on_hci_connection_complete_event(self, event):
if event.status == HCI_SUCCESS:
if event.status == hci.HCI_SUCCESS:
# Create/update the connection
logger.debug(
f'### BR/EDR CONNECTION: [0x{event.connection_handle:04X}] '
@@ -726,7 +831,7 @@ class Host(AbortableEventEmitter):
logger.warning('!!! DISCONNECTION COMPLETE: unknown handle')
return
if event.status == HCI_SUCCESS:
if event.status == hci.HCI_SUCCESS:
logger.debug(
f'### DISCONNECTION: [0x{handle:04X}] '
f'{connection.peer_address} '
@@ -735,7 +840,9 @@ class Host(AbortableEventEmitter):
# Notify the listeners
self.emit('disconnection', handle, event.reason)
(
# Remove the handle reference
_ = (
self.connections.pop(handle, 0)
or self.cis_links.pop(handle, 0)
or self.sco_links.pop(handle, 0)
@@ -752,7 +859,7 @@ class Host(AbortableEventEmitter):
return
# Notify the client
if event.status == HCI_SUCCESS:
if event.status == hci.HCI_SUCCESS:
connection_parameters = ConnectionParameters(
event.connection_interval,
event.peripheral_latency,
@@ -772,7 +879,7 @@ class Host(AbortableEventEmitter):
return
# Notify the client
if event.status == HCI_SUCCESS:
if event.status == hci.HCI_SUCCESS:
connection_phy = ConnectionPHY(event.tx_phy, event.rx_phy)
self.emit('connection_phy_update', connection.handle, connection_phy)
else:
@@ -791,6 +898,7 @@ class Host(AbortableEventEmitter):
event.status,
event.advertising_handle,
event.connection_handle,
event.num_completed_extended_advertising_events,
)
def on_hci_le_cis_request_event(self, event):
@@ -804,10 +912,10 @@ class Host(AbortableEventEmitter):
def on_hci_le_cis_established_event(self, event):
# The remaining parameters are unused for now.
if event.status == HCI_SUCCESS:
if event.status == hci.HCI_SUCCESS:
self.cis_links[event.connection_handle] = CisLink(
handle=event.connection_handle,
peer_address=Address.ANY,
peer_address=hci.Address.ANY,
)
self.emit('cis_establishment', event.connection_handle)
else:
@@ -823,7 +931,7 @@ class Host(AbortableEventEmitter):
# For now, just accept everything
# TODO: delegate the decision
self.send_command_sync(
HCI_LE_Remote_Connection_Parameter_Request_Reply_Command(
hci.HCI_LE_Remote_Connection_Parameter_Request_Reply_Command(
connection_handle=event.connection_handle,
interval_min=event.interval_min,
interval_max=event.interval_max,
@@ -854,12 +962,12 @@ class Host(AbortableEventEmitter):
),
)
if long_term_key:
response = HCI_LE_Long_Term_Key_Request_Reply_Command(
response = hci.HCI_LE_Long_Term_Key_Request_Reply_Command(
connection_handle=event.connection_handle,
long_term_key=long_term_key,
)
else:
response = HCI_LE_Long_Term_Key_Request_Negative_Reply_Command(
response = hci.HCI_LE_Long_Term_Key_Request_Negative_Reply_Command(
connection_handle=event.connection_handle
)
@@ -868,7 +976,7 @@ class Host(AbortableEventEmitter):
asyncio.create_task(send_long_term_key())
def on_hci_synchronous_connection_complete_event(self, event):
if event.status == HCI_SUCCESS:
if event.status == hci.HCI_SUCCESS:
# Create/update the connection
logger.debug(
f'### SCO CONNECTION: [0x{event.connection_handle:04X}] '
@@ -897,16 +1005,16 @@ class Host(AbortableEventEmitter):
pass
def on_hci_role_change_event(self, event):
if event.status == HCI_SUCCESS:
if event.status == hci.HCI_SUCCESS:
logger.debug(
f'role change for {event.bd_addr}: '
f'{HCI_Constant.role_name(event.new_role)}'
f'{hci.HCI_Constant.role_name(event.new_role)}'
)
self.emit('role_change', event.bd_addr, event.new_role)
else:
logger.debug(
f'role change for {event.bd_addr} failed: '
f'{HCI_Constant.error_name(event.status)}'
f'{hci.HCI_Constant.error_name(event.status)}'
)
self.emit('role_change_failure', event.bd_addr, event.status)
@@ -922,7 +1030,7 @@ class Host(AbortableEventEmitter):
def on_hci_authentication_complete_event(self, event):
# Notify the client
if event.status == HCI_SUCCESS:
if event.status == hci.HCI_SUCCESS:
self.emit('connection_authentication', event.connection_handle)
else:
self.emit(
@@ -933,7 +1041,7 @@ class Host(AbortableEventEmitter):
def on_hci_encryption_change_event(self, event):
# Notify the client
if event.status == HCI_SUCCESS:
if event.status == hci.HCI_SUCCESS:
self.emit(
'connection_encryption_change',
event.connection_handle,
@@ -946,7 +1054,7 @@ class Host(AbortableEventEmitter):
def on_hci_encryption_key_refresh_complete_event(self, event):
# Notify the client
if event.status == HCI_SUCCESS:
if event.status == hci.HCI_SUCCESS:
self.emit('connection_encryption_key_refresh', event.connection_handle)
else:
self.emit(
@@ -967,16 +1075,16 @@ class Host(AbortableEventEmitter):
def on_hci_link_key_notification_event(self, event):
logger.debug(
f'link key for {event.bd_addr}: {event.link_key.hex()}, '
f'type={HCI_Constant.link_key_type_name(event.key_type)}'
f'type={hci.HCI_Constant.link_key_type_name(event.key_type)}'
)
self.emit('link_key', event.bd_addr, event.link_key, event.key_type)
def on_hci_simple_pairing_complete_event(self, event):
logger.debug(
f'simple pairing complete for {event.bd_addr}: '
f'status={HCI_Constant.status_name(event.status)}'
f'status={hci.HCI_Constant.status_name(event.status)}'
)
if event.status == HCI_SUCCESS:
if event.status == hci.HCI_SUCCESS:
self.emit('classic_pairing', event.bd_addr)
else:
self.emit('classic_pairing_failure', event.bd_addr, event.status)
@@ -996,11 +1104,11 @@ class Host(AbortableEventEmitter):
self.link_key_provider(event.bd_addr),
)
if link_key:
response = HCI_Link_Key_Request_Reply_Command(
response = hci.HCI_Link_Key_Request_Reply_Command(
bd_addr=event.bd_addr, link_key=link_key
)
else:
response = HCI_Link_Key_Request_Negative_Reply_Command(
response = hci.HCI_Link_Key_Request_Negative_Reply_Command(
bd_addr=event.bd_addr
)
@@ -1057,7 +1165,7 @@ class Host(AbortableEventEmitter):
)
def on_hci_remote_name_request_complete_event(self, event):
if event.status != HCI_SUCCESS:
if event.status != hci.HCI_SUCCESS:
self.emit('remote_name_failure', event.bd_addr, event.status)
else:
utf8_name = event.remote_name
@@ -1075,7 +1183,7 @@ class Host(AbortableEventEmitter):
)
def on_hci_le_read_remote_features_complete_event(self, event):
if event.status != HCI_SUCCESS:
if event.status != hci.HCI_SUCCESS:
self.emit(
'le_remote_features_failure', event.connection_handle, event.status
)

View File

@@ -173,7 +173,7 @@ L2CAP_MTU_CONFIGURATION_PARAMETER_TYPE = 0x01
@dataclasses.dataclass
class ClassicChannelSpec:
psm: Optional[int] = None
mtu: int = L2CAP_MIN_BR_EDR_MTU
mtu: int = L2CAP_DEFAULT_MTU
@dataclasses.dataclass
@@ -208,7 +208,7 @@ class L2CAP_PDU:
@staticmethod
def from_bytes(data: bytes) -> L2CAP_PDU:
# Sanity check
# Check parameters
if len(data) < 4:
raise ValueError('not enough data for L2CAP header')
@@ -749,6 +749,8 @@ class ClassicChannel(EventEmitter):
sink: Optional[Callable[[bytes], Any]]
state: State
connection: Connection
mtu: int
peer_mtu: int
def __init__(
self,
@@ -765,6 +767,7 @@ class ClassicChannel(EventEmitter):
self.signaling_cid = signaling_cid
self.state = self.State.CLOSED
self.mtu = mtu
self.peer_mtu = L2CAP_MIN_BR_EDR_MTU
self.psm = psm
self.source_cid = source_cid
self.destination_cid = 0
@@ -861,7 +864,7 @@ class ClassicChannel(EventEmitter):
[
(
L2CAP_MAXIMUM_TRANSMISSION_UNIT_CONFIGURATION_OPTION_TYPE,
struct.pack('<H', L2CAP_DEFAULT_MTU),
struct.pack('<H', self.mtu),
)
]
)
@@ -926,8 +929,8 @@ class ClassicChannel(EventEmitter):
options = L2CAP_Control_Frame.decode_configuration_options(request.options)
for option in options:
if option[0] == L2CAP_MTU_CONFIGURATION_PARAMETER_TYPE:
self.mtu = struct.unpack('<H', option[1])[0]
logger.debug(f'MTU = {self.mtu}')
self.peer_mtu = struct.unpack('<H', option[1])[0]
logger.debug(f'peer MTU = {self.peer_mtu}')
self.send_control_frame(
L2CAP_Configure_Response(
@@ -1026,7 +1029,7 @@ class ClassicChannel(EventEmitter):
return (
f'Channel({self.source_cid}->{self.destination_cid}, '
f'PSM={self.psm}, '
f'MTU={self.mtu}, '
f'MTU={self.mtu}/{self.peer_mtu}, '
f'state={self.state.name})'
)

View File

@@ -34,8 +34,11 @@ from bumble.device import (
DEVICE_DEFAULT_SCAN_INTERVAL,
DEVICE_DEFAULT_SCAN_WINDOW,
Advertisement,
AdvertisingParameters,
AdvertisingEventProperties,
AdvertisingType,
Device,
Phy,
)
from bumble.gatt import Service
from bumble.hci import (
@@ -47,9 +50,12 @@ from bumble.hci import (
from google.protobuf import any_pb2 # pytype: disable=pyi-error
from google.protobuf import empty_pb2 # pytype: disable=pyi-error
from pandora.host_grpc_aio import HostServicer
from pandora import host_pb2
from pandora.host_pb2 import (
NOT_CONNECTABLE,
NOT_DISCOVERABLE,
DISCOVERABLE_LIMITED,
DISCOVERABLE_GENERAL,
PRIMARY_1M,
PRIMARY_CODED,
SECONDARY_1M,
@@ -65,6 +71,7 @@ from pandora.host_pb2 import (
ConnectResponse,
DataTypes,
DisconnectRequest,
DiscoverabilityMode,
InquiryResponse,
PrimaryPhy,
ReadLocalAddressResponse,
@@ -94,6 +101,25 @@ SECONDARY_PHY_MAP: Dict[int, SecondaryPhy] = {
3: SECONDARY_CODED,
}
PRIMARY_PHY_TO_BUMBLE_PHY_MAP: Dict[PrimaryPhy, Phy] = {
PRIMARY_1M: Phy.LE_1M,
PRIMARY_CODED: Phy.LE_CODED,
}
SECONDARY_PHY_TO_BUMBLE_PHY_MAP: Dict[SecondaryPhy, Phy] = {
SECONDARY_NONE: Phy.LE_1M,
SECONDARY_1M: Phy.LE_1M,
SECONDARY_2M: Phy.LE_2M,
SECONDARY_CODED: Phy.LE_CODED,
}
OWN_ADDRESS_MAP: Dict[host_pb2.OwnAddressType, bumble.hci.OwnAddressType] = {
host_pb2.PUBLIC: bumble.hci.OwnAddressType.PUBLIC,
host_pb2.RANDOM: bumble.hci.OwnAddressType.RANDOM,
host_pb2.RESOLVABLE_OR_PUBLIC: bumble.hci.OwnAddressType.RESOLVABLE_OR_PUBLIC,
host_pb2.RESOLVABLE_OR_RANDOM: bumble.hci.OwnAddressType.RESOLVABLE_OR_RANDOM,
}
class HostService(HostServicer):
waited_connections: Set[int]
@@ -281,10 +307,113 @@ class HostService(HostServicer):
async def Advertise(
self, request: AdvertiseRequest, context: grpc.ServicerContext
) -> AsyncGenerator[AdvertiseResponse, None]:
if not request.legacy:
raise NotImplementedError(
"TODO: add support for extended advertising in Bumble"
try:
if request.legacy:
async for rsp in self.legacy_advertise(request, context):
yield rsp
else:
async for rsp in self.extended_advertise(request, context):
yield rsp
finally:
pass
async def extended_advertise(
self, request: AdvertiseRequest, context: grpc.ServicerContext
) -> AsyncGenerator[AdvertiseResponse, None]:
advertising_data = bytes(self.unpack_data_types(request.data))
scan_response_data = bytes(self.unpack_data_types(request.scan_response_data))
scannable = len(scan_response_data) != 0
advertising_event_properties = AdvertisingEventProperties(
is_connectable=request.connectable,
is_scannable=scannable,
is_directed=request.target is not None,
is_high_duty_cycle_directed_connectable=False,
is_legacy=False,
is_anonymous=False,
include_tx_power=False,
)
peer_address = Address.ANY
if request.target:
# Need to reverse bytes order since Bumble Address is using MSB.
target_bytes = bytes(reversed(request.target))
if request.target_variant() == "public":
peer_address = Address(target_bytes, Address.PUBLIC_DEVICE_ADDRESS)
else:
peer_address = Address(target_bytes, Address.RANDOM_DEVICE_ADDRESS)
advertising_parameters = AdvertisingParameters(
advertising_event_properties=advertising_event_properties,
own_address_type=OWN_ADDRESS_MAP[request.own_address_type],
peer_address=peer_address,
primary_advertising_phy=PRIMARY_PHY_TO_BUMBLE_PHY_MAP[request.primary_phy],
secondary_advertising_phy=SECONDARY_PHY_TO_BUMBLE_PHY_MAP[
request.secondary_phy
],
)
if advertising_interval := request.interval:
advertising_parameters.primary_advertising_interval_min = int(
advertising_interval
)
advertising_parameters.primary_advertising_interval_max = int(
advertising_interval
)
if interval_range := request.interval_range:
advertising_parameters.primary_advertising_interval_max += int(
interval_range
)
advertising_set = await self.device.create_advertising_set(
advertising_parameters=advertising_parameters,
advertising_data=advertising_data,
scan_response_data=scan_response_data,
)
pending_connection: asyncio.Future[
bumble.device.Connection
] = asyncio.get_running_loop().create_future()
if request.connectable:
def on_connection(connection: bumble.device.Connection) -> None:
if (
connection.transport == BT_LE_TRANSPORT
and connection.role == BT_PERIPHERAL_ROLE
):
pending_connection.set_result(connection)
self.device.on('connection', on_connection)
try:
# Advertise until RPC is canceled
while True:
if not advertising_set.enabled:
self.log.debug('Advertise (extended)')
await advertising_set.start()
if not request.connectable:
await asyncio.sleep(1)
continue
connection = await pending_connection
pending_connection = asyncio.get_running_loop().create_future()
cookie = any_pb2.Any(value=connection.handle.to_bytes(4, 'big'))
yield AdvertiseResponse(connection=Connection(cookie=cookie))
await asyncio.sleep(1)
finally:
try:
self.log.debug('Stop Advertise (extended)')
await advertising_set.stop()
await advertising_set.remove()
except Exception:
pass
async def legacy_advertise(
self, request: AdvertiseRequest, context: grpc.ServicerContext
) -> AsyncGenerator[AdvertiseResponse, None]:
if advertising_interval := request.interval:
self.device.config.advertising_interval_min = int(advertising_interval)
self.device.config.advertising_interval_max = int(advertising_interval)
@@ -357,14 +486,10 @@ class HostService(HostServicer):
target_bytes = bytes(reversed(request.target))
if request.target_variant() == "public":
target = Address(target_bytes, Address.PUBLIC_DEVICE_ADDRESS)
advertising_type = (
AdvertisingType.DIRECTED_CONNECTABLE_HIGH_DUTY
) # FIXME: HIGH_DUTY ?
advertising_type = AdvertisingType.DIRECTED_CONNECTABLE_LOW_DUTY
else:
target = Address(target_bytes, Address.RANDOM_DEVICE_ADDRESS)
advertising_type = (
AdvertisingType.DIRECTED_CONNECTABLE_HIGH_DUTY
) # FIXME: HIGH_DUTY ?
advertising_type = AdvertisingType.DIRECTED_CONNECTABLE_LOW_DUTY
if request.connectable:
@@ -422,11 +547,16 @@ class HostService(HostServicer):
self, request: ScanRequest, context: grpc.ServicerContext
) -> AsyncGenerator[ScanningResponse, None]:
# TODO: modify `start_scanning` to accept floats instead of int for ms values
if request.phys:
raise NotImplementedError("TODO: add support for `request.phys`")
self.log.debug('Scan')
scanning_phys = []
if PRIMARY_1M in request.phys:
scanning_phys.append(int(Phy.LE_1M))
if PRIMARY_CODED in request.phys:
scanning_phys.append(int(Phy.LE_CODED))
if not scanning_phys:
scanning_phys = [int(Phy.LE_1M), int(Phy.LE_CODED)]
scan_queue: asyncio.Queue[Advertisement] = asyncio.Queue()
handler = self.device.on('advertisement', scan_queue.put_nowait)
await self.device.start_scanning(
@@ -439,6 +569,7 @@ class HostService(HostServicer):
scan_window=int(request.window)
if request.window
else DEVICE_DEFAULT_SCAN_WINDOW,
scanning_phys=scanning_phys,
)
try:
@@ -735,6 +866,16 @@ class HostService(HostServicer):
)
)
flag_map = {
NOT_DISCOVERABLE: 0x00,
DISCOVERABLE_LIMITED: AdvertisingData.LE_LIMITED_DISCOVERABLE_MODE_FLAG,
DISCOVERABLE_GENERAL: AdvertisingData.LE_GENERAL_DISCOVERABLE_MODE_FLAG,
}
if dt.le_discoverability_mode:
flags = flag_map[dt.le_discoverability_mode]
ad_structures.append((AdvertisingData.FLAGS, flags.to_bytes(1, 'big')))
return AdvertisingData(ad_structures)
def pack_data_types(self, ad: AdvertisingData) -> DataTypes:

228
bumble/profiles/vcp.py Normal file
View File

@@ -0,0 +1,228 @@
# Copyright 2021-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.
# -----------------------------------------------------------------------------
# Imports
# -----------------------------------------------------------------------------
from __future__ import annotations
import enum
from bumble import att
from bumble import device
from bumble import gatt
from bumble import gatt_client
from typing import Optional
# -----------------------------------------------------------------------------
# Constants
# -----------------------------------------------------------------------------
MIN_VOLUME = 0
MAX_VOLUME = 255
class ErrorCode(enum.IntEnum):
'''
See Volume Control Service 1.6. Application error codes.
'''
INVALID_CHANGE_COUNTER = 0x80
OPCODE_NOT_SUPPORTED = 0x81
class VolumeFlags(enum.IntFlag):
'''
See Volume Control Service 3.3. Volume Flags.
'''
VOLUME_SETTING_PERSISTED = 0x01
# RFU
class VolumeControlPointOpcode(enum.IntEnum):
'''
See Volume Control Service Table 3.3: Volume Control Point procedure requirements.
'''
# fmt: off
RELATIVE_VOLUME_DOWN = 0x00
RELATIVE_VOLUME_UP = 0x01
UNMUTE_RELATIVE_VOLUME_DOWN = 0x02
UNMUTE_RELATIVE_VOLUME_UP = 0x03
SET_ABSOLUTE_VOLUME = 0x04
UNMUTE = 0x05
MUTE = 0x06
# -----------------------------------------------------------------------------
# Server
# -----------------------------------------------------------------------------
class VolumeControlService(gatt.TemplateService):
UUID = gatt.GATT_VOLUME_CONTROL_SERVICE
volume_state: gatt.Characteristic
volume_control_point: gatt.Characteristic
volume_flags: gatt.Characteristic
volume_setting: int
muted: int
change_counter: int
def __init__(
self,
step_size: int = 16,
volume_setting: int = 0,
muted: int = 0,
change_counter: int = 0,
volume_flags: int = 0,
) -> None:
self.step_size = step_size
self.volume_setting = volume_setting
self.muted = muted
self.change_counter = change_counter
self.volume_state = gatt.Characteristic(
uuid=gatt.GATT_VOLUME_STATE_CHARACTERISTIC,
properties=(
gatt.Characteristic.Properties.READ
| gatt.Characteristic.Properties.NOTIFY
),
permissions=gatt.Characteristic.Permissions.READ_REQUIRES_ENCRYPTION,
value=gatt.CharacteristicValue(read=self._on_read_volume_state),
)
self.volume_control_point = gatt.Characteristic(
uuid=gatt.GATT_VOLUME_CONTROL_POINT_CHARACTERISTIC,
properties=gatt.Characteristic.Properties.WRITE,
permissions=gatt.Characteristic.Permissions.WRITE_REQUIRES_ENCRYPTION,
value=gatt.CharacteristicValue(write=self._on_write_volume_control_point),
)
self.volume_flags = gatt.Characteristic(
uuid=gatt.GATT_VOLUME_FLAGS_CHARACTERISTIC,
properties=gatt.Characteristic.Properties.READ,
permissions=gatt.Characteristic.Permissions.READ_REQUIRES_ENCRYPTION,
value=bytes([volume_flags]),
)
super().__init__(
[
self.volume_state,
self.volume_control_point,
self.volume_flags,
]
)
@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
def _on_write_volume_control_point(
self, connection: Optional[device.Connection], value: bytes
) -> None:
assert connection
opcode = VolumeControlPointOpcode(value[0])
change_counter = value[1]
if change_counter != self.change_counter:
raise att.ATT_Error(ErrorCode.INVALID_CHANGE_COUNTER)
handler = getattr(self, '_on_' + opcode.name.lower())
if handler(*value[2:]):
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
)
def _on_relative_volume_down(self) -> bool:
old_volume = self.volume_setting
self.volume_setting = max(self.volume_setting - self.step_size, MIN_VOLUME)
return self.volume_setting != old_volume
def _on_relative_volume_up(self) -> bool:
old_volume = self.volume_setting
self.volume_setting = min(self.volume_setting + self.step_size, MAX_VOLUME)
return self.volume_setting != old_volume
def _on_unmute_relative_volume_down(self) -> bool:
old_volume, old_muted_state = self.volume_setting, self.muted
self.volume_setting = max(self.volume_setting - self.step_size, MIN_VOLUME)
self.muted = 0
return (self.volume_setting, self.muted) != (old_volume, old_muted_state)
def _on_unmute_relative_volume_up(self) -> bool:
old_volume, old_muted_state = self.volume_setting, self.muted
self.volume_setting = min(self.volume_setting + self.step_size, MAX_VOLUME)
self.muted = 0
return (self.volume_setting, self.muted) != (old_volume, old_muted_state)
def _on_set_absolute_volume(self, volume_setting: int) -> bool:
old_volume_setting = self.volume_setting
self.volume_setting = volume_setting
return old_volume_setting != self.volume_setting
def _on_unmute(self) -> bool:
old_muted_state = self.muted
self.muted = 0
return self.muted != old_muted_state
def _on_mute(self) -> bool:
old_muted_state = self.muted
self.muted = 1
return self.muted != old_muted_state
# -----------------------------------------------------------------------------
# Client
# -----------------------------------------------------------------------------
class VolumeControlServiceProxy(gatt_client.ProfileServiceProxy):
SERVICE_CLASS = VolumeControlService
volume_control_point: gatt_client.CharacteristicProxy
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(
gatt.GATT_VOLUME_STATE_CHARACTERISTIC
)[0],
'BBB',
)
self.volume_control_point = service_proxy.get_characteristics_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',
)

View File

@@ -22,10 +22,13 @@ import asyncio
import dataclasses
import enum
from typing import Callable, Dict, List, Optional, Tuple, Union, TYPE_CHECKING
from typing_extensions import Self
from pyee import EventEmitter
from . import core, l2cap
from bumble import core
from bumble import l2cap
from bumble import sdp
from .colors import color
from .core import (
UUID,
@@ -35,15 +38,6 @@ from .core import (
InvalidStateError,
ProtocolError,
)
from .sdp import (
SDP_SERVICE_RECORD_HANDLE_ATTRIBUTE_ID,
SDP_BROWSE_GROUP_LIST_ATTRIBUTE_ID,
SDP_SERVICE_CLASS_ID_LIST_ATTRIBUTE_ID,
SDP_PROTOCOL_DESCRIPTOR_LIST_ATTRIBUTE_ID,
SDP_PUBLIC_BROWSE_ROOT,
DataElement,
ServiceAttribute,
)
if TYPE_CHECKING:
from bumble.device import Device, Connection
@@ -110,6 +104,7 @@ CRC_TABLE = bytes([
0XBA, 0X2B, 0X59, 0XC8, 0XBD, 0X2C, 0X5E, 0XCF
])
RFCOMM_DEFAULT_L2CAP_MTU = 2048
RFCOMM_DEFAULT_WINDOW_SIZE = 7
RFCOMM_DEFAULT_MAX_FRAME_SIZE = 2000
@@ -122,29 +117,33 @@ RFCOMM_DYNAMIC_CHANNEL_NUMBER_END = 30
# -----------------------------------------------------------------------------
def make_service_sdp_records(
service_record_handle: int, channel: int, uuid: Optional[UUID] = None
) -> List[ServiceAttribute]:
) -> List[sdp.ServiceAttribute]:
"""
Create SDP records for an RFComm service given a channel number and an
optional UUID. A Service Class Attribute is included only if the UUID is not None.
"""
records = [
ServiceAttribute(
SDP_SERVICE_RECORD_HANDLE_ATTRIBUTE_ID,
DataElement.unsigned_integer_32(service_record_handle),
sdp.ServiceAttribute(
sdp.SDP_SERVICE_RECORD_HANDLE_ATTRIBUTE_ID,
sdp.DataElement.unsigned_integer_32(service_record_handle),
),
ServiceAttribute(
SDP_BROWSE_GROUP_LIST_ATTRIBUTE_ID,
DataElement.sequence([DataElement.uuid(SDP_PUBLIC_BROWSE_ROOT)]),
sdp.ServiceAttribute(
sdp.SDP_BROWSE_GROUP_LIST_ATTRIBUTE_ID,
sdp.DataElement.sequence(
[sdp.DataElement.uuid(sdp.SDP_PUBLIC_BROWSE_ROOT)]
),
),
ServiceAttribute(
SDP_PROTOCOL_DESCRIPTOR_LIST_ATTRIBUTE_ID,
DataElement.sequence(
sdp.ServiceAttribute(
sdp.SDP_PROTOCOL_DESCRIPTOR_LIST_ATTRIBUTE_ID,
sdp.DataElement.sequence(
[
DataElement.sequence([DataElement.uuid(BT_L2CAP_PROTOCOL_ID)]),
DataElement.sequence(
sdp.DataElement.sequence(
[sdp.DataElement.uuid(BT_L2CAP_PROTOCOL_ID)]
),
sdp.DataElement.sequence(
[
DataElement.uuid(BT_RFCOMM_PROTOCOL_ID),
DataElement.unsigned_integer_8(channel),
sdp.DataElement.uuid(BT_RFCOMM_PROTOCOL_ID),
sdp.DataElement.unsigned_integer_8(channel),
]
),
]
@@ -154,15 +153,81 @@ def make_service_sdp_records(
if uuid:
records.append(
ServiceAttribute(
SDP_SERVICE_CLASS_ID_LIST_ATTRIBUTE_ID,
DataElement.sequence([DataElement.uuid(uuid)]),
sdp.ServiceAttribute(
sdp.SDP_SERVICE_CLASS_ID_LIST_ATTRIBUTE_ID,
sdp.DataElement.sequence([sdp.DataElement.uuid(uuid)]),
)
)
return records
# -----------------------------------------------------------------------------
async def find_rfcomm_channels(connection: Connection) -> Dict[int, List[UUID]]:
"""Searches all RFCOMM channels and their associated UUID from SDP service records.
Args:
connection: ACL connection to make SDP search.
Returns:
Dictionary mapping from channel number to service class UUID list.
"""
results = {}
async with sdp.Client(connection) as sdp_client:
search_result = await sdp_client.search_attributes(
uuids=[core.BT_RFCOMM_PROTOCOL_ID],
attribute_ids=[
sdp.SDP_PROTOCOL_DESCRIPTOR_LIST_ATTRIBUTE_ID,
sdp.SDP_SERVICE_CLASS_ID_LIST_ATTRIBUTE_ID,
],
)
for attribute_lists in search_result:
service_classes: List[UUID] = []
channel: Optional[int] = None
for attribute in attribute_lists:
# The layout is [[L2CAP_PROTOCOL], [RFCOMM_PROTOCOL, RFCOMM_CHANNEL]].
if attribute.id == sdp.SDP_PROTOCOL_DESCRIPTOR_LIST_ATTRIBUTE_ID:
protocol_descriptor_list = attribute.value.value
channel = protocol_descriptor_list[1].value[1].value
elif attribute.id == sdp.SDP_SERVICE_CLASS_ID_LIST_ATTRIBUTE_ID:
service_class_id_list = attribute.value.value
service_classes = [
service_class.value for service_class in service_class_id_list
]
if not service_classes or not channel:
logger.warning(f"Bad result {attribute_lists}.")
else:
results[channel] = service_classes
return results
# -----------------------------------------------------------------------------
async def find_rfcomm_channel_with_uuid(
connection: Connection, uuid: str | UUID
) -> Optional[int]:
"""Searches an RFCOMM channel associated with given UUID from service records.
Args:
connection: ACL connection to make SDP search.
uuid: UUID of service record to search for.
Returns:
RFCOMM channel number if found, otherwise None.
"""
if isinstance(uuid, str):
uuid = UUID(uuid)
return next(
(
channel
for channel, class_id_list in (
await find_rfcomm_channels(connection)
).items()
if uuid in class_id_list
),
None,
)
# -----------------------------------------------------------------------------
def compute_fcs(buffer: bytes) -> int:
result = 0xFF
@@ -409,7 +474,7 @@ class DLC(EventEmitter):
# Compute the MTU
max_overhead = 4 + 1 # header with 2-byte length + fcs
self.mtu = min(
max_frame_size, self.multiplexer.l2cap_channel.mtu - max_overhead
max_frame_size, self.multiplexer.l2cap_channel.peer_mtu - max_overhead
)
def change_state(self, new_state: State) -> None:
@@ -844,8 +909,11 @@ class Client:
multiplexer: Optional[Multiplexer]
l2cap_channel: Optional[l2cap.ClassicChannel]
def __init__(self, connection: Connection) -> None:
def __init__(
self, connection: Connection, l2cap_mtu: int = RFCOMM_DEFAULT_L2CAP_MTU
) -> None:
self.connection = connection
self.l2cap_mtu = l2cap_mtu
self.l2cap_channel = None
self.multiplexer = None
@@ -853,7 +921,7 @@ class Client:
# Create a new L2CAP connection
try:
self.l2cap_channel = await self.connection.create_l2cap_channel(
spec=l2cap.ClassicChannelSpec(RFCOMM_PSM)
spec=l2cap.ClassicChannelSpec(psm=RFCOMM_PSM, mtu=self.l2cap_mtu)
)
except ProtocolError as error:
logger.warning(f'L2CAP connection failed: {error}')
@@ -876,22 +944,33 @@ class Client:
self.multiplexer = None
# Close the L2CAP channel
# TODO
if self.l2cap_channel:
await self.l2cap_channel.disconnect()
self.l2cap_channel = None
async def __aenter__(self) -> Multiplexer:
return await self.start()
async def __aexit__(self, *args) -> None:
await self.shutdown()
# -----------------------------------------------------------------------------
class Server(EventEmitter):
acceptors: Dict[int, Callable[[DLC], None]]
def __init__(self, device: Device) -> None:
def __init__(
self, device: Device, l2cap_mtu: int = RFCOMM_DEFAULT_L2CAP_MTU
) -> None:
super().__init__()
self.device = device
self.multiplexer = None
self.acceptors = {}
# Register ourselves with the L2CAP channel manager
device.create_l2cap_server(
spec=l2cap.ClassicChannelSpec(psm=RFCOMM_PSM), handler=self.on_connection
self.l2cap_server = device.create_l2cap_server(
spec=l2cap.ClassicChannelSpec(psm=RFCOMM_PSM, mtu=l2cap_mtu),
handler=self.on_connection,
)
def listen(self, acceptor: Callable[[DLC], None], channel: int = 0) -> int:
@@ -941,3 +1020,9 @@ class Server(EventEmitter):
acceptor = self.acceptors.get(dlc.dlci >> 1)
if acceptor:
acceptor(dlc)
def __enter__(self) -> Self:
return self
def __exit__(self, *args) -> None:
self.l2cap_server.close()

View File

@@ -19,6 +19,7 @@ from __future__ import annotations
import logging
import struct
from typing import Dict, List, Type, Optional, Tuple, Union, NewType, TYPE_CHECKING
from typing_extensions import Self
from . import core, l2cap
from .colors import color
@@ -920,6 +921,13 @@ class Client:
return ServiceAttribute.list_from_data_elements(attribute_list_sequence.value)
async def __aenter__(self) -> Self:
await self.connect()
return self
async def __aexit__(self, *args) -> None:
await self.disconnect()
# -----------------------------------------------------------------------------
class Server:

View File

@@ -168,11 +168,13 @@ class PacketReader:
def __init__(self, source: io.BufferedReader) -> None:
self.source = source
self.at_end = False
def next_packet(self) -> Optional[bytes]:
# Get the packet type
packet_type = self.source.read(1)
if len(packet_type) != 1:
self.at_end = True
return None
# Get the packet info based on its type

View File

@@ -113,9 +113,10 @@ async def open_pyusb_transport(spec: str) -> Transport:
self.loop.call_soon_threadsafe(self.stop_event.set)
class UsbPacketSource(asyncio.Protocol, ParserSource):
def __init__(self, device, sco_enabled):
def __init__(self, device, metadata, sco_enabled):
super().__init__()
self.device = device
self.metadata = metadata
self.loop = asyncio.get_running_loop()
self.queue = asyncio.Queue()
self.dequeue_task = None
@@ -216,6 +217,15 @@ async def open_pyusb_transport(spec: str) -> Transport:
if ':' in spec:
vendor_id, product_id = spec.split(':')
device = usb_find(idVendor=int(vendor_id, 16), idProduct=int(product_id, 16))
elif '-' in spec:
def device_path(device):
if device.port_numbers:
return f'{device.bus}-{".".join(map(str, device.port_numbers))}'
else:
return str(device.bus)
device = usb_find(custom_match=lambda device: device_path(device) == spec)
else:
device_index = int(spec)
devices = list(
@@ -235,6 +245,9 @@ async def open_pyusb_transport(spec: str) -> Transport:
raise ValueError('device not found')
logger.debug(f'USB Device: {device}')
# Collect the metadata
device_metadata = {'vendor_id': device.idVendor, 'product_id': device.idProduct}
# Detach the kernel driver if needed
if device.is_kernel_driver_active(0):
logger.debug("detaching kernel driver")
@@ -289,7 +302,7 @@ async def open_pyusb_transport(spec: str) -> Transport:
# except usb.USBError:
# logger.warning('failed to set alternate setting')
packet_source = UsbPacketSource(device, sco_enabled)
packet_source = UsbPacketSource(device, device_metadata, sco_enabled)
packet_sink = UsbPacketSink(device)
packet_source.start()
packet_sink.start()

View File

@@ -396,6 +396,16 @@ async def open_usb_transport(spec: str) -> Transport:
break
device_index -= 1
device.close()
elif '-' in spec:
def device_path(device):
return f'{device.getBusNumber()}-{".".join(map(str, device.getPortNumberList()))}'
for device in context.getDeviceIterator(skip_on_error=True):
if device_path(device) == spec:
found = device
break
device.close()
else:
# Look for a compatible device by index
def device_is_bluetooth_hci(device):

View File

@@ -226,13 +226,13 @@ class CompositeEventEmitter(AbortableEventEmitter):
if self._listener:
# Call the deregistration methods for each base class that has them
for cls in self._listener.__class__.mro():
if hasattr(cls, '_bumble_register_composite'):
cls._bumble_deregister_composite(listener, self)
if '_bumble_register_composite' in cls.__dict__:
cls._bumble_deregister_composite(self._listener, self)
self._listener = listener
if listener:
# Call the registration methods for each base class that has them
for cls in listener.__class__.mro():
if hasattr(cls, '_bumble_deregister_composite'):
if '_bumble_deregister_composite' in cls.__dict__:
cls._bumble_register_composite(listener, self)

View File

@@ -12,12 +12,25 @@ a host that send custom HCI commands that the controller may not understand.
```
python hci_bridge.py <host-transport-spec> <controller-transport-spec> [command-short-circuit-list]
```
The command-short-circuit-list field is specified by a series of comma separated Opcode Group
Field (OGF) : OpCode Command Field (OCF) pairs. The OGF/OCF values are specified in the Blutooth
core specification.
For the commands that are listed in the short-circuit-list, the HCI bridge will always generate
a Command Complete Event for the specified op code. The return parameter will be HCI_SUCCESS.
This feature can only be used for commands that return Command Complete. Other events will not be
generated by the HCI bridge tool.
!!! example "UDP to Serial"
```
python hci_bridge.py udp:0.0.0.0:9000,127.0.0.1:9001 serial:/dev/tty.usbmodem0006839912171,1000000 0x3f:0x0070,0x3f:0x0074,0x3f:0x0077,0x3f:0x0078
```
In this example, the short circuit list is specified to respond to the Vendor-specific Opcode Group
Field (0x3f) commands 0x70, 0x74, 0x77, 0x78 with Command Complete. The short circuit list can be
used where the Host uses some HCI commands that are not supported/implemented by the Controller.
!!! example "PTY to Link Relay"
```
python hci_bridge.py serial:emulated_uart_pty,1000000 link-relay:ws://127.0.0.1:10723/test
@@ -28,3 +41,4 @@ a host that send custom HCI commands that the controller may not understand.
(through which the communication with other virtual controllers will be mediated).
NOTE: this assumes you're running a Link Relay on port `10723`.

View File

@@ -10,7 +10,7 @@ used with particular HCI controller.
When the transport for an HCI controller is instantiated from a transport name,
a driver may also be forced by specifying ``driver=<driver-name>`` in the optional
metadata portion of the transport name. For example,
``usb:[driver=-rtk]0`` indicates that the ``rtk`` driver should be used with the
``usb:[driver=rtk]0`` indicates that the ``rtk`` driver should be used with the
first USB device, even if a normal probe would not have selected it based on the
USB vendor ID and product ID.

View File

@@ -10,6 +10,7 @@ The moniker for a USB transport is either:
* `usb:<vendor>:<product>`
* `usb:<vendor>:<product>/<serial-number>`
* `usb:<vendor>:<product>#<index>`
* `usb:<bus>-<port_numbers>`
with `<index>` as a 0-based index (0 being the first one) to select amongst all the matching devices when there are more than one.
In the `usb:<index>` form, matching devices are the ones supporting Bluetooth HCI, as declared by their Class, Subclass and Protocol.
@@ -17,6 +18,8 @@ In the `usb:<vendor>:<product>#<index>` form, matching devices are the ones with
`<vendor>` and `<product>` are a vendor ID and product ID in hexadecimal.
with `<port_numbers>` as a list of all port numbers from root separated with dots `.`
In addition, if the moniker ends with the symbol "!", the device will be used in "forced" mode:
the first USB interface of the device will be used, regardless of the interface class/subclass.
This may be useful for some devices that use a custom class/subclass but may nonetheless work as-is.
@@ -37,6 +40,9 @@ This may be useful for some devices that use a custom class/subclass but may non
`usb:0B05:17CB!`
The BT USB dongle vendor=0B05 and product=17CB, in "forced" mode.
`usb:3-3.4.1`
The BT USB dongle on bus 3 on port path 3, 4, 1.
## Alternative
The library includes two different implementations of the USB transport, implemented using different python bindings for `libusb`.

View File

@@ -2,5 +2,6 @@
"name": "Bumble-LEA",
"keystore": "JsonKeyStore",
"address": "F0:F1:F2:F3:F4:FA",
"class_of_device": 2376708,
"advertising_interval": 100
}

View File

@@ -0,0 +1,9 @@
{
"name": "Bumble-LEA",
"keystore": "JsonKeyStore",
"address": "F0:F1:F2:F3:F4:FA",
"classic_enabled": true,
"cis_enabled": true,
"class_of_device": 2376708,
"advertising_interval": 100
}

View File

@@ -74,7 +74,7 @@ def codec_capabilities():
# -----------------------------------------------------------------------------
def on_avdtp_connection(read_function, protocol):
packet_source = SbcPacketSource(
read_function, protocol.l2cap_channel.mtu, codec_capabilities()
read_function, protocol.l2cap_channel.peer_mtu, codec_capabilities()
)
packet_pump = MediaPacketPump(packet_source.packets)
protocol.add_source(packet_source.codec_capabilities, packet_pump)
@@ -98,7 +98,7 @@ async def stream_packets(read_function, protocol):
# Stream the packets
packet_source = SbcPacketSource(
read_function, protocol.l2cap_channel.mtu, codec_capabilities()
read_function, protocol.l2cap_channel.peer_mtu, codec_capabilities()
)
packet_pump = MediaPacketPump(packet_source.packets)
source = protocol.add_source(packet_source.codec_capabilities, packet_pump)

View File

@@ -19,9 +19,11 @@ import asyncio
import logging
import sys
import os
import struct
from bumble.core import AdvertisingData
from bumble.device import AdvertisingType, Device
from bumble.hci import Address
from bumble.transport import open_transport_or_link
@@ -52,6 +54,16 @@ async def main():
print('<<< connected')
device = Device.from_config_file_with_hci(sys.argv[1], hci_source, hci_sink)
if advertising_type.is_scannable:
device.scan_response_data = bytes(
AdvertisingData(
[
(AdvertisingData.APPEARANCE, struct.pack('<H', 0x0340)),
]
)
)
await device.power_on()
await device.start_advertising(advertising_type=advertising_type, target=target)
await hci_source.wait_for_termination()

View File

@@ -22,10 +22,11 @@ import os
from bumble.device import (
Device,
Connection,
AdvertisingParameters,
AdvertisingEventProperties,
)
from bumble.hci import (
OwnAddressType,
HCI_LE_Set_Extended_Advertising_Parameters_Command,
)
from bumble.transport import open_transport_or_link
@@ -61,12 +62,7 @@ async def main() -> None:
devices[1].cis_enabled = True
await asyncio.gather(*[device.power_on() for device in devices])
await devices[0].start_extended_advertising(
advertising_properties=(
HCI_LE_Set_Extended_Advertising_Parameters_Command.AdvertisingProperties.CONNECTABLE_ADVERTISING
),
own_address_type=OwnAddressType.PUBLIC,
)
advertising_set = await devices[0].create_advertising_set()
connection = await devices[1].connect(
devices[0].public_address, own_address_type=OwnAddressType.PUBLIC

View File

@@ -98,13 +98,7 @@ async def main() -> None:
)
+ csis.get_advertising_data()
)
await device.start_extended_advertising(
advertising_properties=(
HCI_LE_Set_Extended_Advertising_Parameters_Command.AdvertisingProperties.CONNECTABLE_ADVERTISING
),
own_address_type=OwnAddressType.RANDOM,
advertising_data=advertising_data,
)
await device.create_advertising_set(advertising_data=advertising_data)
await asyncio.gather(
*[hci_transport.source.terminated for hci_transport in hci_transports]

View File

@@ -19,8 +19,13 @@ import asyncio
import logging
import sys
import os
from bumble.device import AdvertisingType, Device
from bumble.hci import Address, HCI_LE_Set_Extended_Advertising_Parameters_Command
from bumble.device import (
AdvertisingParameters,
AdvertisingEventProperties,
AdvertisingType,
Device,
)
from bumble.hci import Address
from bumble.transport import open_transport_or_link
@@ -35,20 +40,16 @@ async def main() -> None:
return
if len(sys.argv) >= 4:
advertising_properties = (
HCI_LE_Set_Extended_Advertising_Parameters_Command.AdvertisingProperties(
int(sys.argv[3])
)
advertising_properties = AdvertisingEventProperties.from_advertising_type(
AdvertisingType(int(sys.argv[3]))
)
else:
advertising_properties = (
HCI_LE_Set_Extended_Advertising_Parameters_Command.AdvertisingProperties.CONNECTABLE_ADVERTISING
)
advertising_properties = AdvertisingEventProperties()
if len(sys.argv) >= 5:
target = Address(sys.argv[4])
peer_address = Address(sys.argv[4])
else:
target = Address.ANY
peer_address = Address.ANY
print('<<< connecting to HCI...')
async with await open_transport_or_link(sys.argv[2]) as hci_transport:
@@ -58,8 +59,11 @@ async def main() -> None:
sys.argv[1], hci_transport.source, hci_transport.sink
)
await device.power_on()
await device.start_extended_advertising(
advertising_properties=advertising_properties, target=target
await device.create_advertising_set(
advertising_parameters=AdvertisingParameters(
advertising_event_properties=advertising_properties,
peer_address=peer_address,
)
)
await hci_transport.source.terminated

View File

@@ -0,0 +1,99 @@
# Copyright 2021-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.
# -----------------------------------------------------------------------------
# Imports
# -----------------------------------------------------------------------------
import asyncio
import logging
import sys
import os
from bumble.device import AdvertisingParameters, AdvertisingEventProperties, Device
from bumble.hci import Address
from bumble.core import AdvertisingData
from bumble.transport import open_transport_or_link
# -----------------------------------------------------------------------------
async def main() -> None:
if len(sys.argv) < 3:
print('Usage: run_extended_advertiser_2.py <config-file> <transport-spec>')
print('example: run_extended_advertiser_2.py device1.json usb:0')
return
print('<<< connecting to HCI...')
async with await open_transport_or_link(sys.argv[2]) as hci_transport:
print('<<< connected')
device = Device.from_config_file_with_hci(
sys.argv[1], hci_transport.source, hci_transport.sink
)
await device.power_on()
if not device.supports_le_extended_advertising:
print("Device does not support extended advertising")
return
print("Max advertising sets:", device.host.number_of_supported_advertising_sets)
print(
"Max advertising data length:", device.host.maximum_advertising_data_length
)
if device.host.number_of_supported_advertising_sets >= 1:
advertising_data1 = AdvertisingData(
[(AdvertisingData.COMPLETE_LOCAL_NAME, "Bumble 1".encode("utf-8"))]
)
set1 = await device.create_advertising_set(
advertising_data=bytes(advertising_data1),
)
print("Selected TX power 1:", set1.selected_tx_power)
advertising_data2 = AdvertisingData(
[(AdvertisingData.COMPLETE_LOCAL_NAME, "Bumble 2".encode("utf-8"))]
)
if device.host.number_of_supported_advertising_sets >= 2:
set2 = await device.create_advertising_set(
random_address=Address("F0:F0:F0:F0:F0:F1"),
advertising_parameters=AdvertisingParameters(),
advertising_data=bytes(advertising_data2),
auto_start=False,
auto_restart=True,
)
print("Selected TX power 2:", set2.selected_tx_power)
await set2.start()
if device.host.number_of_supported_advertising_sets >= 3:
scan_response_data3 = AdvertisingData(
[(AdvertisingData.COMPLETE_LOCAL_NAME, "Bumble 3".encode("utf-8"))]
)
set3 = await device.create_advertising_set(
random_address=Address("F0:F0:F0:F0:F0:F2"),
advertising_parameters=AdvertisingParameters(
advertising_event_properties=AdvertisingEventProperties(
is_connectable=False, is_scannable=True
)
),
scan_response_data=bytes(scan_response_data3),
)
print("Selected TX power 3:", set2.selected_tx_power)
await hci_transport.source.terminated
# -----------------------------------------------------------------------------
logging.basicConfig(level=os.environ.get('BUMBLE_LOGLEVEL', 'DEBUG').upper())
asyncio.run(main())

View File

@@ -21,11 +21,13 @@ import os
import logging
import json
import websockets
import functools
from typing import Optional
from bumble.device import Device
from bumble import rfcomm
from bumble import hci
from bumble.device import Device, Connection
from bumble.transport import open_transport_or_link
from bumble.rfcomm import Server as RfcommServer
from bumble import hfp
from bumble.hfp import HfProtocol
@@ -57,12 +59,44 @@ class UiServer:
# -----------------------------------------------------------------------------
def on_dlc(dlc, configuration: hfp.Configuration):
def on_dlc(dlc: rfcomm.DLC, configuration: hfp.Configuration):
print('*** DLC connected', dlc)
protocol = HfProtocol(dlc, configuration)
UiServer.protocol = protocol
asyncio.create_task(protocol.run())
def on_sco_request(connection: Connection, link_type: int, protocol: HfProtocol):
if connection == protocol.dlc.multiplexer.l2cap_channel.connection:
if link_type == hci.HCI_Connection_Complete_Event.SCO_LINK_TYPE:
esco_parameters = hfp.ESCO_PARAMETERS[
hfp.DefaultCodecParameters.SCO_CVSD_D1
]
elif protocol.active_codec == hfp.AudioCodec.MSBC:
esco_parameters = hfp.ESCO_PARAMETERS[
hfp.DefaultCodecParameters.ESCO_MSBC_T2
]
elif protocol.active_codec == hfp.AudioCodec.CVSD:
esco_parameters = hfp.ESCO_PARAMETERS[
hfp.DefaultCodecParameters.ESCO_CVSD_S4
]
connection.abort_on(
'disconnection',
connection.device.send_command(
hci.HCI_Enhanced_Accept_Synchronous_Connection_Request_Command(
bd_addr=connection.peer_address, **esco_parameters.asdict()
)
),
)
handler = functools.partial(on_sco_request, protocol=protocol)
dlc.multiplexer.l2cap_channel.connection.device.on('sco_request', handler)
dlc.multiplexer.l2cap_channel.once(
'close',
lambda: dlc.multiplexer.l2cap_channel.connection.device.remove_listener(
'sco_request', handler
),
)
# -----------------------------------------------------------------------------
async def main():
@@ -101,7 +135,7 @@ async def main():
device.classic_enabled = True
# Create and register a server
rfcomm_server = RfcommServer(device)
rfcomm_server = rfcomm.Server(device)
# Listen for incoming DLC connections
channel_number = rfcomm_server.listen(lambda dlc: on_dlc(dlc, configuration))

View File

@@ -22,13 +22,12 @@ import os
import struct
import secrets
from bumble.core import AdvertisingData
from bumble.device import Device, CisLink
from bumble.device import Device, CisLink, AdvertisingParameters
from bumble.hci import (
CodecID,
CodingFormat,
OwnAddressType,
HCI_IsoDataPacket,
HCI_LE_Set_Extended_Advertising_Parameters_Command,
)
from bumble.profiles.bap import (
CodecSpecificCapabilities,
@@ -99,14 +98,14 @@ async def main() -> None:
coding_format=CodingFormat(CodecID.LC3),
codec_specific_capabilities=CodecSpecificCapabilities(
supported_sampling_frequencies=(
SupportedSamplingFrequency.FREQ_24000
SupportedSamplingFrequency.FREQ_48000
),
supported_frame_durations=(
SupportedFrameDuration.DURATION_10000_US_SUPPORTED
),
supported_audio_channel_counts=[1],
min_octets_per_codec_frame=60,
max_octets_per_codec_frame=60,
min_octets_per_codec_frame=120,
max_octets_per_codec_frame=120,
supported_max_codec_frames_per_sdu=1,
),
),
@@ -159,7 +158,7 @@ async def main() -> None:
+ struct.pack(
'<HHHHHHI',
18, # Header length.
24000 // 100, # Sampling Rate(/100Hz).
48000 // 100, # Sampling Rate(/100Hz).
0, # Bitrate(unused).
1, # Channels.
10000 // 10, # Frame duration(/10us).
@@ -179,11 +178,7 @@ async def main() -> None:
device.once('cis_establishment', on_cis)
await device.start_extended_advertising(
advertising_properties=(
HCI_LE_Set_Extended_Advertising_Parameters_Command.AdvertisingProperties.CONNECTABLE_ADVERTISING
),
own_address_type=OwnAddressType.RANDOM,
advertising_set = await device.create_advertising_set(
advertising_data=advertising_data,
)

View File

@@ -0,0 +1,191 @@
# Copyright 2021-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.
# -----------------------------------------------------------------------------
# Imports
# -----------------------------------------------------------------------------
import asyncio
import logging
import sys
import os
import secrets
import websockets
import json
from bumble.core import AdvertisingData
from bumble.device import Device, AdvertisingParameters, AdvertisingEventProperties
from bumble.hci import (
CodecID,
CodingFormat,
OwnAddressType,
)
from bumble.profiles.bap import (
CodecSpecificCapabilities,
ContextType,
AudioLocation,
SupportedSamplingFrequency,
SupportedFrameDuration,
PacRecord,
PublishedAudioCapabilitiesService,
AudioStreamControlService,
)
from bumble.profiles.cap import CommonAudioServiceService
from bumble.profiles.csip import CoordinatedSetIdentificationService, SirkType
from bumble.profiles.vcp import VolumeControlService
from bumble.transport import open_transport_or_link
from typing import Optional
def dumps_volume_state(volume_setting: int, muted: int, change_counter: int) -> str:
return json.dumps(
{
'volume_setting': volume_setting,
'muted': muted,
'change_counter': change_counter,
}
)
# -----------------------------------------------------------------------------
async def main() -> None:
if len(sys.argv) < 3:
print('Usage: run_vcp_renderer.py <config-file>' '<transport-spec-for-device>')
return
print('<<< connecting to HCI...')
async with await open_transport_or_link(sys.argv[2]) as hci_transport:
print('<<< connected')
device = Device.from_config_file_with_hci(
sys.argv[1], hci_transport.source, hci_transport.sink
)
await device.power_on()
# Add "placeholder" services to enable Android LEA features.
csis = CoordinatedSetIdentificationService(
set_identity_resolving_key=secrets.token_bytes(16),
set_identity_resolving_key_type=SirkType.PLAINTEXT,
)
device.add_service(CommonAudioServiceService(csis))
device.add_service(
PublishedAudioCapabilitiesService(
supported_source_context=ContextType.PROHIBITED,
available_source_context=ContextType.PROHIBITED,
supported_sink_context=ContextType.MEDIA,
available_sink_context=ContextType.MEDIA,
sink_audio_locations=(
AudioLocation.FRONT_LEFT | AudioLocation.FRONT_RIGHT
),
sink_pac=[
# Codec Capability Setting 48_4
PacRecord(
coding_format=CodingFormat(CodecID.LC3),
codec_specific_capabilities=CodecSpecificCapabilities(
supported_sampling_frequencies=(
SupportedSamplingFrequency.FREQ_48000
),
supported_frame_durations=(
SupportedFrameDuration.DURATION_10000_US_SUPPORTED
),
supported_audio_channel_counts=[1],
min_octets_per_codec_frame=120,
max_octets_per_codec_frame=120,
supported_max_codec_frames_per_sdu=1,
),
),
],
)
)
device.add_service(AudioStreamControlService(device, sink_ase_id=[1, 2]))
vcs = VolumeControlService()
device.add_service(vcs)
ws: Optional[websockets.WebSocketServerProtocol] = None
def on_volume_state(volume_setting: int, muted: int, change_counter: int):
if ws:
asyncio.create_task(
ws.send(dumps_volume_state(volume_setting, muted, change_counter))
)
vcs.on('volume_state', on_volume_state)
advertising_data = (
bytes(
AdvertisingData(
[
(
AdvertisingData.COMPLETE_LOCAL_NAME,
bytes('Bumble LE Audio', 'utf-8'),
),
(
AdvertisingData.FLAGS,
bytes(
[
AdvertisingData.LE_GENERAL_DISCOVERABLE_MODE_FLAG
| AdvertisingData.BR_EDR_HOST_FLAG
| AdvertisingData.BR_EDR_CONTROLLER_FLAG
]
),
),
(
AdvertisingData.INCOMPLETE_LIST_OF_16_BIT_SERVICE_CLASS_UUIDS,
bytes(PublishedAudioCapabilitiesService.UUID),
),
]
)
)
+ csis.get_advertising_data()
)
await device.create_advertising_set(
advertising_parameters=AdvertisingParameters(
advertising_event_properties=AdvertisingEventProperties(),
own_address_type=OwnAddressType.PUBLIC,
),
advertising_data=advertising_data,
)
async def serve(websocket: websockets.WebSocketServerProtocol, _path):
nonlocal ws
await websocket.send(
dumps_volume_state(vcs.volume_setting, vcs.muted, vcs.change_counter)
)
ws = websocket
async for message in websocket:
volume_state = json.loads(message)
vcs.volume_state_bytes = bytes(
[
volume_state['volume_setting'],
volume_state['muted'],
volume_state['change_counter'],
]
)
await device.notify_subscribers(
vcs.volume_state, vcs.volume_state_bytes
)
ws = None
await websockets.serve(serve, 'localhost', 8989)
await hci_transport.source.terminated
# -----------------------------------------------------------------------------
logging.basicConfig(level=os.environ.get('BUMBLE_LOGLEVEL', 'DEBUG').upper())
asyncio.run(main())

103
examples/vcp_renderer.html Normal file
View File

@@ -0,0 +1,103 @@
<html data-bs-theme="dark">
<head>
<link href="https://cdn.jsdelivr.net/npm/bootstrap@5.3.2/dist/css/bootstrap.min.css" rel="stylesheet"
integrity="sha384-T3c6CoIi6uLrA9TneNEoa7RxnatzjcDSCmG1MXxSR1GAsXEV/Dwwykc2MPK8M2HN" crossorigin="anonymous">
</head>
<body>
<div class="container">
<label for="server-port" class="form-label">Server Port</label>
<div class="input-group mb-3">
<input type="text" class="form-control" aria-label="Port Number" value="8989" id="port">
<button class="btn btn-primary" type="button" onclick="connect()">Connect</button>
</div>
<div class="row">
<div class="col">
<label for="volume_setting" class="form-label">Volume Setting</label>
<input type="range" class="form-range" min="0" max="255" id="volume_setting">
</div>
<div class="col">
<label for="change_counter" class="form-label">Change Counter</label>
<input type="range" class="form-range" min="0" max="255" id="change_counter">
</div>
<div class="col">
<div class="form-check form-switch">
<input class="form-check-input" type="checkbox" role="switch" id="muted">
<label class="form-check-label" for="muted">Muted</label>
</div>
</div>
</div>
<button class="btn btn-primary" type="button" onclick="update_state()">Notify New Volume State</button>
<hr>
<div id="socketStateContainer" class="bg-body-tertiary p-3 rounded-2">
<h3>Log</h3>
<code id="socketState">
</code>
</div>
</div>
<script>
let portInput = document.getElementById("port")
let volumeSetting = document.getElementById("volume_setting")
let muted = document.getElementById("muted")
let changeCounter = document.getElementById("change_counter")
let socket = null
function connect() {
if (socket != null) {
return
}
socket = new WebSocket(`ws://localhost:${portInput.value}`);
socket.onopen = _ => {
socketState.innerText += 'OPEN\n'
}
socket.onclose = _ => {
socketState.innerText += 'CLOSED\n'
socket = null
}
socket.onerror = (error) => {
socketState.innerText += 'ERROR\n'
console.log(`ERROR: ${error}`)
}
socket.onmessage = (event) => {
socketState.innerText += `<- ${event.data}\n`
let volume_state = JSON.parse(event.data)
volumeSetting.value = volume_state.volume_setting
changeCounter.value = volume_state.change_counter
muted.checked = volume_state.muted ? true : false
}
}
function send(message) {
if (socket && socket.readyState == WebSocket.OPEN) {
let jsonMessage = JSON.stringify(message)
socketState.innerText += `-> ${jsonMessage}\n`
socket.send(jsonMessage)
} else {
socketState.innerText += 'NOT CONNECTED\n'
}
}
function update_state() {
send({
volume_setting: parseInt(volumeSetting.value),
change_counter: parseInt(changeCounter.value),
muted: muted.checked ? 1 : 0
})
}
</script>
<script src="https://cdn.jsdelivr.net/npm/bootstrap@5.3.2/dist/js/bootstrap.bundle.min.js"
integrity="sha384-C6RzsynM9kWDrMNeT87bh95OGNyZPhcTNXj1NW7RuBCsyN/o0jlpcV8Qyq46cDfL"
crossorigin="anonymous"></script>
</body>
</html>

View File

@@ -1,7 +1,10 @@
# Next
# 0.2.0
- Code-gen company ID table
- Unstable support for extended advertisements
- CLI tools for downloading Realtek firmware
- PDL-generated types for HCI commands
# 0.1.0
- Initial release
- Initial release

2
rust/Cargo.lock generated
View File

@@ -182,7 +182,7 @@ dependencies = [
[[package]]
name = "bumble"
version = "0.1.0"
version = "0.2.0"
dependencies = [
"anyhow",
"bytes",

View File

@@ -1,7 +1,7 @@
[package]
name = "bumble"
description = "Rust API for the Bumble Bluetooth stack"
version = "0.1.0"
version = "0.2.0"
edition = "2021"
license = "Apache-2.0"
homepage = "https://google.github.io/bumble/index.html"
@@ -10,7 +10,7 @@ documentation = "https://docs.rs/crate/bumble"
authors = ["Marshall Pierce <marshallpierce@google.com>"]
keywords = ["bluetooth", "ble"]
categories = ["api-bindings", "network-programming"]
rust-version = "1.70.0"
rust-version = "1.76.0"
# https://github.com/frewsxcv/cargo-all-features#options
[package.metadata.cargo-all-features]

View File

@@ -37,6 +37,11 @@ PYTHONPATH=..:[virtualenv site-packages] \
cargo run --features bumble-tools --bin bumble -- --help
```
Notable subcommands:
- `firmware realtek download`: download Realtek firmware for various chipsets so that it can be automatically loaded when needed
- `usb probe`: show USB devices, highlighting the ones usable for Bluetooth
# Development
Run the tests:
@@ -63,4 +68,4 @@ To regenerate the assigned number tables based on the Python codebase:
```
PYTHONPATH=.. cargo run --bin gen-assigned-numbers --features dev-tools
```
```

View File

@@ -35,7 +35,7 @@ impl Controller {
/// module specifies the defaults. Must be called from a thread with a Python event loop, which
/// should be true on `tokio::main` and `async_std::main`.
///
/// For more info, see https://awestlake87.github.io/pyo3-asyncio/master/doc/pyo3_asyncio/#event-loop-references-and-contextvars.
/// For more info, see <https://awestlake87.github.io/pyo3-asyncio/master/doc/pyo3_asyncio/#event-loop-references-and-contextvars>.
pub async fn new(
name: &str,
host_source: Option<TransportSource>,

View File

@@ -149,7 +149,7 @@ impl ToPyObject for Address {
/// An error meaning that the u64 value did not represent a valid BT address.
#[derive(Debug)]
pub struct InvalidAddress(u64);
pub struct InvalidAddress(#[allow(unused)] u64);
impl TryInto<packets::Address> for Address {
type Error = ConversionError<InvalidAddress>;

View File

@@ -71,7 +71,7 @@ impl LeConnectionOrientedChannel {
/// Must be called from a thread with a Python event loop, which should be true on
/// `tokio::main` and `async_std::main`.
///
/// For more info, see https://awestlake87.github.io/pyo3-asyncio/master/doc/pyo3_asyncio/#event-loop-references-and-contextvars.
/// For more info, see <https://awestlake87.github.io/pyo3-asyncio/master/doc/pyo3_asyncio/#event-loop-references-and-contextvars>.
pub async fn disconnect(&mut self) -> PyResult<()> {
Python::with_gil(|py| {
self.0

View File

@@ -52,7 +52,7 @@ install_requires =
pyserial-asyncio >= 0.5; platform_system!='Emscripten'
pyserial >= 3.5; platform_system!='Emscripten'
pyusb >= 1.2; platform_system!='Emscripten'
websockets >= 8.1; platform_system!='Emscripten'
websockets >= 12.0; platform_system!='Emscripten'
[options.entry_points]
console_scripts =
@@ -98,8 +98,8 @@ development =
types-invoke >= 1.7.3
types-protobuf >= 4.21.0
avatar =
pandora-avatar == 0.0.5
rootcanal == 1.6.0 ; python_version>='3.10'
pandora-avatar == 0.0.8
rootcanal == 1.9.0 ; python_version>='3.10'
documentation =
mkdocs >= 1.4.0
mkdocs-material >= 8.5.6

View File

@@ -28,7 +28,7 @@ from bumble.core import (
BT_PERIPHERAL_ROLE,
ConnectionParameters,
)
from bumble.device import Connection, Device
from bumble.device import AdvertisingParameters, Connection, Device
from bumble.host import AclPacketQueue, Host
from bumble.hci import (
HCI_ACCEPT_CONNECTION_REQUEST_COMMAND,
@@ -50,7 +50,8 @@ from bumble.gatt import (
GATT_APPEARANCE_CHARACTERISTIC,
)
from .test_utils import TwoDevices
from .test_utils import TwoDevices, async_barrier
# -----------------------------------------------------------------------------
# Logging
@@ -254,12 +255,12 @@ async def test_legacy_advertising():
device = Device(host=mock.AsyncMock(Host))
# Start advertising
advertiser = await device.start_legacy_advertising()
assert device.legacy_advertiser
await device.start_advertising()
assert device.is_advertising
# Stop advertising
await advertiser.stop()
assert not device.legacy_advertiser
await device.stop_advertising()
assert not device.is_advertising
# -----------------------------------------------------------------------------
@@ -273,7 +274,7 @@ async def test_legacy_advertising_connection(own_address_type):
peer_address = Address('F0:F1:F2:F3:F4:F5')
# Start advertising
advertiser = await device.start_legacy_advertising()
await device.start_advertising()
device.on_connection(
0x0001,
BT_LE_TRANSPORT,
@@ -301,7 +302,7 @@ async def test_legacy_advertising_connection(own_address_type):
async def test_legacy_advertising_disconnection(auto_restart):
device = Device(host=mock.AsyncMock(spec=Host))
peer_address = Address('F0:F1:F2:F3:F4:F5')
advertiser = await device.start_legacy_advertising(auto_restart=auto_restart)
await device.start_advertising(auto_restart=auto_restart)
device.on_connection(
0x0001,
BT_LE_TRANSPORT,
@@ -310,20 +311,18 @@ async def test_legacy_advertising_disconnection(auto_restart):
ConnectionParameters(0, 0, 0),
)
device.start_legacy_advertising = mock.AsyncMock()
device.on_advertising_set_termination(
HCI_SUCCESS, device.legacy_advertising_set.advertising_handle, 0x0001, 0
)
device.on_disconnection(0x0001, 0)
await async_barrier()
await async_barrier()
if auto_restart:
device.start_legacy_advertising.assert_called_with(
advertising_type=advertiser.advertising_type,
own_address_type=advertiser.own_address_type,
auto_restart=advertiser.auto_restart,
advertising_data=advertiser.advertising_data,
scan_response_data=advertiser.scan_response_data,
)
assert device.is_advertising
else:
device.start_legacy_advertising.assert_not_called()
assert not device.is_advertising
# -----------------------------------------------------------------------------
@@ -332,12 +331,13 @@ async def test_extended_advertising():
device = Device(host=mock.AsyncMock(Host))
# Start advertising
advertiser = await device.start_extended_advertising()
assert device.extended_advertisers
advertising_set = await device.create_advertising_set()
assert device.extended_advertising_sets
assert advertising_set.enabled
# Stop advertising
await advertiser.stop()
assert not device.extended_advertisers
await advertising_set.stop()
assert not advertising_set.enabled
# -----------------------------------------------------------------------------
@@ -349,8 +349,8 @@ async def test_extended_advertising():
async def test_extended_advertising_connection(own_address_type):
device = Device(host=mock.AsyncMock(spec=Host))
peer_address = Address('F0:F1:F2:F3:F4:F5')
advertiser = await device.start_extended_advertising(
own_address_type=own_address_type
advertising_set = await device.create_advertising_set(
advertising_parameters=AdvertisingParameters(own_address_type=own_address_type)
)
device.on_connection(
0x0001,
@@ -361,8 +361,9 @@ async def test_extended_advertising_connection(own_address_type):
)
device.on_advertising_set_termination(
HCI_SUCCESS,
advertiser.handle,
advertising_set.advertising_handle,
0x0001,
0,
)
if own_address_type == OwnAddressType.PUBLIC:
@@ -375,45 +376,6 @@ async def test_extended_advertising_connection(own_address_type):
await asyncio.sleep(0.0001)
# -----------------------------------------------------------------------------
@pytest.mark.parametrize(
'auto_restart,',
(True, False),
)
@pytest.mark.asyncio
async def test_extended_advertising_disconnection(auto_restart):
device = Device(host=mock.AsyncMock(spec=Host))
peer_address = Address('F0:F1:F2:F3:F4:F5')
advertiser = await device.start_extended_advertising(auto_restart=auto_restart)
device.on_connection(
0x0001,
BT_LE_TRANSPORT,
peer_address,
BT_PERIPHERAL_ROLE,
ConnectionParameters(0, 0, 0),
)
device.on_advertising_set_termination(
HCI_SUCCESS,
advertiser.handle,
0x0001,
)
device.start_extended_advertising = mock.AsyncMock()
device.on_disconnection(0x0001, 0)
if auto_restart:
device.start_extended_advertising.assert_called_with(
advertising_properties=advertiser.advertising_properties,
own_address_type=advertiser.own_address_type,
auto_restart=advertiser.auto_restart,
advertising_data=advertiser.advertising_data,
scan_response_data=advertiser.scan_response_data,
)
else:
device.start_extended_advertising.assert_not_called()
# -----------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_get_remote_le_features():

View File

@@ -50,6 +50,7 @@ from bumble.att import (
ATT_Error_Response,
ATT_Read_By_Group_Type_Request,
)
from .test_utils import async_barrier
# -----------------------------------------------------------------------------
@@ -456,13 +457,6 @@ class LinkedDevices:
self.paired = [None, None, None]
# -----------------------------------------------------------------------------
async def async_barrier():
ready = asyncio.get_running_loop().create_future()
asyncio.get_running_loop().call_soon(ready.set_result, None)
await ready
# -----------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_read_write():

View File

@@ -23,6 +23,8 @@ from bumble.hci import (
HCI_LE_READ_BUFFER_SIZE_COMMAND,
HCI_RESET_COMMAND,
HCI_SUCCESS,
HCI_LE_CONNECTION_COMPLETE_EVENT,
HCI_LE_ENHANCED_CONNECTION_COMPLETE_V2_EVENT,
Address,
CodingFormat,
CodecID,
@@ -274,8 +276,14 @@ def test_HCI_Set_Event_Mask_Command():
# -----------------------------------------------------------------------------
def test_HCI_LE_Set_Event_Mask_Command():
command = HCI_LE_Set_Event_Mask_Command(
le_event_mask=bytes.fromhex('0011223344556677')
le_event_mask=HCI_LE_Set_Event_Mask_Command.mask(
[
HCI_LE_CONNECTION_COMPLETE_EVENT,
HCI_LE_ENHANCED_CONNECTION_COMPLETE_V2_EVENT,
]
)
)
assert command.le_event_mask == bytes.fromhex('0100000000010000')
basic_check(command)

62
tests/host_test.py Normal file
View File

@@ -0,0 +1,62 @@
# Copyright 2021-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.
# -----------------------------------------------------------------------------
# Imports
# -----------------------------------------------------------------------------
import logging
import pytest
from bumble.controller import Controller
from bumble.host import Host
from bumble.transport import AsyncPipeSink
# -----------------------------------------------------------------------------
# Logging
# -----------------------------------------------------------------------------
logger = logging.getLogger(__name__)
# -----------------------------------------------------------------------------
@pytest.mark.asyncio
@pytest.mark.parametrize(
'supported_commands, lmp_features',
[
(
# Default commands
'2000800000c000000000e4000000a822000000000000040000f7ffff7f000000'
'30f0f9ff01008004000000000000000000000000000000000000000000000000',
# Only LE LMP feature
'0000000060000000',
),
(
# All commands
'ffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff'
'ffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff',
# 3 pages of LMP features
'000102030405060708090A0B0C0D0E0F011112131415161718191A1B1C1D1E1F',
),
],
)
async def test_reset(supported_commands: str, lmp_features: str):
controller = Controller('C')
controller.supported_commands = bytes.fromhex(supported_commands)
controller.lmp_features = bytes.fromhex(lmp_features)
host = Host(controller, AsyncPipeSink(controller))
await host.reset()
assert host.local_lmp_features == int.from_bytes(
bytes.fromhex(lmp_features), 'little'
)

View File

@@ -227,12 +227,34 @@ async def test_bidirectional_transfer():
assert server_received_bytes == message_bytes
# -----------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_mtu():
devices = TwoDevices()
await devices.setup_connection()
def on_channel_open(channel):
assert channel.peer_mtu == 456
def on_channel(channel):
channel.on('open', lambda: on_channel_open(channel))
server = devices.devices[1].create_l2cap_server(
spec=ClassicChannelSpec(mtu=345), handler=on_channel
)
client_channel = await devices.connections[0].create_l2cap_channel(
spec=ClassicChannelSpec(server.psm, mtu=456)
)
assert client_channel.peer_mtu == 345
# -----------------------------------------------------------------------------
async def run():
test_helpers()
await test_basic_connection()
await test_transfer()
await test_bidirectional_transfer()
await test_mtu()
# -----------------------------------------------------------------------------

View File

@@ -17,9 +17,20 @@
# -----------------------------------------------------------------------------
import asyncio
import pytest
from typing import List
from . import test_utils
from bumble.rfcomm import RFCOMM_Frame, Server, Client, DLC
from bumble import core
from bumble.rfcomm import (
RFCOMM_Frame,
Server,
Client,
DLC,
make_service_sdp_records,
find_rfcomm_channels,
find_rfcomm_channel_with_uuid,
RFCOMM_PSM,
)
# -----------------------------------------------------------------------------
@@ -49,17 +60,18 @@ def test_frames():
# -----------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_basic_connection():
async def test_basic_connection() -> None:
devices = test_utils.TwoDevices()
await devices.setup_connection()
accept_future: asyncio.Future[DLC] = asyncio.get_running_loop().create_future()
channel = Server(devices[0]).listen(acceptor=accept_future.set_result)
assert devices.connections[1]
multiplexer = await Client(devices.connections[1]).start()
dlcs = await asyncio.gather(accept_future, multiplexer.open_dlc(channel))
queues = [asyncio.Queue(), asyncio.Queue()]
queues: List[asyncio.Queue] = [asyncio.Queue(), asyncio.Queue()]
for dlc, queue in zip(dlcs, queues):
dlc.sink = queue.put_nowait
@@ -70,6 +82,45 @@ async def test_basic_connection():
assert await queues[0].get() == b'Lorem ipsum dolor sit amet'
# -----------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_service_record():
HANDLE = 2
CHANNEL = 1
SERVICE_UUID = core.UUID('00000000-0000-0000-0000-000000000001')
devices = test_utils.TwoDevices()
await devices.setup_connection()
devices[0].sdp_service_records[HANDLE] = make_service_sdp_records(
HANDLE, CHANNEL, SERVICE_UUID
)
assert SERVICE_UUID in (await find_rfcomm_channels(devices.connections[1]))[CHANNEL]
assert (
await find_rfcomm_channel_with_uuid(devices.connections[1], SERVICE_UUID)
== CHANNEL
)
# -----------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_context():
devices = test_utils.TwoDevices()
await devices.setup_connection()
server = Server(devices[0])
with server:
assert server.l2cap_server is not None
client = Client(devices.connections[1])
async with client:
assert client.l2cap_channel is not None
assert client.l2cap_channel is None
assert RFCOMM_PSM not in devices[0].l2cap_channel_manager.servers
# -----------------------------------------------------------------------------
if __name__ == '__main__':
test_frames()

View File

@@ -38,6 +38,7 @@ from .test_utils import TwoDevices
# pylint: disable=invalid-name
# -----------------------------------------------------------------------------
# -----------------------------------------------------------------------------
def basic_check(x: DataElement) -> None:
serialized = bytes(x)
@@ -269,6 +270,20 @@ async def test_service_search_attribute():
assert expect.value == actual.value
# -----------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_client_async_context():
devices = TwoDevices()
await devices.setup_connection()
client = Client(devices.connections[1])
async with client:
assert client.channel is not None
assert client.channel is None
# -----------------------------------------------------------------------------
async def run():
test_data_elements()

View File

@@ -12,6 +12,10 @@
# See the License for the specific language governing permissions and
# limitations under the License.
# -----------------------------------------------------------------------------
# Imports
# -----------------------------------------------------------------------------
import asyncio
from typing import List, Optional
from bumble.controller import Controller
@@ -22,6 +26,7 @@ from bumble.transport import AsyncPipeSink
from bumble.hci import Address
# -----------------------------------------------------------------------------
class TwoDevices:
connections: List[Optional[Connection]]
@@ -75,3 +80,10 @@ class TwoDevices:
def __getitem__(self, index: int) -> Device:
return self.devices[index]
# -----------------------------------------------------------------------------
async def async_barrier():
ready = asyncio.get_running_loop().create_future()
asyncio.get_running_loop().call_soon(ready.set_result, None)
await ready

120
tests/vcp_test.py Normal file
View File

@@ -0,0 +1,120 @@
# Copyright 2021-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.
# -----------------------------------------------------------------------------
# Imports
# -----------------------------------------------------------------------------
import pytest
import pytest_asyncio
import logging
from bumble import device
from bumble.profiles import vcp
from .test_utils import TwoDevices
# -----------------------------------------------------------------------------
# Logging
# -----------------------------------------------------------------------------
logger = logging.getLogger(__name__)
# -----------------------------------------------------------------------------
@pytest_asyncio.fixture
async def vcp_client():
devices = TwoDevices()
devices[0].add_service(
vcp.VolumeControlService(volume_setting=32, muted=1, volume_flags=1)
)
await devices.setup_connection()
# Mock encryption.
devices.connections[0].encryption = 1
devices.connections[1].encryption = 1
peer = device.Peer(devices.connections[1])
vcp_client = await peer.discover_service_and_create_proxy(
vcp.VolumeControlServiceProxy
)
yield vcp_client
# -----------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_init_service(vcp_client: vcp.VolumeControlServiceProxy):
assert (await vcp_client.volume_flags.read_value()) == 1
assert (await vcp_client.volume_state.read_value()) == (32, 1, 0)
# -----------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_relative_volume_down(vcp_client: vcp.VolumeControlServiceProxy):
await vcp_client.volume_control_point.write_value(
bytes([vcp.VolumeControlPointOpcode.RELATIVE_VOLUME_DOWN, 0])
)
assert (await vcp_client.volume_state.read_value()) == (16, 1, 1)
# -----------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_relative_volume_up(vcp_client: vcp.VolumeControlServiceProxy):
await vcp_client.volume_control_point.write_value(
bytes([vcp.VolumeControlPointOpcode.RELATIVE_VOLUME_UP, 0])
)
assert (await vcp_client.volume_state.read_value()) == (48, 1, 1)
# -----------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_unmute_relative_volume_down(vcp_client: vcp.VolumeControlServiceProxy):
await vcp_client.volume_control_point.write_value(
bytes([vcp.VolumeControlPointOpcode.UNMUTE_RELATIVE_VOLUME_DOWN, 0])
)
assert (await vcp_client.volume_state.read_value()) == (16, 0, 1)
# -----------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_unmute_relative_volume_up(vcp_client: vcp.VolumeControlServiceProxy):
await vcp_client.volume_control_point.write_value(
bytes([vcp.VolumeControlPointOpcode.UNMUTE_RELATIVE_VOLUME_UP, 0])
)
assert (await vcp_client.volume_state.read_value()) == (48, 0, 1)
# -----------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_set_absolute_volume(vcp_client: vcp.VolumeControlServiceProxy):
await vcp_client.volume_control_point.write_value(
bytes([vcp.VolumeControlPointOpcode.SET_ABSOLUTE_VOLUME, 0, 255])
)
assert (await vcp_client.volume_state.read_value()) == (255, 1, 1)
# -----------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_mute(vcp_client: vcp.VolumeControlServiceProxy):
await vcp_client.volume_control_point.write_value(
bytes([vcp.VolumeControlPointOpcode.MUTE, 0])
)
assert (await vcp_client.volume_state.read_value()) == (32, 1, 0)
# -----------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_unmute(vcp_client: vcp.VolumeControlServiceProxy):
await vcp_client.volume_control_point.write_value(
bytes([vcp.VolumeControlPointOpcode.UNMUTE, 0])
)
assert (await vcp_client.volume_state.read_value()) == (32, 0, 1)