auracaster/bumble_mirror
3
0
Fork 1
mirror of https://github.com/google/bumble.git synced 2026-04-17 00:35:31 +00:00
Code Issues Packages Projects Releases Wiki Activity

format with Black

Browse Source
This commit is contained in:
Gilles Boccon-Gibod
2022-12-10 08:53:51 -08:00
parent 297246fa4c
commit 135df0dcc0
104 changed files with 8646 additions and 5766 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
examples/async_runner.py
View File

@@ -80,6 +80,7 @@ async def main():
await my_work_queue2.run()
print("MAIN: end (should never get here)")
# -----------------------------------------------------------------------------
logging.basicConfig(level = os.environ.get('BUMBLE_LOGLEVEL', 'DEBUG').upper())
logging.basicConfig(level=os.environ.get('BUMBLE_LOGLEVEL', 'DEBUG').upper())
asyncio.run(main())

6
examples/battery_client.py
View File

@@ -55,7 +55,9 @@ async def main():
# Subscribe to and read the battery level
if battery_service.battery_level:
await battery_service.battery_level.subscribe(
lambda value: print(f'{color("Battery Level Update:", "green")} {value}')
lambda value: print(
f'{color("Battery Level Update:", "green")} {value}'
)
)
value = await battery_service.battery_level.read_value()
print(f'{color("Initial Battery Level:", "green")} {value}')
@@ -64,5 +66,5 @@ async def main():
# -----------------------------------------------------------------------------
logging.basicConfig(level = os.environ.get('BUMBLE_LOGLEVEL', 'DEBUG').upper())
logging.basicConfig(level=os.environ.get('BUMBLE_LOGLEVEL', 'DEBUG').upper())
asyncio.run(main())

24
examples/battery_server.py
View File

@@ -44,11 +44,19 @@ async def main():
# Set the advertising data
device.advertising_data = bytes(
AdvertisingData([
(AdvertisingData.COMPLETE_LOCAL_NAME, bytes('Bumble Battery', 'utf-8')),
(AdvertisingData.INCOMPLETE_LIST_OF_16_BIT_SERVICE_CLASS_UUIDS, bytes(battery_service.uuid)),
(AdvertisingData.APPEARANCE, struct.pack('<H', 0x0340))
])
AdvertisingData(
[
(
AdvertisingData.COMPLETE_LOCAL_NAME,
bytes('Bumble Battery', 'utf-8'),
),
(
AdvertisingData.INCOMPLETE_LIST_OF_16_BIT_SERVICE_CLASS_UUIDS,
bytes(battery_service.uuid),
),
(AdvertisingData.APPEARANCE, struct.pack('<H', 0x0340)),
]
)
)
# Go!
@@ -58,9 +66,11 @@ async def main():
# Notify every 3 seconds
while True:
await asyncio.sleep(3.0)
await device.notify_subscribers(battery_service.battery_level_characteristic)
await device.notify_subscribers(
battery_service.battery_level_characteristic
)
# -----------------------------------------------------------------------------
logging.basicConfig(level = os.environ.get('BUMBLE_LOGLEVEL', 'DEBUG').upper())
logging.basicConfig(level=os.environ.get('BUMBLE_LOGLEVEL', 'DEBUG').upper())
asyncio.run(main())

57
examples/device_information_client.py
View File

@@ -28,7 +28,9 @@ from bumble.transport import open_transport
# -----------------------------------------------------------------------------
async def main():
if len(sys.argv) != 3:
print('Usage: device_information_client.py <transport-spec> <bluetooth-address>')
print(
'Usage: device_information_client.py <transport-spec> <bluetooth-address>'
)
print('example: device_information_client.py usb:0 E1:CA:72:48:C4:E8')
return
@@ -49,7 +51,9 @@ async def main():
# Discover the Device Information service
peer = Peer(connection)
print('=== Discovering Device Information Service')
device_information_service = await peer.discover_service_and_create_proxy(DeviceInformationServiceProxy)
device_information_service = await peer.discover_service_and_create_proxy(
DeviceInformationServiceProxy
)
# Check that the service was found
if device_information_service is None:
@@ -58,23 +62,52 @@ async def main():
# Read and print the fields
if device_information_service.manufacturer_name is not None:
print(color('Manufacturer Name: ', 'green'), await device_information_service.manufacturer_name.read_value())
print(
color('Manufacturer Name: ', 'green'),
await device_information_service.manufacturer_name.read_value(),
)
if device_information_service.model_number is not None:
print(color('Model Number: ', 'green'), await device_information_service.model_number.read_value())
print(
color('Model Number: ', 'green'),
await device_information_service.model_number.read_value(),
)
if device_information_service.serial_number is not None:
print(color('Serial Number: ', 'green'), await device_information_service.serial_number.read_value())
print(
color('Serial Number: ', 'green'),
await device_information_service.serial_number.read_value(),
)
if device_information_service.hardware_revision is not None:
print(color('Hardware Revision: ', 'green'), await device_information_service.hardware_revision.read_value())
print(
color('Hardware Revision: ', 'green'),
await device_information_service.hardware_revision.read_value(),
)
if device_information_service.firmware_revision is not None:
print(color('Firmware Revision: ', 'green'), await device_information_service.firmware_revision.read_value())
print(
color('Firmware Revision: ', 'green'),
await device_information_service.firmware_revision.read_value(),
)
if device_information_service.software_revision is not None:
print(color('Software Revision: ', 'green'), await device_information_service.software_revision.read_value())
print(
color('Software Revision: ', 'green'),
await device_information_service.software_revision.read_value(),
)
if device_information_service.system_id is not None:
print(color('System ID: ', 'green'), await device_information_service.system_id.read_value())
if device_information_service.ieee_regulatory_certification_data_list is not None:
print(color('Regulatory Certification:', 'green'), (await device_information_service.ieee_regulatory_certification_data_list.read_value()).hex())
print(
color('System ID: ', 'green'),
await device_information_service.system_id.read_value(),
)
if (
device_information_service.ieee_regulatory_certification_data_list
is not None
):
print(
color('Regulatory Certification:', 'green'),
(
await device_information_service.ieee_regulatory_certification_data_list.read_value()
).hex(),
)
# -----------------------------------------------------------------------------
logging.basicConfig(level = os.environ.get('BUMBLE_LOGLEVEL', 'DEBUG').upper())
logging.basicConfig(level=os.environ.get('BUMBLE_LOGLEVEL', 'DEBUG').upper())
asyncio.run(main())

28
examples/device_information_server.py
View File

@@ -39,21 +39,26 @@ async def main():
# Add a Device Information Service to the GATT sever
device_information_service = DeviceInformationService(
manufacturer_name = 'ACME',
model_number = 'AB-102',
serial_number = '7654321',
hardware_revision = '1.1.3',
software_revision = '2.5.6',
system_id = (0x123456, 0x8877665544)
manufacturer_name='ACME',
model_number='AB-102',
serial_number='7654321',
hardware_revision='1.1.3',
software_revision='2.5.6',
system_id=(0x123456, 0x8877665544),
)
device.add_service(device_information_service)
# Set the advertising data
device.advertising_data = bytes(
AdvertisingData([
(AdvertisingData.COMPLETE_LOCAL_NAME, bytes('Bumble Device', 'utf-8')),
(AdvertisingData.APPEARANCE, struct.pack('<H', 0x0340))
])
AdvertisingData(
[
(
AdvertisingData.COMPLETE_LOCAL_NAME,
bytes('Bumble Device', 'utf-8'),
),
(AdvertisingData.APPEARANCE, struct.pack('<H', 0x0340)),
]
)
)
# Go!
@@ -61,6 +66,7 @@ async def main():
await device.start_advertising(auto_restart=True)
await hci_source.wait_for_termination()
# -----------------------------------------------------------------------------
logging.basicConfig(level = os.environ.get('BUMBLE_LOGLEVEL', 'DEBUG').upper())
logging.basicConfig(level=os.environ.get('BUMBLE_LOGLEVEL', 'DEBUG').upper())
asyncio.run(main())

6
examples/heart_rate_client.py
View File

@@ -61,12 +61,14 @@ async def main():
# Subscribe to the heart rate measurement
if heart_rate_service.heart_rate_measurement:
await heart_rate_service.heart_rate_measurement.subscribe(
lambda value: print(f'{color("Heart Rate Measurement:", "green")} {value}')
lambda value: print(
f'{color("Heart Rate Measurement:", "green")} {value}'
)
)
await peer.sustain()
# -----------------------------------------------------------------------------
logging.basicConfig(level = os.environ.get('BUMBLE_LOGLEVEL', 'DEBUG').upper())
logging.basicConfig(level=os.environ.get('BUMBLE_LOGLEVEL', 'DEBUG').upper())
asyncio.run(main())

58
examples/heart_rate_server.py
View File

