Skip to content

Commit 3b0aa71

Browse files
kawasakikawasaki
committed
example: Add async example of GATT interface
Add an example script which shows properties of services and characteristics that the specified BLE device provides. I confirmed this script using micro:bit with Scratch HEX installed. $ python examples/sdbus_async/gatt_props.py XX:XX:XX:XX:XX:XX Wait discovery for 5 seconds Conneting to /org/bluez/hci0/dev_XX_XX_XX_XX_XX_XX... Conneted Find services... service0008 service000c service0013 ============================================= service: service0008 UUID: 00001801-0000-1000-8000-00805f9b34fb device: /org/bluez/hci0/dev_XX_XX_XX_XX_XX_XX primary: True includes paths: [] missing propeties: ['handle'] char0009 UUID: 00002a05-0000-1000-8000-00805f9b34fb service: /org/bluez/hci0/dev_XX_XX_XX_XX_XX_XX/service0008 value: b'' flags: ['indicate'] notifying: False MTU: 23 missing propeties: ['write acquired', 'notify acquired', 'handle'] ============================================= ... Signed-off-by: Shin'ichiro Kawasaki <kawasaki@juno.dti.ne.jp>
1 parent 3856ee3 commit 3b0aa71

1 file changed

Lines changed: 192 additions & 0 deletions

File tree

‎examples/sdbus_async/gatt_props.py‎