@@ -50,34 +50,52 @@ async def main():
# Add a Device Information Service and Heart Rate Service to the GATT sever
device_information_service = DeviceInformationService(
manufacturer_name = 'ACME',
model_number = 'HR-102',
serial_number = '7654321',
hardware_revision = '1.1.3',
software_revision = '2.5.6',
system_id = (0x123456, 0x8877665544)
manufacturer_name='ACME',
model_number='HR-102',
serial_number='7654321',
hardware_revision='1.1.3',
software_revision='2.5.6',
system_id=(0x123456, 0x8877665544),
)
heart_rate_service = HeartRateService(
read_heart_rate_measurement = lambda _: HeartRateService.HeartRateMeasurement(
heart_rate = 100 + int(50 * math.sin(time.time() * math.pi / 60)),
sensor_contact_detected = random.choice((True, False, None)),
energy_expended = random.choice((int((time.time() - energy_start_time) * 100), None)),
rr_intervals = random.choice(((random.randint(900, 1100) / 1000, random.randint(900, 1100) / 1000), None))
read_heart_rate_measurement=lambda _: HeartRateService.HeartRateMeasurement(
heart_rate=100 + int(50 * math.sin(time.time() * math.pi / 60)),
sensor_contact_detected=random.choice((True, False, None)),
energy_expended=random.choice(
(int((time.time() - energy_start_time) * 100), None)
),
rr_intervals=random.choice(
(
(
random.randint(900, 1100) / 1000,
random.randint(900, 1100) / 1000,
),
None,
)
),
),
body_sensor_location=HeartRateService.BodySensorLocation.WRIST,
reset_energy_expended=lambda _: reset_energy_expended()
reset_energy_expended=lambda _: reset_energy_expended(),
)
device.add_services([device_information_service, heart_rate_service])
# Set the advertising data
device.advertising_data = bytes(
AdvertisingData([
(AdvertisingData.COMPLETE_LOCAL_NAME, bytes('Bumble Heart', 'utf-8')),
(AdvertisingData.INCOMPLETE_LIST_OF_16_BIT_SERVICE_CLASS_UUIDS, bytes(heart_rate_service.uuid)),
(AdvertisingData.APPEARANCE, struct.pack('<H', 0x0340))
])
AdvertisingData(
[
(
AdvertisingData.COMPLETE_LOCAL_NAME,
bytes('Bumble Heart', 'utf-8'),
),
(
AdvertisingData.INCOMPLETE_LIST_OF_16_BIT_SERVICE_CLASS_UUIDS,
bytes(heart_rate_service.uuid),
),
(AdvertisingData.APPEARANCE, struct.pack('<H', 0x0340)),
]
)
)
# Go!
@@ -87,9 +105,11 @@ async def main():
# Notify every 3 seconds
while True:
await asyncio.sleep(3.0)
await device.notify_subscribers(heart_rate_service.heart_rate_measurement_characteristic)
await device.notify_subscribers(
heart_rate_service.heart_rate_measurement_characteristic
)
# -----------------------------------------------------------------------------
logging.basicConfig(level = os.environ.get('BUMBLE_LOGLEVEL', 'DEBUG').upper())
logging.basicConfig(level=os.environ.get('BUMBLE_LOGLEVEL', 'DEBUG').upper())
asyncio.run(main())

329
examples/keyboard.py
View File

@@ -43,56 +43,90 @@ from bumble.gatt import (
GATT_PROTOCOL_MODE_CHARACTERISTIC,
GATT_HID_INFORMATION_CHARACTERISTIC,
GATT_HID_CONTROL_POINT_CHARACTERISTIC,
GATT_REPORT_REFERENCE_DESCRIPTOR
GATT_REPORT_REFERENCE_DESCRIPTOR,
)
# -----------------------------------------------------------------------------
# Protocol Modes
HID_BOOT_PROTOCOL = 0x00
HID_BOOT_PROTOCOL = 0x00
HID_REPORT_PROTOCOL = 0x01
# Report Types
HID_INPUT_REPORT = 0x01
HID_OUTPUT_REPORT = 0x02
HID_INPUT_REPORT = 0x01
HID_OUTPUT_REPORT = 0x02
HID_FEATURE_REPORT = 0x03
# Report Map
HID_KEYBOARD_REPORT_MAP = bytes([
0x05, 0x01, # Usage Page (Generic Desktop Ctrls)
0x09, 0x06, # Usage (Keyboard)
0xA1, 0x01, # Collection (Application)
0x85, 0x01, # . Report ID (1)
0x05, 0x07, # . Usage Page (Kbrd/Keypad)
0x19, 0xE0, # . Usage Minimum (0xE0)
0x29, 0xE7, # . Usage Maximum (0xE7)
0x15, 0x00, # . Logical Minimum (0)
0x25, 0x01, # . Logical Maximum (1)
0x75, 0x01, # . Report Size (1)
0x95, 0x08, # . Report Count (8)
0x81, 0x02, # . Input (Data,Var,Abs,No Wrap,Linear,Preferred State,No Null Position)
0x95, 0x01, # . Report Count (1)
0x75, 0x08, # . Report Size (8)
0x81, 0x01, # . Input (Const,Array,Abs,No Wrap,Linear,Preferred State,No Null Position)
0x95, 0x06, # . Report Count (6)
0x75, 0x08, # . Report Size (8)
0x15, 0x00, # . Logical Minimum (0x00)
0x25, 0x94, # . Logical Maximum (0x94)
0x05, 0x07, # . Usage Page (Kbrd/Keypad)
0x19, 0x00, # . Usage Minimum (0x00)
0x29, 0x94, # . Usage Maximum (0x94)
0x81, 0x00, # . Input (Data,Array,Abs,No Wrap,Linear,Preferred State,No Null Position)
0x95, 0x05, # . Report Count (5)
0x75, 0x01, # . Report Size (1)
0x05, 0x08, # . Usage Page (LEDs)
0x19, 0x01, # . Usage Minimum (Num Lock)
0x29, 0x05, # . Usage Maximum (Kana)
0x91, 0x02, # . Output (Data,Var,Abs,No Wrap,Linear,Preferred State,No Null Position,Non-volatile)
0x95, 0x01, # . Report Count (1)
0x75, 0x03, # . Report Size (3)
0x91, 0x01, # . Output (Const,Array,Abs,No Wrap,Linear,Preferred State,No Null Position,Non-volatile)
0xC0 # End Collection
])
HID_KEYBOARD_REPORT_MAP = bytes(
[
0x05,
0x01, # Usage Page (Generic Desktop Ctrls)
0x09,
0x06, # Usage (Keyboard)
0xA1,
0x01, # Collection (Application)
0x85,
0x01, # . Report ID (1)
0x05,
0x07, # . Usage Page (Kbrd/Keypad)
0x19,
0xE0, # . Usage Minimum (0xE0)
0x29,
0xE7, # . Usage Maximum (0xE7)
0x15,
0x00, # . Logical Minimum (0)
0x25,
0x01, # . Logical Maximum (1)
0x75,
0x01, # . Report Size (1)
0x95,
0x08, # . Report Count (8)
0x81,
0x02, # . Input (Data,Var,Abs,No Wrap,Linear,Preferred State,No Null Position)
0x95,
0x01, # . Report Count (1)
0x75,
0x08, # . Report Size (8)
0x81,
0x01, # . Input (Const,Array,Abs,No Wrap,Linear,Preferred State,No Null Position)
0x95,
0x06, # . Report Count (6)
0x75,
0x08, # . Report Size (8)
0x15,
0x00, # . Logical Minimum (0x00)
0x25,
0x94, # . Logical Maximum (0x94)
0x05,
0x07, # . Usage Page (Kbrd/Keypad)
0x19,
0x00, # . Usage Minimum (0x00)
0x29,
0x94, # . Usage Maximum (0x94)
0x81,
0x00, # . Input (Data,Array,Abs,No Wrap,Linear,Preferred State,No Null Position)
0x95,
0x05, # . Report Count (5)
0x75,
0x01, # . Report Size (1)
0x05,
0x08, # . Usage Page (LEDs)
0x19,
0x01, # . Usage Minimum (Num Lock)
0x29,
0x05, # . Usage Maximum (Kana)
0x91,
0x02, # . Output (Data,Var,Abs,No Wrap,Linear,Preferred State,No Null Position,Non-volatile)
0x95,
0x01, # . Report Count (1)
0x75,
0x03, # . Report Size (3)
0x91,
0x01, # . Output (Const,Array,Abs,No Wrap,Linear,Preferred State,No Null Position,Non-volatile)
0xC0, # End Collection
]
)
# -----------------------------------------------------------------------------
@@ -133,31 +167,41 @@ async def keyboard_host(device, peer_address):
return
await peer.discover_characteristics()
protocol_mode_characteristics = peer.get_characteristics_by_uuid(GATT_PROTOCOL_MODE_CHARACTERISTIC)
protocol_mode_characteristics = peer.get_characteristics_by_uuid(
GATT_PROTOCOL_MODE_CHARACTERISTIC
)
if not protocol_mode_characteristics:
print(color('!!! No Protocol Mode characteristic', 'red'))
return
protocol_mode_characteristic = protocol_mode_characteristics[0]
hid_information_characteristics = peer.get_characteristics_by_uuid(GATT_HID_INFORMATION_CHARACTERISTIC)
hid_information_characteristics = peer.get_characteristics_by_uuid(
GATT_HID_INFORMATION_CHARACTERISTIC
)
if not hid_information_characteristics:
print(color('!!! No HID Information characteristic', 'red'))
return
hid_information_characteristic = hid_information_characteristics[0]
report_map_characteristics = peer.get_characteristics_by_uuid(GATT_REPORT_MAP_CHARACTERISTIC)
report_map_characteristics = peer.get_characteristics_by_uuid(
GATT_REPORT_MAP_CHARACTERISTIC
)
if not report_map_characteristics:
print(color('!!! No Report Map characteristic', 'red'))
return
report_map_characteristic = report_map_characteristics[0]
control_point_characteristics = peer.get_characteristics_by_uuid(GATT_HID_CONTROL_POINT_CHARACTERISTIC)
control_point_characteristics = peer.get_characteristics_by_uuid(
GATT_HID_CONTROL_POINT_CHARACTERISTIC
)
if not control_point_characteristics:
print(color('!!! No Control Point characteristic', 'red'))
return
# control_point_characteristic = control_point_characteristics[0]
report_characteristics = peer.get_characteristics_by_uuid(GATT_REPORT_CHARACTERISTIC)
report_characteristics = peer.get_characteristics_by_uuid(
GATT_REPORT_CHARACTERISTIC
)
if not report_characteristics:
print(color('!!! No Report characteristic', 'red'))
return
@@ -165,13 +209,20 @@ async def keyboard_host(device, peer_address):
print(color('REPORT:', 'yellow'), characteristic)
if characteristic.properties & Characteristic.NOTIFY:
await peer.discover_descriptors(characteristic)
report_reference_descriptor = characteristic.get_descriptor(GATT_REPORT_REFERENCE_DESCRIPTOR)
report_reference_descriptor = characteristic.get_descriptor(
GATT_REPORT_REFERENCE_DESCRIPTOR
)
if report_reference_descriptor:
report_reference = await peer.read_value(report_reference_descriptor)
print(color(' Report Reference:', 'blue'), report_reference.hex())
else:
report_reference = bytes([0, 0])
await peer.subscribe(characteristic, lambda value, param=f'[{i}] {report_reference.hex()}': on_report(param, value))
await peer.subscribe(
characteristic,
lambda value, param=f'[{i}] {report_reference.hex()}': on_report(
param, value
),
)
protocol_mode = await peer.read_value(protocol_mode_characteristic)
print(f'Protocol Mode: {protocol_mode.hex()}')
@@ -192,77 +243,91 @@ async def keyboard_device(device, command):
Characteristic.READABLE | Characteristic.WRITEABLE,
bytes([0, 0, 0, 0, 0, 0, 0, 0]),
[
Descriptor(GATT_REPORT_REFERENCE_DESCRIPTOR, Descriptor.READABLE, bytes([0x01, HID_INPUT_REPORT]))
]
Descriptor(
GATT_REPORT_REFERENCE_DESCRIPTOR,
Descriptor.READABLE,
bytes([0x01, HID_INPUT_REPORT]),
)
],
)
# Create an 'output report' characteristic to receive keyboard reports from the host
output_report_characteristic = Characteristic(
GATT_REPORT_CHARACTERISTIC,
Characteristic.READ | Characteristic.WRITE | Characteristic.WRITE_WITHOUT_RESPONSE,
Characteristic.READ
| Characteristic.WRITE
| Characteristic.WRITE_WITHOUT_RESPONSE,
Characteristic.READABLE | Characteristic.WRITEABLE,
bytes([0]),
[
Descriptor(GATT_REPORT_REFERENCE_DESCRIPTOR, Descriptor.READABLE, bytes([0x01, HID_OUTPUT_REPORT]))
]
Descriptor(
GATT_REPORT_REFERENCE_DESCRIPTOR,
Descriptor.READABLE,
bytes([0x01, HID_OUTPUT_REPORT]),
)
],
)
# Add the services to the GATT sever
device.add_services([
Service(
GATT_DEVICE_INFORMATION_SERVICE,
[
Characteristic(
GATT_MANUFACTURER_NAME_STRING_CHARACTERISTIC,
Characteristic.READ,
Characteristic.READABLE,
'Bumble'
)
]
),
Service(
GATT_HUMAN_INTERFACE_DEVICE_SERVICE,
[
Characteristic(
GATT_PROTOCOL_MODE_CHARACTERISTIC,
Characteristic.READ,
Characteristic.READABLE,
bytes([HID_REPORT_PROTOCOL])
),
Characteristic(
GATT_HID_INFORMATION_CHARACTERISTIC,
Characteristic.READ,
Characteristic.READABLE,
bytes([0x11, 0x01, 0x00, 0x03]) # bcdHID=1.1, bCountryCode=0x00, Flags=RemoteWake|NormallyConnectable
),
Characteristic(
GATT_HID_CONTROL_POINT_CHARACTERISTIC,
Characteristic.WRITE_WITHOUT_RESPONSE,
Characteristic.WRITEABLE,
CharacteristicValue(write=on_hid_control_point_write)
),
Characteristic(
GATT_REPORT_MAP_CHARACTERISTIC,
Characteristic.READ,
Characteristic.READABLE,
HID_KEYBOARD_REPORT_MAP
),
input_report_characteristic,
output_report_characteristic
]
),
Service(
GATT_BATTERY_SERVICE,
[
Characteristic(
GATT_BATTERY_LEVEL_CHARACTERISTIC,
Characteristic.READ,
Characteristic.READABLE,
bytes([100])
)
]
)
])
device.add_services(
[
Service(
GATT_DEVICE_INFORMATION_SERVICE,
[
Characteristic(
GATT_MANUFACTURER_NAME_STRING_CHARACTERISTIC,
Characteristic.READ,
Characteristic.READABLE,
'Bumble',
)
],
),
Service(
GATT_HUMAN_INTERFACE_DEVICE_SERVICE,
[
Characteristic(
GATT_PROTOCOL_MODE_CHARACTERISTIC,
Characteristic.READ,
Characteristic.READABLE,
bytes([HID_REPORT_PROTOCOL]),
),
Characteristic(
GATT_HID_INFORMATION_CHARACTERISTIC,
Characteristic.READ,
Characteristic.READABLE,
bytes(
[0x11, 0x01, 0x00, 0x03]
), # bcdHID=1.1, bCountryCode=0x00, Flags=RemoteWake|NormallyConnectable
),
Characteristic(
GATT_HID_CONTROL_POINT_CHARACTERISTIC,
Characteristic.WRITE_WITHOUT_RESPONSE,
Characteristic.WRITEABLE,
CharacteristicValue(write=on_hid_control_point_write),
),
Characteristic(
GATT_REPORT_MAP_CHARACTERISTIC,
Characteristic.READ,
Characteristic.READABLE,
HID_KEYBOARD_REPORT_MAP,
),
input_report_characteristic,
output_report_characteristic,
],
),
Service(
GATT_BATTERY_SERVICE,
[
Characteristic(
GATT_BATTERY_LEVEL_CHARACTERISTIC,
Characteristic.READ,
Characteristic.READABLE,
bytes([100]),
)
],
),
]
)
# Debug print
for attribute in device.gatt_server.attributes:
@@ -270,13 +335,20 @@ async def keyboard_device(device, command):
# Set the advertising data
device.advertising_data = bytes(
AdvertisingData([
(AdvertisingData.COMPLETE_LOCAL_NAME, bytes('Bumble Keyboard', 'utf-8')),
(AdvertisingData.INCOMPLETE_LIST_OF_16_BIT_SERVICE_CLASS_UUIDS,
bytes(GATT_HUMAN_INTERFACE_DEVICE_SERVICE)),
(AdvertisingData.APPEARANCE, struct.pack('<H', 0x03C1)),
(AdvertisingData.FLAGS, bytes([0x05]))
])
AdvertisingData(
[
(
AdvertisingData.COMPLETE_LOCAL_NAME,
bytes('Bumble Keyboard', 'utf-8'),
),
(
AdvertisingData.INCOMPLETE_LIST_OF_16_BIT_SERVICE_CLASS_UUIDS,
bytes(GATT_HUMAN_INTERFACE_DEVICE_SERVICE),
),
(AdvertisingData.APPEARANCE, struct.pack('<H', 0x03C1)),
(AdvertisingData.FLAGS, bytes([0x05])),
]
)
)
# Attach a listener
@@ -303,14 +375,21 @@ async def keyboard_device(device, command):
code = ord(key)
if code >= ord('a') and code <= ord('z'):
hid_code = 0x04 + code - ord('a')
input_report_characteristic.value = bytes([0, 0, hid_code, 0, 0, 0, 0, 0])
await device.notify_subscribers(input_report_characteristic)
input_report_characteristic.value = bytes(
[0, 0, hid_code, 0, 0, 0, 0, 0]
)
await device.notify_subscribers(
input_report_characteristic
)
elif message_type == 'keyup':
input_report_characteristic.value = bytes.fromhex('0000000000000000')
input_report_characteristic.value = bytes.fromhex(
'0000000000000000'
)
await device.notify_subscribers(input_report_characteristic)
except websockets.exceptions.ConnectionClosedOK:
pass
await websockets.serve(serve, 'localhost', 8989)
await asyncio.get_event_loop().create_future()
else:
@@ -321,7 +400,9 @@ async def keyboard_device(device, command):
# Keypress for the letter
keycode = 0x04 + letter - 0x61
input_report_characteristic.value = bytes([0, 0, keycode, 0, 0, 0, 0, 0])
input_report_characteristic.value = bytes(
[0, 0, keycode, 0, 0, 0, 0, 0]
)
await device.notify_subscribers(input_report_characteristic)
# Key release
@@ -335,10 +416,16 @@ async def main():
print('Usage: python keyboard.py <device-config> <transport-spec> <command>')
print(' where <command> is one of:')
print(' connect <address> (run a keyboard host, connecting to a keyboard)')
print(' web (run a keyboard with keypress input from a web page, see keyboard.html')
print(' sim (run a keyboard simulation, emitting a canned sequence of keystrokes')
print(
' web (run a keyboard with keypress input from a web page, see keyboard.html'
)
print(
' sim (run a keyboard simulation, emitting a canned sequence of keystrokes'
)
print('example: python keyboard.py keyboard.json usb:0 sim')
print('example: python keyboard.py keyboard.json usb:0 connect A0:A1:A2:A3:A4:A5')
print(
'example: python keyboard.py keyboard.json usb:0 connect A0:A1:A2:A3:A4:A5'
)
return
async with await open_transport_or_link(sys.argv[2]) as (hci_source, hci_sink):
@@ -355,5 +442,5 @@ async def main():
# -----------------------------------------------------------------------------
logging.basicConfig(level = os.environ.get('BUMBLE_LOGLEVEL', 'DEBUG').upper())
logging.basicConfig(level=os.environ.get('BUMBLE_LOGLEVEL', 'DEBUG').upper())
asyncio.run(main())

45
examples/run_a2dp_info.py
View File

@@ -27,12 +27,9 @@ from bumble.core import (
BT_BR_EDR_TRANSPORT,
BT_AVDTP_PROTOCOL_ID,
BT_AUDIO_SINK_SERVICE,
BT_L2CAP_PROTOCOL_ID
)
from bumble.avdtp import (
Protocol as AVDTP_Protocol,
find_avdtp_service_with_connection
BT_L2CAP_PROTOCOL_ID,
)
from bumble.avdtp import Protocol as AVDTP_Protocol, find_avdtp_service_with_connection
from bumble.a2dp import make_audio_source_service_sdp_records
from bumble.sdp import (
Client as SDP_Client,
@@ -40,7 +37,7 @@ from bumble.sdp import (
DataElement,
SDP_PROTOCOL_DESCRIPTOR_LIST_ATTRIBUTE_ID,
SDP_BLUETOOTH_PROFILE_DESCRIPTOR_LIST_ATTRIBUTE_ID,
SDP_SERVICE_CLASS_ID_LIST_ATTRIBUTE_ID
SDP_SERVICE_CLASS_ID_LIST_ATTRIBUTE_ID,
)
@@ -48,7 +45,9 @@ from bumble.sdp import (
def sdp_records():
service_record_handle = 0x00010001
return {
service_record_handle: make_audio_source_service_sdp_records(service_record_handle)
service_record_handle: make_audio_source_service_sdp_records(
service_record_handle
)
}
@@ -64,8 +63,8 @@ async def find_a2dp_service(device, connection):
[
SDP_PROTOCOL_DESCRIPTOR_LIST_ATTRIBUTE_ID,
SDP_BLUETOOTH_PROFILE_DESCRIPTOR_LIST_ATTRIBUTE_ID,
SDP_SERVICE_CLASS_ID_LIST_ATTRIBUTE_ID
]
SDP_SERVICE_CLASS_ID_LIST_ATTRIBUTE_ID,
],
)
print(color('==================================', 'blue'))
@@ -78,8 +77,7 @@ async def find_a2dp_service(device, connection):
# Service classes
service_class_id_list = ServiceAttribute.find_attribute_in_list(
attribute_list,
SDP_SERVICE_CLASS_ID_LIST_ATTRIBUTE_ID
attribute_list, SDP_SERVICE_CLASS_ID_LIST_ATTRIBUTE_ID
)
if service_class_id_list:
if service_class_id_list.value:
@@ -89,8 +87,7 @@ async def find_a2dp_service(device, connection):
# Protocol info
protocol_descriptor_list = ServiceAttribute.find_attribute_in_list(
attribute_list,
SDP_PROTOCOL_DESCRIPTOR_LIST_ATTRIBUTE_ID
attribute_list, SDP_PROTOCOL_DESCRIPTOR_LIST_ATTRIBUTE_ID
)
if protocol_descriptor_list:
print(color(' Protocol:', 'green'))
@@ -103,18 +100,24 @@ async def find_a2dp_service(device, connection):
if len(protocol_descriptor.value) >= 2:
avdtp_version_major = protocol_descriptor.value[1].value >> 8
avdtp_version_minor = protocol_descriptor.value[1].value & 0xFF
print(f'{color(" AVDTP Version:", "cyan")} {avdtp_version_major}.{avdtp_version_minor}')
print(
f'{color(" AVDTP Version:", "cyan")} {avdtp_version_major}.{avdtp_version_minor}'
)
service_version = (avdtp_version_major, avdtp_version_minor)
# Profile info
bluetooth_profile_descriptor_list = ServiceAttribute.find_attribute_in_list(
attribute_list,
SDP_BLUETOOTH_PROFILE_DESCRIPTOR_LIST_ATTRIBUTE_ID
attribute_list, SDP_BLUETOOTH_PROFILE_DESCRIPTOR_LIST_ATTRIBUTE_ID
)
if bluetooth_profile_descriptor_list:
if bluetooth_profile_descriptor_list.value:
if bluetooth_profile_descriptor_list.value[0].type == DataElement.SEQUENCE:
bluetooth_profile_descriptors = bluetooth_profile_descriptor_list.value
if (
bluetooth_profile_descriptor_list.value[0].type
== DataElement.SEQUENCE
):
bluetooth_profile_descriptors = (
bluetooth_profile_descriptor_list.value
)
else:
# Sometimes, instead of a list of lists, we just find a list. Fix that
bluetooth_profile_descriptors = [bluetooth_profile_descriptor_list]
@@ -123,7 +126,9 @@ async def find_a2dp_service(device, connection):
for bluetooth_profile_descriptor in bluetooth_profile_descriptors:
version_major = bluetooth_profile_descriptor.value[1].value >> 8
version_minor = bluetooth_profile_descriptor.value[1].value & 0xFF
print(f' {bluetooth_profile_descriptor.value[0].value} - version {version_major}.{version_minor}')
print(
f' {bluetooth_profile_descriptor.value[0].value} - version {version_major}.{version_minor}'
)
await sdp_client.disconnect()
return service_version
@@ -184,5 +189,5 @@ async def main():
# -----------------------------------------------------------------------------
logging.basicConfig(level = os.environ.get('BUMBLE_LOGLEVEL', 'DEBUG').upper())
logging.basicConfig(level=os.environ.get('BUMBLE_LOGLEVEL', 'DEBUG').upper())
asyncio.run(main())

57
examples/run_a2dp_sink.py
View File