Lines changed: 192 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,192 @@
1+
# SPDX-License-Identifier: LGPL-2.1-or-later
2+
# Copyright (C) 2023 Shin'ichiro Kawasaki <kawasaki@juno.dti.ne.jp>
3+
4+
# This example shows how to get properties of GATT services and characteristics
5+
# of a BLE device.
6+
# It does:
7+
# - Discover the specified BLE device with BlueZ D-Bus Adapter API [1]
8+
# - Connect to the device with BlueZ D-Bus Device API [2]
9+
# - Get services and characteristics of the device with D-Bus introspection
10+
# - Get properties of services and characteristics with BlueZ D-Bus GATT API [3]
11+
# - Also read the characteristics if it has "read" flag.
12+
13+
# [1] https://git.kernel.org/pub/scm/bluetooth/bluez.git/tree/doc/adapter-api.txt
14+
# [2] https://git.kernel.org/pub/scm/bluetooth/bluez.git/tree/doc/device-api.txt
15+
# [3] https://git.kernel.org/pub/scm/bluetooth/bluez.git/tree/doc/gatt-api.txt
16+
17+
from __future__ import annotations
18+
19+
import sys
20+
import xml.etree.ElementTree as ET
21+
from asyncio import run, sleep
22+
from typing import Any
23+
24+
from sdbus import SdBus, sd_bus_open_system
25+
from sdbus.dbus_proxy_async_interfaces import DbusIntrospectableAsync
26+
27+
from sdbus_async.bluez.adapter_api import AdapterInterfaceAsync
28+
from sdbus_async.bluez.device_api import DeviceInterfaceAsync
29+
from sdbus_async.bluez.gatt_api import (
30+
GattCharacteristicInterfaceAsync,
31+
GattServiceInterfaceAsync,
32+
)
33+
34+
35+
def usage() -> None:
36+
print(f"Usage: {sys.argv[0]} DEV_ADDR")
37+
print("\tDEV_ADDR: Address of BLE device in fomrat XX:XX:XX:XX:XX:XX")
38+
exit(1)
39+
40+
41+
if len(sys.argv) < 2:
42+
usage()
43+
44+
DEV_ADDR = sys.argv[1]
45+
DEV_PATH = '/org/bluez/hci0/dev_' + DEV_ADDR.replace(":", "_")
46+
DISCOVERY_SECONDS = 5
47+
48+
49+
async def discover(dbus: SdBus) -> None:
50+
adapter = AdapterInterfaceAsync()
51+
adapter._connect('org.bluez', '/org/bluez/hci0', bus=dbus)
52+
try:
53+
await adapter.start_discovery()
54+
print(f"Wait discovery for {DISCOVERY_SECONDS} seconds")
55+
await sleep(DISCOVERY_SECONDS)
56+
await adapter.stop_discovery()
57+
except NotImplementedError as e:
58+
print(e)
59+
except Exception as e:
60+
print(e)
61+
print(f"failure in discovery: {e}")
62+
63+
64+
async def print_props(prop_dict: dict[str, Any], indent: str) -> None:
65+
missing_properties = []
66+
for key in prop_dict.keys():
67+
try:
68+
print(f"{indent}{key}: {await prop_dict[key]}")
69+
except Exception:
70+
missing_properties.append(key)
71+
print("missing propeties:", missing_properties)
72+
73+
74+
async def print_char_props(dbus: SdBus, service: str, char_str: str) -> None:
75+
path = DEV_PATH + '/' + service + '/' + char_str
76+
char = GattCharacteristicInterfaceAsync()
77+
char._connect('org.bluez', path, bus=dbus)
78+
char_properties = {
79+
'UUID': char.uuid,
80+
'service': char.service_path,
81+
'value': char.value,
82+
'flags': char.flags,
83+
'write acquired': char.write_acquired,
84+
'notify acquired': char.notify_acquired,
85+
'notifying': char.notifying,
86+
'handle': char.handle,
87+
'MTU': char.mtu,
88+
}
89+
await print_props(char_properties, " ")
90+
flags: list[str] = await char.flags
91+
for flag in flags:
92+
if flag == 'read':
93+
value = await char.read_value({})
94+
print(f"read value: {str(value)}")
95+
96+
97+
async def print_service_props(dbus: SdBus, service_name: str) -> None:
98+
path = DEV_PATH + '/' + service_name
99+
print('=============================================')
100+
print(f"service: {service_name}")
101+
service = GattServiceInterfaceAsync()
102+
service._connect('org.bluez', path, bus=dbus)
103+
service_properties = {
104+
'UUID': service.uuid,
105+
'device': service.device_path,
106+
'primary': service.primary,
107+
'includes paths': service.includes_paths,
108+
'handle': service.handle,
109+
}
110+
await print_props(service_properties, "")
111+
112+
chars = await find_chars(dbus, path)
113+
if chars:
114+
for char in chars:
115+
print(f" {char}")
116+
await print_char_props(dbus, service_name, char)
117+
return None
118+
119+
120+
async def find_chars(dbus: SdBus, path: str) -> list[str]:
121+
# do D-Bus introspect to the service path and get characteristic paths
122+
adapter_introspect = DbusIntrospectableAsync()
123+
adapter_introspect._connect('org.bluez', path, bus=dbus)
124+
s = await adapter_introspect.dbus_introspect()
125+
parser = ET.fromstring(s)
126+
nodes = parser.findall("./node")
127+
if not nodes:
128+
print("characteristic not found")
129+
return []
130+
131+
chars = []
132+
for node in nodes:
133+
chars.append(node.attrib['name'])
134+
return chars
135+
136+
137+
async def find_service(dbus: SdBus) -> list[str]:
138+
# do D-Bus introspect to the device path and get service paths under it
139+
adapter_introspect = DbusIntrospectableAsync()
140+
adapter_introspect._connect('org.bluez', DEV_PATH, bus=dbus)
141+
s = await adapter_introspect.dbus_introspect()
142+
parser = ET.fromstring(s)
143+
nodes = parser.findall("./node")
144+
if not nodes:
145+
print("service not found")
146+
return []
147+
148+
services = []
149+
for node in nodes:
150+
print(f" {node.attrib['name']}")
151+
services.append(node.attrib['name'])
152+
return services
153+
154+
155+
async def main() -> None:
156+
157+
# connect to D-Bus
158+
dbus = sd_bus_open_system()
159+
device = DeviceInterfaceAsync()
160+
device._connect('org.bluez', DEV_PATH, bus=dbus)
161+
162+
# discover the specified BLE device
163+
await discover(dbus)
164+
165+
# connect to the device
166+
print(f"Conneting to {DEV_PATH}...")
167+
try:
168+
await device.connect()
169+
except NotImplementedError as e:
170+
print(e)
171+
return
172+
except Exception as e:
173+
print(e)
174+
print(f"failed to connect: {e}")
175+
return
176+
print("Conneted")
177+
178+
# Get services and characteristics and print their properties
179+
await sleep(2)
180+
for i in range(3):
181+
print("Find services...")
182+
services = await find_service(dbus)
183+
if services:
184+
for service in services:
185+
await print_service_props(dbus, service)
186+
break
187+
await sleep(1)
188+
189+
# Clean up
190+
await device.disconnect()
191+
192+
run(main())

0 commit comments

Comments
 (0)