@@ -28,7 +28,7 @@ from bumble.avdtp import (
AVDTP_AUDIO_MEDIA_TYPE,
Protocol,
Listener,
MediaCodecCapabilities
MediaCodecCapabilities,
)
from bumble.a2dp import (
make_audio_sink_service_sdp_records,
@@ -39,19 +39,19 @@ from bumble.a2dp import (
SBC_LOUDNESS_ALLOCATION_METHOD,
SBC_STEREO_CHANNEL_MODE,
SBC_JOINT_STEREO_CHANNEL_MODE,
SbcMediaCodecInformation
SbcMediaCodecInformation,
)
Context = {
'output': None
}
Context = {'output': None}
# -----------------------------------------------------------------------------
def sdp_records():
service_record_handle = 0x00010001
return {
service_record_handle: make_audio_sink_service_sdp_records(service_record_handle)
service_record_handle: make_audio_sink_service_sdp_records(
service_record_handle
)
}
@@ -59,22 +59,25 @@ def sdp_records():
def codec_capabilities():
# NOTE: this shouldn't be hardcoded, but passed on the command line instead
return MediaCodecCapabilities(
media_type = AVDTP_AUDIO_MEDIA_TYPE,
media_codec_type = A2DP_SBC_CODEC_TYPE,
media_codec_information = SbcMediaCodecInformation.from_lists(
sampling_frequencies = [48000, 44100, 32000, 16000],
channel_modes = [
media_type=AVDTP_AUDIO_MEDIA_TYPE,
media_codec_type=A2DP_SBC_CODEC_TYPE,
media_codec_information=SbcMediaCodecInformation.from_lists(
sampling_frequencies=[48000, 44100, 32000, 16000],
channel_modes=[
SBC_MONO_CHANNEL_MODE,
SBC_DUAL_CHANNEL_MODE,
SBC_STEREO_CHANNEL_MODE,
SBC_JOINT_STEREO_CHANNEL_MODE
SBC_JOINT_STEREO_CHANNEL_MODE,
],
block_lengths = [4, 8, 12, 16],
subbands = [4, 8],
allocation_methods = [SBC_LOUDNESS_ALLOCATION_METHOD, SBC_SNR_ALLOCATION_METHOD],
minimum_bitpool_value = 2,
maximum_bitpool_value = 53
)
block_lengths=[4, 8, 12, 16],
subbands=[4, 8],
allocation_methods=[
SBC_LOUDNESS_ALLOCATION_METHOD,
SBC_SNR_ALLOCATION_METHOD,
],
minimum_bitpool_value=2,
maximum_bitpool_value=53,
),
)
@@ -87,10 +90,10 @@ def on_avdtp_connection(server):
# -----------------------------------------------------------------------------
def on_rtp_packet(packet):
header = packet.payload[0]
fragmented = header >> 7
start = (header >> 6) & 0x01
last = (header >> 5) & 0x01
header = packet.payload[0]
fragmented = header >> 7
start = (header >> 6) & 0x01
last = (header >> 5) & 0x01
number_of_frames = header & 0x0F
if fragmented:
@@ -104,7 +107,9 @@ def on_rtp_packet(packet):
# -----------------------------------------------------------------------------
async def main():
if len(sys.argv) < 4:
print('Usage: run_a2dp_sink.py <device-config> <transport-spec> <sbc-file> [<bt-addr>]')
print(
'Usage: run_a2dp_sink.py <device-config> <transport-spec> <sbc-file> [<bt-addr>]'
)
print('example: run_a2dp_sink.py classic1.json usb:0 output.sbc')
return
@@ -133,7 +138,9 @@ async def main():
# Connect to the source
target_address = sys.argv[4]
print(f'=== Connecting to {target_address}...')
connection = await device.connect(target_address, transport=BT_BR_EDR_TRANSPORT)
connection = await device.connect(
target_address, transport=BT_BR_EDR_TRANSPORT
)
print(f'=== Connected to {connection.peer_address}!')
# Request authentication
@@ -159,5 +166,5 @@ async def main():
# -----------------------------------------------------------------------------
logging.basicConfig(level = os.environ.get('BUMBLE_LOGLEVEL', 'DEBUG').upper())
logging.basicConfig(level=os.environ.get('BUMBLE_LOGLEVEL', 'DEBUG').upper())
asyncio.run(main())

64
examples/run_a2dp_source.py
View File

@@ -30,7 +30,7 @@ from bumble.avdtp import (
MediaCodecCapabilities,
MediaPacketPump,
Protocol,
Listener
Listener,
)
from bumble.a2dp import (
SBC_JOINT_STEREO_CHANNEL_MODE,
@@ -38,7 +38,7 @@ from bumble.a2dp import (
make_audio_source_service_sdp_records,
A2DP_SBC_CODEC_TYPE,
SbcMediaCodecInformation,
SbcPacketSource
SbcPacketSource,
)
@@ -46,7 +46,9 @@ from bumble.a2dp import (
def sdp_records():
service_record_handle = 0x00010001
return {
service_record_handle: make_audio_source_service_sdp_records(service_record_handle)
service_record_handle: make_audio_source_service_sdp_records(
service_record_handle
)
}
@@ -54,23 +56,25 @@ def sdp_records():
def codec_capabilities():
# NOTE: this shouldn't be hardcoded, but should be inferred from the input file instead
return MediaCodecCapabilities(
media_type = AVDTP_AUDIO_MEDIA_TYPE,
media_codec_type = A2DP_SBC_CODEC_TYPE,
media_codec_information = SbcMediaCodecInformation.from_discrete_values(
sampling_frequency = 44100,
channel_mode = SBC_JOINT_STEREO_CHANNEL_MODE,
block_length = 16,
subbands = 8,
allocation_method = SBC_LOUDNESS_ALLOCATION_METHOD,
minimum_bitpool_value = 2,
maximum_bitpool_value = 53
)
media_type=AVDTP_AUDIO_MEDIA_TYPE,
media_codec_type=A2DP_SBC_CODEC_TYPE,
media_codec_information=SbcMediaCodecInformation.from_discrete_values(
sampling_frequency=44100,
channel_mode=SBC_JOINT_STEREO_CHANNEL_MODE,
block_length=16,
subbands=8,
allocation_method=SBC_LOUDNESS_ALLOCATION_METHOD,
minimum_bitpool_value=2,
maximum_bitpool_value=53,
),
)
# -----------------------------------------------------------------------------
def on_avdtp_connection(read_function, protocol):
packet_source = SbcPacketSource(read_function, protocol.l2cap_channel.mtu, codec_capabilities())
packet_source = SbcPacketSource(
read_function, protocol.l2cap_channel.mtu, codec_capabilities()
)
packet_pump = MediaPacketPump(packet_source.packets)
protocol.add_source(packet_source.codec_capabilities, packet_pump)
@@ -83,14 +87,18 @@ async def stream_packets(read_function, protocol):
print('@@@', endpoint)
# Select a sink
sink = protocol.find_remote_sink_by_codec(AVDTP_AUDIO_MEDIA_TYPE, A2DP_SBC_CODEC_TYPE)
sink = protocol.find_remote_sink_by_codec(
AVDTP_AUDIO_MEDIA_TYPE, A2DP_SBC_CODEC_TYPE
)
if sink is None:
print(color('!!! no SBC sink found', 'red'))
return
print(f'### Selected sink: {sink.seid}')
# Stream the packets
packet_source = SbcPacketSource(read_function, protocol.l2cap_channel.mtu, codec_capabilities())
packet_source = SbcPacketSource(
read_function, protocol.l2cap_channel.mtu, codec_capabilities()
)
packet_pump = MediaPacketPump(packet_source.packets)
source = protocol.add_source(packet_source.codec_capabilities, packet_pump)
stream = await protocol.create_stream(source, sink)
@@ -107,8 +115,12 @@ async def stream_packets(read_function, protocol):
# -----------------------------------------------------------------------------
async def main():
if len(sys.argv) < 4:
print('Usage: run_a2dp_source.py <device-config> <transport-spec> <sbc-file> [<bluetooth-address>]')
print('example: run_a2dp_source.py classic1.json usb:0 test.sbc E1:CA:72:48:C4:E8')
print(
'Usage: run_a2dp_source.py <device-config> <transport-spec> <sbc-file> [<bluetooth-address>]'
)
print(
'example: run_a2dp_source.py classic1.json usb:0 test.sbc E1:CA:72:48:C4:E8'
)
return
print('<<< connecting to HCI...')
@@ -134,7 +146,9 @@ async def main():
# Connect to a peer
target_address = sys.argv[4]
print(f'=== Connecting to {target_address}...')
connection = await device.connect(target_address, transport=BT_BR_EDR_TRANSPORT)
connection = await device.connect(
target_address, transport=BT_BR_EDR_TRANSPORT
)
print(f'=== Connected to {connection.peer_address}!')
# Request authentication
@@ -148,7 +162,9 @@ async def main():
print('*** Encryption on')
# Look for an A2DP service
avdtp_version = await find_avdtp_service_with_connection(device, connection)
avdtp_version = await find_avdtp_service_with_connection(
device, connection
)
if not avdtp_version:
print(color('!!! no A2DP service found'))
return
@@ -161,7 +177,9 @@ async def main():
else:
# Create a listener to wait for AVDTP connections
listener = Listener(Listener.create_registrar(device), version=(1, 2))
listener.on('connection', lambda protocol: on_avdtp_connection(read, protocol))
listener.on(
'connection', lambda protocol: on_avdtp_connection(read, protocol)
)
# Become connectable and wait for a connection
await device.set_discoverable(True)
@@ -171,5 +189,5 @@ async def main():
# -----------------------------------------------------------------------------
logging.basicConfig(level = os.environ.get('BUMBLE_LOGLEVEL', 'DEBUG').upper())
logging.basicConfig(level=os.environ.get('BUMBLE_LOGLEVEL', 'DEBUG').upper())
asyncio.run(main())

7
examples/run_advertiser.py
View File

@@ -30,7 +30,9 @@ from bumble.transport import open_transport_or_link
# -----------------------------------------------------------------------------
async def main():
if len(sys.argv) < 3:
print('Usage: run_advertiser.py <config-file> <transport-spec> [type] [address]')
print(
'Usage: run_advertiser.py <config-file> <transport-spec> [type] [address]'
)
print('example: run_advertiser.py device1.json usb:0')
return
@@ -56,6 +58,7 @@ async def main():
await device.start_advertising(advertising_type=advertising_type, target=target)
await hci_source.wait_for_termination()
# -----------------------------------------------------------------------------
logging.basicConfig(level = os.environ.get('BUMBLE_LOGLEVEL', 'DEBUG').upper())
logging.basicConfig(level=os.environ.get('BUMBLE_LOGLEVEL', 'DEBUG').upper())
asyncio.run(main())

135
examples/run_asha_sink.py
View File

@@ -25,28 +25,34 @@ from bumble.core import AdvertisingData
from bumble.device import Device
from bumble.transport import open_transport_or_link
from bumble.hci import UUID
from bumble.gatt import (
Service,
Characteristic,
CharacteristicValue
)
from bumble.gatt import Service, Characteristic, CharacteristicValue
# -----------------------------------------------------------------------------
# Constants
# -----------------------------------------------------------------------------
ASHA_SERVICE = UUID.from_16_bits(0xFDF0, 'Audio Streaming for Hearing Aid')
ASHA_READ_ONLY_PROPERTIES_CHARACTERISTIC = UUID('6333651e-c481-4a3e-9169-7c902aad37bb', 'ReadOnlyProperties')
ASHA_AUDIO_CONTROL_POINT_CHARACTERISTIC = UUID('f0d4de7e-4a88-476c-9d9f-1937b0996cc0', 'AudioControlPoint')
ASHA_AUDIO_STATUS_CHARACTERISTIC = UUID('38663f1a-e711-4cac-b641-326b56404837', 'AudioStatus')
ASHA_VOLUME_CHARACTERISTIC = UUID('00e4ca9e-ab14-41e4-8823-f9e70c7e91df', 'Volume')
ASHA_LE_PSM_OUT_CHARACTERISTIC = UUID('2d410339-82b6-42aa-b34e-e2e01df8cc1a', 'LE_PSM_OUT')
ASHA_SERVICE = UUID.from_16_bits(0xFDF0, 'Audio Streaming for Hearing Aid')
ASHA_READ_ONLY_PROPERTIES_CHARACTERISTIC = UUID(
'6333651e-c481-4a3e-9169-7c902aad37bb', 'ReadOnlyProperties'
)
ASHA_AUDIO_CONTROL_POINT_CHARACTERISTIC = UUID(
'f0d4de7e-4a88-476c-9d9f-1937b0996cc0', 'AudioControlPoint'
)
ASHA_AUDIO_STATUS_CHARACTERISTIC = UUID(
'38663f1a-e711-4cac-b641-326b56404837', 'AudioStatus'
)
ASHA_VOLUME_CHARACTERISTIC = UUID('00e4ca9e-ab14-41e4-8823-f9e70c7e91df', 'Volume')
ASHA_LE_PSM_OUT_CHARACTERISTIC = UUID(
'2d410339-82b6-42aa-b34e-e2e01df8cc1a', 'LE_PSM_OUT'
)
# -----------------------------------------------------------------------------
async def main():
if len(sys.argv) != 4:
print('Usage: python run_asha_sink.py <device-config> <transport-spec> <audio-file>')
print(
'Usage: python run_asha_sink.py <device-config> <transport-spec> <audio-file>'
)
print('example: python run_asha_sink.py device1.json usb:0 audio_out.g722')
return
@@ -62,14 +68,18 @@ async def main():
if opcode == 1:
# Start
audio_type = ('Unknown', 'Ringtone', 'Phone Call', 'Media')[value[2]]
print(f'### START: codec={value[1]}, audio_type={audio_type}, volume={value[3]}, otherstate={value[4]}')
print(
f'### START: codec={value[1]}, audio_type={audio_type}, volume={value[3]}, otherstate={value[4]}'
)
elif opcode == 2:
print('### STOP')
elif opcode == 3:
print(f'### STATUS: connected={value[1]}')
# Respond with a status
asyncio.create_task(device.notify_subscribers(audio_status_characteristic, force=True))
asyncio.create_task(
device.notify_subscribers(audio_status_characteristic, force=True)
)
# Handler for volume control
def on_volume_write(connection, value):
@@ -91,63 +101,91 @@ async def main():
ASHA_READ_ONLY_PROPERTIES_CHARACTERISTIC,
Characteristic.READ,
Characteristic.READABLE,
bytes([
0x01, # Version
0x00, # Device Capabilities [Left, Monaural]
0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07, 0x08, # HiSyncId
0x01, # Feature Map [LE CoC audio output streaming supported]
0x00, 0x00, # Render Delay
0x00, 0x00, # RFU
0x02, 0x00 # Codec IDs [G.722 at 16 kHz]
])
bytes(
[
0x01, # Version
0x00, # Device Capabilities [Left, Monaural]
0x01,
0x02,
0x03,
0x04,
0x05,
0x06,
0x07,
0x08, # HiSyncId
0x01, # Feature Map [LE CoC audio output streaming supported]
0x00,
0x00, # Render Delay
0x00,
0x00, # RFU
0x02,
0x00, # Codec IDs [G.722 at 16 kHz]
]
),
)
audio_control_point_characteristic = Characteristic(
ASHA_AUDIO_CONTROL_POINT_CHARACTERISTIC,
Characteristic.WRITE | Characteristic.WRITE_WITHOUT_RESPONSE,
Characteristic.WRITEABLE,
CharacteristicValue(write=on_audio_control_point_write)
CharacteristicValue(write=on_audio_control_point_write),
)
audio_status_characteristic = Characteristic(
ASHA_AUDIO_STATUS_CHARACTERISTIC,
Characteristic.READ | Characteristic.NOTIFY,
Characteristic.READABLE,
bytes([0])
bytes([0]),
)
volume_characteristic = Characteristic(
ASHA_VOLUME_CHARACTERISTIC,
Characteristic.WRITE_WITHOUT_RESPONSE,
Characteristic.WRITEABLE,
CharacteristicValue(write=on_volume_write)
CharacteristicValue(write=on_volume_write),
)
le_psm_out_characteristic = Characteristic(
ASHA_LE_PSM_OUT_CHARACTERISTIC,
Characteristic.READ,
Characteristic.READABLE,
struct.pack('<H', psm)
struct.pack('<H', psm),
)
device.add_service(
Service(
ASHA_SERVICE,
[
read_only_properties_characteristic,
audio_control_point_characteristic,
audio_status_characteristic,
volume_characteristic,
le_psm_out_characteristic,
],
)
)
device.add_service(Service(
ASHA_SERVICE,
[
read_only_properties_characteristic,
audio_control_point_characteristic,
audio_status_characteristic,
volume_characteristic,
le_psm_out_characteristic
]
))
# Set the advertising data
device.advertising_data = bytes(
AdvertisingData([
(AdvertisingData.COMPLETE_LOCAL_NAME, bytes(device.name, 'utf-8')),
(AdvertisingData.FLAGS, bytes([0x06])),
(AdvertisingData.INCOMPLETE_LIST_OF_16_BIT_SERVICE_CLASS_UUIDS, bytes(ASHA_SERVICE)),
(AdvertisingData.SERVICE_DATA_16_BIT_UUID, bytes(ASHA_SERVICE) + bytes([
0x01, # Protocol Version
0x00, # Capability
0x01, 0x02, 0x03, 0x04 # Truncated HiSyncID
]))
])
AdvertisingData(
[
(AdvertisingData.COMPLETE_LOCAL_NAME, bytes(device.name, 'utf-8')),
(AdvertisingData.FLAGS, bytes([0x06])),
(
AdvertisingData.INCOMPLETE_LIST_OF_16_BIT_SERVICE_CLASS_UUIDS,
bytes(ASHA_SERVICE),
),
(
AdvertisingData.SERVICE_DATA_16_BIT_UUID,
bytes(ASHA_SERVICE)
+ bytes(
[
0x01, # Protocol Version
0x00, # Capability
0x01,
0x02,
0x03,
0x04, # Truncated HiSyncID
]
),
),
]
)
)
# Go!
@@ -156,6 +194,7 @@ async def main():
await hci_source.wait_for_termination()
# -----------------------------------------------------------------------------
logging.basicConfig(level = os.environ.get('BUMBLE_LOGLEVEL', 'DEBUG').upper())
logging.basicConfig(level=os.environ.get('BUMBLE_LOGLEVEL', 'DEBUG').upper())
asyncio.run(main())

43
examples/run_classic_connect.py
View File

@@ -24,14 +24,22 @@ from colors import color
from bumble.device import Device
from bumble.transport import open_transport_or_link
from bumble.core import BT_BR_EDR_TRANSPORT, BT_L2CAP_PROTOCOL_ID
from bumble.sdp import Client as SDP_Client, SDP_PUBLIC_BROWSE_ROOT, SDP_ALL_ATTRIBUTES_RANGE
from bumble.sdp import (
Client as SDP_Client,
SDP_PUBLIC_BROWSE_ROOT,
SDP_ALL_ATTRIBUTES_RANGE,
)
# -----------------------------------------------------------------------------
async def main():
if len(sys.argv) < 3:
print('Usage: run_classic_connect.py <device-config> <transport-spec> <bluetooth-addresses..>')
print('example: run_classic_connect.py classic1.json usb:04b4:f901 E1:CA:72:48:C4:E8')
print(
'Usage: run_classic_connect.py <device-config> <transport-spec> <bluetooth-addresses..>'
)
print(
'example: run_classic_connect.py classic1.json usb:04b4:f901 E1:CA:72:48:C4:E8'
)
return
print('<<< connecting to HCI...')
@@ -53,32 +61,49 @@ async def main():
await sdp_client.connect(connection)
# List all services in the root browse group
service_record_handles = await sdp_client.search_services([SDP_PUBLIC_BROWSE_ROOT])
service_record_handles = await sdp_client.search_services(
[SDP_PUBLIC_BROWSE_ROOT]
)
print(color('\n==================================', 'blue'))
print(color('SERVICES:', 'yellow'), service_record_handles)
# For each service in the root browse group, get all its attributes
for service_record_handle in service_record_handles:
attributes = await sdp_client.get_attributes(service_record_handle, [SDP_ALL_ATTRIBUTES_RANGE])
attributes = await sdp_client.get_attributes(
service_record_handle, [SDP_ALL_ATTRIBUTES_RANGE]
)
print(color(f'SERVICE {service_record_handle:04X} attributes:', 'yellow'))
for attribute in attributes:
print(' ', attribute.to_string(color=True))
# Search for services with an L2CAP service attribute
search_result = await sdp_client.search_attributes([BT_L2CAP_PROTOCOL_ID], [SDP_ALL_ATTRIBUTES_RANGE])
search_result = await sdp_client.search_attributes(
[BT_L2CAP_PROTOCOL_ID], [SDP_ALL_ATTRIBUTES_RANGE]
)
print(color('\n==================================', 'blue'))
print(color('SEARCH RESULTS:', 'yellow'))
for attribute_list in search_result:
print(color('SERVICE:', 'green'))
print(' ' + '\n '.join([attribute.to_string(color=True) for attribute in attribute_list]))
print(
' '
+ '\n '.join(
[attribute.to_string(color=True) for attribute in attribute_list]
)
)
await sdp_client.disconnect()
await hci_source.wait_for_termination()
# Connect to a peer
target_addresses = sys.argv[3:]
await asyncio.wait([asyncio.create_task(connect(target_address)) for target_address in target_addresses])
await asyncio.wait(
[
asyncio.create_task(connect(target_address))
for target_address in target_addresses
]
)
# -----------------------------------------------------------------------------
logging.basicConfig(level = os.environ.get('BUMBLE_LOGLEVEL', 'DEBUG').upper())
logging.basicConfig(level=os.environ.get('BUMBLE_LOGLEVEL', 'DEBUG').upper())
asyncio.run(main())

65
examples/run_classic_discoverable.py
View File

@@ -30,48 +30,62 @@ from bumble.sdp import (
SDP_SERVICE_RECORD_HANDLE_ATTRIBUTE_ID,
SDP_SERVICE_CLASS_ID_LIST_ATTRIBUTE_ID,
SDP_PROTOCOL_DESCRIPTOR_LIST_ATTRIBUTE_ID,
SDP_BLUETOOTH_PROFILE_DESCRIPTOR_LIST_ATTRIBUTE_ID
SDP_BLUETOOTH_PROFILE_DESCRIPTOR_LIST_ATTRIBUTE_ID,
)
from bumble.core import (
BT_AUDIO_SINK_SERVICE,
BT_L2CAP_PROTOCOL_ID,
BT_AVDTP_PROTOCOL_ID,
BT_ADVANCED_AUDIO_DISTRIBUTION_SERVICE
BT_ADVANCED_AUDIO_DISTRIBUTION_SERVICE,
)
# -----------------------------------------------------------------------------
SDP_SERVICE_RECORDS = {
0x00010001: [
ServiceAttribute(SDP_SERVICE_RECORD_HANDLE_ATTRIBUTE_ID, DataElement.unsigned_integer_32(0x00010001)),
ServiceAttribute(SDP_BROWSE_GROUP_LIST_ATTRIBUTE_ID, DataElement.sequence([
DataElement.uuid(SDP_PUBLIC_BROWSE_ROOT)
])),
ServiceAttribute(
SDP_SERVICE_RECORD_HANDLE_ATTRIBUTE_ID,
DataElement.unsigned_integer_32(0x00010001),
),
ServiceAttribute(
SDP_BROWSE_GROUP_LIST_ATTRIBUTE_ID,
DataElement.sequence([DataElement.uuid(SDP_PUBLIC_BROWSE_ROOT)]),
),
ServiceAttribute(
SDP_SERVICE_CLASS_ID_LIST_ATTRIBUTE_ID,
DataElement.sequence([DataElement.uuid(BT_AUDIO_SINK_SERVICE)])
DataElement.sequence([DataElement.uuid(BT_AUDIO_SINK_SERVICE)]),
),
ServiceAttribute(
SDP_PROTOCOL_DESCRIPTOR_LIST_ATTRIBUTE_ID,
DataElement.sequence([
DataElement.sequence([
DataElement.uuid(BT_L2CAP_PROTOCOL_ID),
DataElement.unsigned_integer_16(25)
]),
DataElement.sequence([
DataElement.uuid(BT_AVDTP_PROTOCOL_ID),
DataElement.unsigned_integer_16(256)
])
])
DataElement.sequence(
[
DataElement.sequence(
[
DataElement.uuid(BT_L2CAP_PROTOCOL_ID),
DataElement.unsigned_integer_16(25),
]
),
DataElement.sequence(
[
DataElement.uuid(BT_AVDTP_PROTOCOL_ID),
DataElement.unsigned_integer_16(256),
]
),
]
),
),
ServiceAttribute(
SDP_BLUETOOTH_PROFILE_DESCRIPTOR_LIST_ATTRIBUTE_ID,
DataElement.sequence([
DataElement.sequence([
DataElement.uuid(BT_ADVANCED_AUDIO_DISTRIBUTION_SERVICE),
DataElement.unsigned_integer_16(256)
])
])
)
DataElement.sequence(
[
DataElement.sequence(
[
DataElement.uuid(BT_ADVANCED_AUDIO_DISTRIBUTION_SERVICE),
DataElement.unsigned_integer_16(256),
]
)
]
),
),
]
}
@@ -99,6 +113,7 @@ async def main():
await hci_source.wait_for_termination()
# -----------------------------------------------------------------------------
logging.basicConfig(level = os.environ.get('BUMBLE_LOGLEVEL', 'DEBUG').upper())
logging.basicConfig(level=os.environ.get('BUMBLE_LOGLEVEL', 'DEBUG').upper())
asyncio.run(main())

21
examples/run_classic_discovery.py
View File

@@ -29,13 +29,23 @@ from bumble.core import DeviceClass
# -----------------------------------------------------------------------------
class DiscoveryListener(Device.Listener):
def on_inquiry_result(self, address, class_of_device, eir_data, rssi):
service_classes, major_device_class, minor_device_class = DeviceClass.split_class_of_device(class_of_device)
(
service_classes,
major_device_class,
minor_device_class,
) = DeviceClass.split_class_of_device(class_of_device)
separator = '\n '
print(f'>>> {color(address, "yellow")}:')
print(f' Device Class (raw): {class_of_device:06X}')
print(f' Device Major Class: {DeviceClass.major_device_class_name(major_device_class)}')
print(f' Device Minor Class: {DeviceClass.minor_device_class_name(major_device_class, minor_device_class)}')
print(f' Device Services: {", ".join(DeviceClass.service_class_labels(service_classes))}')
print(
f' Device Major Class: {DeviceClass.major_device_class_name(major_device_class)}'
)
print(
f' Device Minor Class: {DeviceClass.minor_device_class_name(major_device_class, minor_device_class)}'
)
print(
f' Device Services: {", ".join(DeviceClass.service_class_labels(service_classes))}'
)
print(f' RSSI: {rssi}')
if eir_data.ad_structures:
print(f' {eir_data.to_string(separator)}')
@@ -59,6 +69,7 @@ async def main():
await hci_source.wait_for_termination()
# -----------------------------------------------------------------------------
logging.basicConfig(level = os.environ.get('BUMBLE_LOGLEVEL', 'DEBUG').upper())
logging.basicConfig(level=os.environ.get('BUMBLE_LOGLEVEL', 'DEBUG').upper())
asyncio.run(main())

11
examples/run_connect_and_encrypt.py
View File

@@ -27,8 +27,12 @@ from bumble.transport import open_transport_or_link
# -----------------------------------------------------------------------------
async def main():
if len(sys.argv) < 3:
print('Usage: run_connect_and_encrypt.py <device-config> <transport-spec> <bluetooth-address>')
print('example: run_connect_and_encrypt.py device1.json usb:0 E1:CA:72:48:C4:E8')
print(
'Usage: run_connect_and_encrypt.py <device-config> <transport-spec> <bluetooth-address>'
)
print(
'example: run_connect_and_encrypt.py device1.json usb:0 E1:CA:72:48:C4:E8'
)
return
print('<<< connecting to HCI...')
@@ -53,6 +57,7 @@ async def main():
await hci_source.wait_for_termination()
# -----------------------------------------------------------------------------
logging.basicConfig(level = os.environ.get('BUMBLE_LOGLEVEL', 'DEBUG').upper())
logging.basicConfig(level=os.environ.get('BUMBLE_LOGLEVEL', 'DEBUG').upper())
asyncio.run(main())

31
examples/run_controller.py
View File

@@ -32,8 +32,12 @@ from bumble.transport import open_transport_or_link
# -----------------------------------------------------------------------------
async def main():
if len(sys.argv) != 4:
print('Usage: run_controller.py <controller-address> <device-config> <transport-spec>')
print('example: run_controller.py F2:F3:F4:F5:F6:F7 device1.json udp:0.0.0.0:22333,172.16.104.161:22333')
print(
'Usage: run_controller.py <controller-address> <device-config> <transport-spec>'
)
print(
'example: run_controller.py F2:F3:F4:F5:F6:F7 device1.json udp:0.0.0.0:22333,172.16.104.161:22333'
)
return
print('>>> connecting to HCI...')
@@ -44,11 +48,13 @@ async def main():
link = LocalLink()
# Create a first controller using the packet source/sink as its host interface
controller1 = Controller('C1', host_source = hci_source, host_sink = hci_sink, link = link)
controller1 = Controller(
'C1', host_source=hci_source, host_sink=hci_sink, link=link
)
controller1.random_address = sys.argv[1]
# Create a second controller using the same link
controller2 = Controller('C2', link = link)
controller2 = Controller('C2', link=link)
# Create a host for the second controller
host = Host()
@@ -59,17 +65,21 @@ async def main():
device.host = host
# Add some basic services to the device's GATT server
descriptor = Descriptor(GATT_CHARACTERISTIC_USER_DESCRIPTION_DESCRIPTOR, Descriptor.READABLE, 'My Description')
descriptor = Descriptor(
GATT_CHARACTERISTIC_USER_DESCRIPTION_DESCRIPTOR,
Descriptor.READABLE,
'My Description',
)
manufacturer_name_characteristic = Characteristic(
GATT_MANUFACTURER_NAME_STRING_CHARACTERISTIC,
Characteristic.READ,
Characteristic.READABLE,
"Fitbit",
[descriptor]
[descriptor],
)
device_info_service = Service(
GATT_DEVICE_INFORMATION_SERVICE, [manufacturer_name_characteristic]
)
device_info_service = Service(GATT_DEVICE_INFORMATION_SERVICE, [
manufacturer_name_characteristic
])
device.add_service(device_info_service)
# Debug print
@@ -82,6 +92,7 @@ async def main():
await hci_source.wait_for_termination()
# -----------------------------------------------------------------------------
logging.basicConfig(level = os.environ.get('BUMBLE_LOGLEVEL', 'DEBUG').upper())
logging.basicConfig(level=os.environ.get('BUMBLE_LOGLEVEL', 'DEBUG').upper())
asyncio.run(main())

17
examples/run_controller_with_scanner.py
View File

@@ -37,7 +37,9 @@ class ScannerListener(Device.Listener):
else:
type_color = 'cyan'
print(f'>>> {color(advertisement.address, address_color)} [{color(address_type_string, type_color)}]: RSSI={advertisement.rssi}, {advertisement.data}')
print(
f'>>> {color(advertisement.address, address_color)} [{color(address_type_string, type_color)}]: RSSI={advertisement.rssi}, {advertisement.data}'
)
# -----------------------------------------------------------------------------
@@ -55,20 +57,25 @@ async def main():
link = LocalLink()
# Create a first controller using the packet source/sink as its host interface
controller1 = Controller('C1', host_source = hci_source, host_sink = hci_sink, link = link)
controller1 = Controller(
'C1', host_source=hci_source, host_sink=hci_sink, link=link
)
controller1.address = 'E0:E1:E2:E3:E4:E5'
# Create a second controller using the same link
controller2 = Controller('C2', link = link)
controller2 = Controller('C2', link=link)
# Create a device with a scanner listener
device = Device.with_hci('Bumble', 'F0:F1:F2:F3:F4:F5', controller2, controller2)
device = Device.with_hci(
'Bumble', 'F0:F1:F2:F3:F4:F5', controller2, controller2
)
device.listener = ScannerListener()
await device.power_on()
await device.start_scanning()
await hci_source.wait_for_termination()
# -----------------------------------------------------------------------------
logging.basicConfig(level = os.environ.get('BUMBLE_LOGLEVEL', 'DEBUG').upper())
logging.basicConfig(level=os.environ.get('BUMBLE_LOGLEVEL', 'DEBUG').upper())
asyncio.run(main())

7
examples/run_gatt_client.py
View File

@@ -70,7 +70,9 @@ class Listener(Device.Listener):
# -----------------------------------------------------------------------------
async def main():
if len(sys.argv) < 3:
print('Usage: run_gatt_client.py <device-config> <transport-spec> [<bluetooth-address>]')
print(
'Usage: run_gatt_client.py <device-config> <transport-spec> [<bluetooth-address>]'
)
print('example: run_gatt_client.py device1.json usb:0 E1:CA:72:48:C4:E8')
return
@@ -93,6 +95,7 @@ async def main():
await asyncio.get_running_loop().create_future()
# -----------------------------------------------------------------------------
logging.basicConfig(level = os.environ.get('BUMBLE_LOGLEVEL', 'DEBUG').upper())
logging.basicConfig(level=os.environ.get('BUMBLE_LOGLEVEL', 'DEBUG').upper())
asyncio.run(main())

27
examples/run_gatt_client_and_server.py
View File

@@ -32,7 +32,7 @@ from bumble.gatt import (
show_services,
GATT_CHARACTERISTIC_USER_DESCRIPTION_DESCRIPTOR,
GATT_MANUFACTURER_NAME_STRING_CHARACTERISTIC,
GATT_DEVICE_INFORMATION_SERVICE
GATT_DEVICE_INFORMATION_SERVICE,
)
@@ -48,32 +48,36 @@ async def main():
link = LocalLink()
# Setup a stack for the client
client_controller = Controller("client controller", link = link)
client_controller = Controller("client controller", link=link)
client_host = Host()
client_host.controller = client_controller
client_device = Device("client", address = 'F0:F1:F2:F3:F4:F5', host = client_host)
client_device = Device("client", address='F0:F1:F2:F3:F4:F5', host=client_host)
await client_device.power_on()
# Setup a stack for the server
server_controller = Controller("server controller", link = link)
server_controller = Controller("server controller", link=link)
server_host = Host()
server_host.controller = server_controller
server_device = Device("server", address = 'F6:F7:F8:F9:FA:FB', host = server_host)
server_device = Device("server", address='F6:F7:F8:F9:FA:FB', host=server_host)
server_device.listener = ServerListener()
await server_device.power_on()
# Add a few entries to the device's GATT server
descriptor = Descriptor(GATT_CHARACTERISTIC_USER_DESCRIPTION_DESCRIPTOR, Descriptor.READABLE, 'My Description')
descriptor = Descriptor(
GATT_CHARACTERISTIC_USER_DESCRIPTION_DESCRIPTOR,
Descriptor.READABLE,
'My Description',
)
manufacturer_name_characteristic = Characteristic(
GATT_MANUFACTURER_NAME_STRING_CHARACTERISTIC,
Characteristic.READ,
Characteristic.READABLE,
"Fitbit",
[descriptor]
[descriptor],
)
device_info_service = Service(
GATT_DEVICE_INFORMATION_SERVICE, [manufacturer_name_characteristic]
)
device_info_service = Service(GATT_DEVICE_INFORMATION_SERVICE, [
manufacturer_name_characteristic
])
server_device.add_service(device_info_service)
# Connect the client to the server
@@ -109,6 +113,7 @@ async def main():
await asyncio.get_running_loop().create_future()
# -----------------------------------------------------------------------------
logging.basicConfig(level = os.environ.get('BUMBLE_LOGLEVEL', 'DEBUG').upper())
logging.basicConfig(level=os.environ.get('BUMBLE_LOGLEVEL', 'DEBUG').upper())
asyncio.run(main())

40
examples/run_gatt_server.py
View File

@@ -22,10 +22,7 @@ import logging
from bumble.device import Device, Connection
from bumble.transport import open_transport_or_link
from bumble.att import (
ATT_Error,
ATT_INSUFFICIENT_ENCRYPTION_ERROR
)
from bumble.att import ATT_Error, ATT_INSUFFICIENT_ENCRYPTION_ERROR
from bumble.gatt import (
Service,
Characteristic,
@@ -33,7 +30,7 @@ from bumble.gatt import (
Descriptor,
GATT_CHARACTERISTIC_USER_DESCRIPTION_DESCRIPTOR,
GATT_MANUFACTURER_NAME_STRING_CHARACTERISTIC,
GATT_DEVICE_INFORMATION_SERVICE
GATT_DEVICE_INFORMATION_SERVICE,
)
@@ -76,7 +73,9 @@ def my_custom_write_with_error(connection, value):
# -----------------------------------------------------------------------------
async def main():
if len(sys.argv) < 3:
print('Usage: run_gatt_server.py <device-config> <transport-spec> [<bluetooth-address>]')
print(
'Usage: run_gatt_server.py <device-config> <transport-spec> [<bluetooth-address>]'
)
print('example: run_gatt_server.py device1.json usb:0 E1:CA:72:48:C4:E8')
return
@@ -89,17 +88,21 @@ async def main():
device.listener = Listener(device)
# Add a few entries to the device's GATT server
descriptor = Descriptor(GATT_CHARACTERISTIC_USER_DESCRIPTION_DESCRIPTOR, Descriptor.READABLE, 'My Description')
descriptor = Descriptor(
GATT_CHARACTERISTIC_USER_DESCRIPTION_DESCRIPTOR,
Descriptor.READABLE,
'My Description',
)
manufacturer_name_characteristic = Characteristic(
GATT_MANUFACTURER_NAME_STRING_CHARACTERISTIC,
Characteristic.READ,
Characteristic.READABLE,
'Fitbit',
[descriptor]
[descriptor],
)
device_info_service = Service(
GATT_DEVICE_INFORMATION_SERVICE, [manufacturer_name_characteristic]
)
device_info_service = Service(GATT_DEVICE_INFORMATION_SERVICE, [
manufacturer_name_characteristic
])
custom_service1 = Service(
'50DB505C-8AC4-4738-8448-3B1D9CC09CC5',
[
@@ -107,21 +110,23 @@ async def main():
'D901B45B-4916-412E-ACCA-376ECB603B2C',
Characteristic.READ | Characteristic.WRITE,
Characteristic.READABLE | Characteristic.WRITEABLE,
CharacteristicValue(read=my_custom_read, write=my_custom_write)
CharacteristicValue(read=my_custom_read, write=my_custom_write),
),
Characteristic(
'552957FB-CF1F-4A31-9535-E78847E1A714',
Characteristic.READ | Characteristic.WRITE,
Characteristic.READABLE | Characteristic.WRITEABLE,
CharacteristicValue(read=my_custom_read_with_error, write=my_custom_write_with_error)
CharacteristicValue(
read=my_custom_read_with_error, write=my_custom_write_with_error
),
),
Characteristic(
'486F64C6-4B5F-4B3B-8AFF-EDE134A8446A',
Characteristic.READ | Characteristic.NOTIFY,
Characteristic.READABLE,
'hello'
)
]
'hello',
),
],
)
device.add_services([device_info_service, custom_service1])
@@ -142,6 +147,7 @@ async def main():
await hci_source.wait_for_termination()
# -----------------------------------------------------------------------------
logging.basicConfig(level = os.environ.get('BUMBLE_LOGLEVEL', 'DEBUG').upper())
logging.basicConfig(level=os.environ.get('BUMBLE_LOGLEVEL', 'DEBUG').upper())
asyncio.run(main())

83
examples/run_hfp_gateway.py
View File

@@ -31,12 +31,9 @@ from bumble.sdp import (
ServiceAttribute,
SDP_PROTOCOL_DESCRIPTOR_LIST_ATTRIBUTE_ID,
SDP_SERVICE_CLASS_ID_LIST_ATTRIBUTE_ID,
SDP_BLUETOOTH_PROFILE_DESCRIPTOR_LIST_ATTRIBUTE_ID
)
from bumble.hci import (
BT_HANDSFREE_SERVICE,
BT_RFCOMM_PROTOCOL_ID
SDP_BLUETOOTH_PROFILE_DESCRIPTOR_LIST_ATTRIBUTE_ID,
)
from bumble.hci import BT_HANDSFREE_SERVICE, BT_RFCOMM_PROTOCOL_ID
from bumble.hfp import HfpProtocol
@@ -52,8 +49,8 @@ async def list_rfcomm_channels(device, connection):
[
SDP_PROTOCOL_DESCRIPTOR_LIST_ATTRIBUTE_ID,
SDP_BLUETOOTH_PROFILE_DESCRIPTOR_LIST_ATTRIBUTE_ID,
SDP_SERVICE_CLASS_ID_LIST_ATTRIBUTE_ID
]
SDP_SERVICE_CLASS_ID_LIST_ATTRIBUTE_ID,
],
)
print(color('==================================', 'blue'))
print(color('Handsfree Services:', 'yellow'))
@@ -61,40 +58,59 @@ async def list_rfcomm_channels(device, connection):
for attribute_list in search_result:
# Look for the RFCOMM Channel number
protocol_descriptor_list = ServiceAttribute.find_attribute_in_list(
attribute_list,
SDP_PROTOCOL_DESCRIPTOR_LIST_ATTRIBUTE_ID
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:
print(color('SERVICE:', 'green'))
print(color(' RFCOMM Channel:', 'cyan'), protocol_descriptor.value[1].value)
print(
color(' RFCOMM Channel:', 'cyan'),
protocol_descriptor.value[1].value,
)
rfcomm_channels.append(protocol_descriptor.value[1].value)
# List profiles
bluetooth_profile_descriptor_list = ServiceAttribute.find_attribute_in_list(
attribute_list,
SDP_BLUETOOTH_PROFILE_DESCRIPTOR_LIST_ATTRIBUTE_ID
bluetooth_profile_descriptor_list = (
ServiceAttribute.find_attribute_in_list(
attribute_list,
SDP_BLUETOOTH_PROFILE_DESCRIPTOR_LIST_ATTRIBUTE_ID,
)
)
if bluetooth_profile_descriptor_list:
if bluetooth_profile_descriptor_list.value:
if bluetooth_profile_descriptor_list.value[0].type == DataElement.SEQUENCE:
bluetooth_profile_descriptors = bluetooth_profile_descriptor_list.value
if (
bluetooth_profile_descriptor_list.value[0].type
== DataElement.SEQUENCE
):
bluetooth_profile_descriptors = (
bluetooth_profile_descriptor_list.value
)
else:
# Sometimes, instead of a list of lists, we just find a list. Fix that
bluetooth_profile_descriptors = [bluetooth_profile_descriptor_list]
bluetooth_profile_descriptors = [
bluetooth_profile_descriptor_list
]
print(color(' Profiles:', 'green'))
for bluetooth_profile_descriptor in bluetooth_profile_descriptors:
version_major = bluetooth_profile_descriptor.value[1].value >> 8
version_minor = bluetooth_profile_descriptor.value[1].value & 0xFF
print(f' {bluetooth_profile_descriptor.value[0].value} - version {version_major}.{version_minor}')
for (
bluetooth_profile_descriptor
) in bluetooth_profile_descriptors:
version_major = (
bluetooth_profile_descriptor.value[1].value >> 8
)
version_minor = (
bluetooth_profile_descriptor.value[1].value
& 0xFF
)
print(
f' {bluetooth_profile_descriptor.value[0].value} - version {version_major}.{version_minor}'
)
# List service classes
service_class_id_list = ServiceAttribute.find_attribute_in_list(
attribute_list,
SDP_SERVICE_CLASS_ID_LIST_ATTRIBUTE_ID
attribute_list, SDP_SERVICE_CLASS_ID_LIST_ATTRIBUTE_ID
)
if service_class_id_list:
if service_class_id_list.value:
@@ -109,9 +125,15 @@ async def list_rfcomm_channels(device, connection):
# -----------------------------------------------------------------------------
async def main():
if len(sys.argv) < 4:
print('Usage: run_hfp_gateway.py <device-config> <transport-spec> <bluetooth-address>')
print(' specifying a channel number, or "discover" to list all RFCOMM channels')
print('example: run_hfp_gateway.py hfp_gateway.json usb:04b4:f901 E1:CA:72:48:C4:E8')
print(
'Usage: run_hfp_gateway.py <device-config> <transport-spec> <bluetooth-address>'
)
print(
' specifying a channel number, or "discover" to list all RFCOMM channels'
)
print(
'example: run_hfp_gateway.py hfp_gateway.json usb:04b4:f901 E1:CA:72:48:C4:E8'
)
return
print('<<< connecting to HCI...')
@@ -173,7 +195,9 @@ async def main():
protocol.send_response_line('+BRSF: 30')
protocol.send_response_line('OK')
elif line.startswith('AT+CIND=?'):
protocol.send_response_line('+CIND: ("call",(0,1)),("callsetup",(0-3)),("service",(0-1)),("signal",(0-5)),("roam",(0,1)),("battchg",(0-5)),("callheld",(0-2))')
protocol.send_response_line(
'+CIND: ("call",(0,1)),("callsetup",(0-3)),("service",(0-1)),("signal",(0-5)),("roam",(0,1)),("battchg",(0-5)),("callheld",(0-2))'
)
protocol.send_response_line('OK')
elif line.startswith('AT+CIND?'):
protocol.send_response_line('+CIND: 0,0,1,4,1,5,0')
@@ -193,7 +217,9 @@ async def main():
elif line.startswith('AT+BIA='):
protocol.send_response_line('OK')
elif line.startswith('AT+BVRA='):
protocol.send_response_line('+BVRA: 1,1,12AA,1,1,"Message 1 from Janina"')
protocol.send_response_line(
'+BVRA: 1,1,12AA,1,1,"Message 1 from Janina"'
)
elif line.startswith('AT+XEVENT='):
protocol.send_response_line('OK')
elif line.startswith('AT+XAPL='):
@@ -204,6 +230,7 @@ async def main():
await hci_source.wait_for_termination()
# -----------------------------------------------------------------------------
logging.basicConfig(level = os.environ.get('BUMBLE_LOGLEVEL', 'DEBUG').upper())
logging.basicConfig(level=os.environ.get('BUMBLE_LOGLEVEL', 'DEBUG').upper())
asyncio.run(main())

60
examples/run_hfp_handsfree.py
View File

@@ -32,13 +32,13 @@ from bumble.sdp import (
SDP_SERVICE_RECORD_HANDLE_ATTRIBUTE_ID,
SDP_SERVICE_CLASS_ID_LIST_ATTRIBUTE_ID,
SDP_PROTOCOL_DESCRIPTOR_LIST_ATTRIBUTE_ID,
SDP_BLUETOOTH_PROFILE_DESCRIPTOR_LIST_ATTRIBUTE_ID
SDP_BLUETOOTH_PROFILE_DESCRIPTOR_LIST_ATTRIBUTE_ID,
)
from bumble.core import (
BT_GENERIC_AUDIO_SERVICE,
BT_HANDSFREE_SERVICE,
BT_L2CAP_PROTOCOL_ID,
BT_RFCOMM_PROTOCOL_ID
BT_RFCOMM_PROTOCOL_ID,
)
from bumble.hfp import HfpProtocol
@@ -49,36 +49,44 @@ def make_sdp_records(rfcomm_channel):
0x00010001: [
ServiceAttribute(
SDP_SERVICE_RECORD_HANDLE_ATTRIBUTE_ID,
DataElement.unsigned_integer_32(0x00010001)
DataElement.unsigned_integer_32(0x00010001),
),
ServiceAttribute(
SDP_SERVICE_CLASS_ID_LIST_ATTRIBUTE_ID,
DataElement.sequence([
DataElement.uuid(BT_HANDSFREE_SERVICE),
DataElement.uuid(BT_GENERIC_AUDIO_SERVICE)
])
DataElement.sequence(
[
DataElement.uuid(BT_HANDSFREE_SERVICE),
DataElement.uuid(BT_GENERIC_AUDIO_SERVICE),
]
),
),
ServiceAttribute(
SDP_PROTOCOL_DESCRIPTOR_LIST_ATTRIBUTE_ID,
DataElement.sequence([
DataElement.sequence([
DataElement.uuid(BT_L2CAP_PROTOCOL_ID)
]),
DataElement.sequence([
DataElement.uuid(BT_RFCOMM_PROTOCOL_ID),
DataElement.unsigned_integer_8(rfcomm_channel)
])
])
DataElement.sequence(
[
DataElement.sequence([DataElement.uuid(BT_L2CAP_PROTOCOL_ID)]),
DataElement.sequence(
[
DataElement.uuid(BT_RFCOMM_PROTOCOL_ID),
DataElement.unsigned_integer_8(rfcomm_channel),
]
),
]
),
),
ServiceAttribute(
SDP_BLUETOOTH_PROFILE_DESCRIPTOR_LIST_ATTRIBUTE_ID,
DataElement.sequence([
DataElement.sequence([
DataElement.uuid(BT_HANDSFREE_SERVICE),
DataElement.unsigned_integer_16(0x0105)
])
])
)
DataElement.sequence(
[
DataElement.sequence(
[
DataElement.uuid(BT_HANDSFREE_SERVICE),
DataElement.unsigned_integer_16(0x0105),
]
)
]
),
),
]
}
@@ -103,6 +111,7 @@ class UiServer:
except websockets.exceptions.ConnectionClosedOK:
pass
await websockets.serve(serve, 'localhost', 8989)
@@ -111,7 +120,7 @@ async def protocol_loop(protocol):
await protocol.initialize_service()
while True:
await(protocol.next_line())
await (protocol.next_line())
# -----------------------------------------------------------------------------
@@ -160,6 +169,7 @@ async def main():
await hci_source.wait_for_termination()
# -----------------------------------------------------------------------------
logging.basicConfig(level = os.environ.get('BUMBLE_LOGLEVEL', 'DEBUG').upper())
logging.basicConfig(level=os.environ.get('BUMBLE_LOGLEVEL', 'DEBUG').upper())
asyncio.run(main())

20
examples/run_notifier.py
View File

@@ -23,10 +23,7 @@ import logging
from bumble.device import Device, Connection
from bumble.transport import open_transport_or_link
from bumble.gatt import (
Service,
Characteristic
)
from bumble.gatt import Service, Characteristic
# -----------------------------------------------------------------------------
@@ -41,7 +38,9 @@ class Listener(Device.Listener, Connection.Listener):
def on_disconnection(self, reason):
print(f'### Disconnected, reason={reason}')
def on_characteristic_subscription(self, connection, characteristic, notify_enabled, indicate_enabled):
def on_characteristic_subscription(
self, connection, characteristic, notify_enabled, indicate_enabled
):
print(
f'$$$ Characteristic subscription for handle {characteristic.handle} from {connection}: '
f'notify {"enabled" if notify_enabled else "disabled"}, '
@@ -55,6 +54,7 @@ class Listener(Device.Listener, Connection.Listener):
def on_my_characteristic_subscription(peer, enabled):
print(f'### My characteristic from {peer}: {"enabled" if enabled else "disabled"}')
# -----------------------------------------------------------------------------
async def main():
if len(sys.argv) < 3:
@@ -75,24 +75,24 @@ async def main():
'486F64C6-4B5F-4B3B-8AFF-EDE134A8446A',
Characteristic.READ | Characteristic.NOTIFY,
Characteristic.READABLE,
bytes([0x40])
bytes([0x40]),
)
characteristic2 = Characteristic(
'8EBDEBAE-0017-418E-8D3B-3A3809492165',
Characteristic.READ | Characteristic.INDICATE,
Characteristic.READABLE,
bytes([0x41])
bytes([0x41]),
)
characteristic3 = Characteristic(
'8EBDEBAE-0017-418E-8D3B-3A3809492165',
Characteristic.READ | Characteristic.NOTIFY | Characteristic.INDICATE,
Characteristic.READABLE,
bytes([0x42])
bytes([0x42]),
)
characteristic3.on('subscription', on_my_characteristic_subscription)
custom_service = Service(
'50DB505C-8AC4-4738-8448-3B1D9CC09CC5',
[characteristic1, characteristic2, characteristic3]
[characteristic1, characteristic2, characteristic3],
)
device.add_services([custom_service])
@@ -123,5 +123,5 @@ async def main():
# -----------------------------------------------------------------------------
logging.basicConfig(level = os.environ.get('BUMBLE_LOGLEVEL', 'DEBUG').upper())
logging.basicConfig(level=os.environ.get('BUMBLE_LOGLEVEL', 'DEBUG').upper())
asyncio.run(main())

70
examples/run_rfcomm_client.py
View File

@@ -31,7 +31,7 @@ from bumble.sdp import (
ServiceAttribute,
SDP_PROTOCOL_DESCRIPTOR_LIST_ATTRIBUTE_ID,
SDP_SERVICE_CLASS_ID_LIST_ATTRIBUTE_ID,
SDP_BLUETOOTH_PROFILE_DESCRIPTOR_LIST_ATTRIBUTE_ID
SDP_BLUETOOTH_PROFILE_DESCRIPTOR_LIST_ATTRIBUTE_ID,
)
from bumble.hci import BT_L2CAP_PROTOCOL_ID, BT_RFCOMM_PROTOCOL_ID
@@ -48,47 +48,66 @@ async def list_rfcomm_channels(device, connection):
[
SDP_PROTOCOL_DESCRIPTOR_LIST_ATTRIBUTE_ID,
SDP_BLUETOOTH_PROFILE_DESCRIPTOR_LIST_ATTRIBUTE_ID,
SDP_SERVICE_CLASS_ID_LIST_ATTRIBUTE_ID
]
SDP_SERVICE_CLASS_ID_LIST_ATTRIBUTE_ID,
],
)
print(color('==================================', 'blue'))
print(color('RFCOMM Services:', 'yellow'))
for attribute_list in search_result:
# Look for the RFCOMM Channel number
protocol_descriptor_list = ServiceAttribute.find_attribute_in_list(
attribute_list,
SDP_PROTOCOL_DESCRIPTOR_LIST_ATTRIBUTE_ID
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:
print(color('SERVICE:', 'green'))
print(color(' RFCOMM Channel:', 'cyan'), protocol_descriptor.value[1].value)
print(
color(' RFCOMM Channel:', 'cyan'),
protocol_descriptor.value[1].value,
)
# List profiles
bluetooth_profile_descriptor_list = ServiceAttribute.find_attribute_in_list(
attribute_list,
SDP_BLUETOOTH_PROFILE_DESCRIPTOR_LIST_ATTRIBUTE_ID
bluetooth_profile_descriptor_list = (
ServiceAttribute.find_attribute_in_list(
attribute_list,
SDP_BLUETOOTH_PROFILE_DESCRIPTOR_LIST_ATTRIBUTE_ID,
)
)
if bluetooth_profile_descriptor_list:
if bluetooth_profile_descriptor_list.value:
if bluetooth_profile_descriptor_list.value[0].type == DataElement.SEQUENCE:
bluetooth_profile_descriptors = bluetooth_profile_descriptor_list.value
if (
bluetooth_profile_descriptor_list.value[0].type
== DataElement.SEQUENCE
):
bluetooth_profile_descriptors = (
bluetooth_profile_descriptor_list.value
)
else:
# Sometimes, instead of a list of lists, we just find a list. Fix that
bluetooth_profile_descriptors = [bluetooth_profile_descriptor_list]
bluetooth_profile_descriptors = [
bluetooth_profile_descriptor_list
]
print(color(' Profiles:', 'green'))
for bluetooth_profile_descriptor in bluetooth_profile_descriptors:
version_major = bluetooth_profile_descriptor.value[1].value >> 8
version_minor = bluetooth_profile_descriptor.value[1].value & 0xFF
print(f' {bluetooth_profile_descriptor.value[0].value} - version {version_major}.{version_minor}')
for (
bluetooth_profile_descriptor
) in bluetooth_profile_descriptors:
version_major = (
bluetooth_profile_descriptor.value[1].value >> 8
)
version_minor = (
bluetooth_profile_descriptor.value[1].value
& 0xFF
)
print(
f' {bluetooth_profile_descriptor.value[0].value} - version {version_major}.{version_minor}'
)
# List service classes
service_class_id_list = ServiceAttribute.find_attribute_in_list(
attribute_list,
SDP_SERVICE_CLASS_ID_LIST_ATTRIBUTE_ID
attribute_list, SDP_SERVICE_CLASS_ID_LIST_ATTRIBUTE_ID
)
if service_class_id_list:
if service_class_id_list.value:
@@ -138,9 +157,15 @@ async def tcp_server(tcp_port, rfcomm_session):
# -----------------------------------------------------------------------------
async def main():
if len(sys.argv) < 5:
print('Usage: run_rfcomm_client.py <device-config> <transport-spec> <bluetooth-address> <channel>|discover [tcp-port]')
print(' specifying a channel number, or "discover" to list all RFCOMM channels')
print('example: run_rfcomm_client.py classic1.json usb:04b4:f901 E1:CA:72:48:C4:E8 8')
print(
'Usage: run_rfcomm_client.py <device-config> <transport-spec> <bluetooth-address> <channel>|discover [tcp-port]'
)
print(
' specifying a channel number, or "discover" to list all RFCOMM channels'
)
print(
'example: run_rfcomm_client.py classic1.json usb:04b4:f901 E1:CA:72:48:C4:E8 8'
)
return
print('<<< connecting to HCI...')
@@ -197,6 +222,7 @@ async def main():
await hci_source.wait_for_termination()
# -----------------------------------------------------------------------------
logging.basicConfig(level = os.environ.get('BUMBLE_LOGLEVEL', 'DEBUG').upper())
logging.basicConfig(level=os.environ.get('BUMBLE_LOGLEVEL', 'DEBUG').upper())
asyncio.run(main())

49
examples/run_rfcomm_server.py
View File

@@ -31,7 +31,7 @@ from bumble.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_PROTOCOL_DESCRIPTOR_LIST_ATTRIBUTE_ID,
)
from bumble.hci import BT_L2CAP_PROTOCOL_ID, BT_RFCOMM_PROTOCOL_ID
@@ -40,22 +40,34 @@ from bumble.hci import BT_L2CAP_PROTOCOL_ID, BT_RFCOMM_PROTOCOL_ID
def sdp_records(channel):
return {
0x00010001: [
ServiceAttribute(SDP_SERVICE_RECORD_HANDLE_ATTRIBUTE_ID, DataElement.unsigned_integer_32(0x00010001)),
ServiceAttribute(SDP_BROWSE_GROUP_LIST_ATTRIBUTE_ID, DataElement.sequence([
DataElement.uuid(SDP_PUBLIC_BROWSE_ROOT)
])),
ServiceAttribute(SDP_SERVICE_CLASS_ID_LIST_ATTRIBUTE_ID, DataElement.sequence([
DataElement.uuid(UUID('E6D55659-C8B4-4B85-96BB-B1143AF6D3AE'))
])),
ServiceAttribute(SDP_PROTOCOL_DESCRIPTOR_LIST_ATTRIBUTE_ID, DataElement.sequence([
DataElement.sequence([
DataElement.uuid(BT_L2CAP_PROTOCOL_ID)
]),
DataElement.sequence([
DataElement.uuid(BT_RFCOMM_PROTOCOL_ID),
DataElement.unsigned_integer_8(channel)
])
]))
ServiceAttribute(
SDP_SERVICE_RECORD_HANDLE_ATTRIBUTE_ID,
DataElement.unsigned_integer_32(0x00010001),
),
ServiceAttribute(
SDP_BROWSE_GROUP_LIST_ATTRIBUTE_ID,
DataElement.sequence([DataElement.uuid(SDP_PUBLIC_BROWSE_ROOT)]),
),
ServiceAttribute(
SDP_SERVICE_CLASS_ID_LIST_ATTRIBUTE_ID,
DataElement.sequence(
[DataElement.uuid(UUID('E6D55659-C8B4-4B85-96BB-B1143AF6D3AE'))]
),
),
ServiceAttribute(
SDP_PROTOCOL_DESCRIPTOR_LIST_ATTRIBUTE_ID,
DataElement.sequence(
[
DataElement.sequence([DataElement.uuid(BT_L2CAP_PROTOCOL_ID)]),
DataElement.sequence(
[
DataElement.uuid(BT_RFCOMM_PROTOCOL_ID),
DataElement.unsigned_integer_8(channel),
]
),
]
),
),
]
}
@@ -113,6 +125,7 @@ async def main():
await hci_source.wait_for_termination()
# -----------------------------------------------------------------------------
logging.basicConfig(level = os.environ.get('BUMBLE_LOGLEVEL', 'DEBUG').upper())
logging.basicConfig(level=os.environ.get('BUMBLE_LOGLEVEL', 'DEBUG').upper())
asyncio.run(main())

13
examples/run_scanner.py
View File

@@ -35,13 +35,15 @@ async def main():
print('<<< connecting to HCI...')
async with await open_transport_or_link(sys.argv[1]) as (hci_source, hci_sink):
print('<<< connected')
filter_duplicates = (len(sys.argv) == 3 and sys.argv[2] == 'filter')
filter_duplicates = len(sys.argv) == 3 and sys.argv[2] == 'filter'
device = Device.with_hci('Bumble', 'F0:F1:F2:F3:F4:F5', hci_source, hci_sink)
@device.on('advertisement')
def _(advertisement):
address_type_string = ('PUBLIC', 'RANDOM', 'PUBLIC_ID', 'RANDOM_ID')[advertisement.address.address_type]
address_type_string = ('PUBLIC', 'RANDOM', 'PUBLIC_ID', 'RANDOM_ID')[
advertisement.address.address_type
]
address_color = 'yellow' if advertisement.is_connectable else 'red'
address_qualifier = ''
if address_type_string.startswith('P'):
@@ -57,13 +59,16 @@ async def main():
type_color = 'white'
separator = '\n '
print(f'>>> {color(advertisement.address, address_color)} [{color(address_type_string, type_color)}]{address_qualifier}:{separator}RSSI:{advertisement.rssi}{separator}{advertisement.data.to_string(separator)}')
print(
f'>>> {color(advertisement.address, address_color)} [{color(address_type_string, type_color)}]{address_qualifier}:{separator}RSSI:{advertisement.rssi}{separator}{advertisement.data.to_string(separator)}'
)
await device.power_on()
await device.start_scanning(filter_duplicates=filter_duplicates)
await hci_source.wait_for_termination()
# -----------------------------------------------------------------------------
logging.basicConfig(level = os.environ.get('BUMBLE_LOGLEVEL', 'DEBUG').upper())
logging.basicConfig(level=os.environ.get('BUMBLE_LOGLEVEL', 'DEBUG').upper())
asyncio.run(main())