Skip to content

Commit 1fa9b27

Browse files
igo95862igo95862
authored
Merge pull request #5 from kawasaki/dev
Thank you!
2 parents 3856ee3 + 3e4c17c commit 1fa9b27

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 DbusInterfaceCommonAsync, SdBus, sd_bus_open_system
25+
26+
from sdbus_async.bluez.adapter_api import AdapterInterfaceAsync
27+
from sdbus_async.bluez.device_api import DeviceInterfaceAsync
28+
from sdbus_async.bluez.gatt_api import (
29+
GattCharacteristicInterfaceAsync,
30+
GattServiceInterfaceAsync,
31+
)
32+
33+
34+
def usage() -> None:
35+
print(f"Usage: {sys.argv[0]} DEV_ADDR")
36+
print("\tDEV_ADDR: Address of BLE device in fomrat XX:XX:XX:XX:XX:XX")
37+
exit(1)
38+
39+
40+
DISCOVERY_SECONDS = 5
41+
42+
43+
async def discover(dbus: SdBus) -> None:
44+
adapter = AdapterInterfaceAsync()
45+
adapter._connect('org.bluez', '/org/bluez/hci0', bus=dbus)
46+
try:
47+
await adapter.start_discovery()
48+
print(f"Wait discovery for {DISCOVERY_SECONDS} seconds")
49+
await sleep(DISCOVERY_SECONDS)
50+
await adapter.stop_discovery()
51+
except NotImplementedError as e:
52+
print(e)
53+
except Exception as e:
54+
print(e)
55+
print(f"failure in discovery: {e}")
56+
57+
58+
async def print_props(prop_dict: dict[str, Any], indent: str) -> None:
59+
missing_properties = []
60+
for key in prop_dict.keys():
61+
try:
62+
print(f"{indent}{key}: {await prop_dict[key]}")
63+
except Exception:
64+
missing_properties.append(key)
65+
print("missing propeties:", missing_properties)
66+
67+
68+
async def print_char_props(dbus: SdBus, dev_path: str,
69+
service: str, char_str: str) -> None:
70+
path = dev_path + '/' + service + '/' + char_str
71+
char = GattCharacteristicInterfaceAsync()
72+
char._connect('org.bluez', path, bus=dbus)
73+
char_properties = {
74+
'UUID': char.uuid,
75+
'service': char.service_path,
76+
'value': char.value,
77+
'flags': char.flags,
78+
'write acquired': char.write_acquired,
79+
'notify acquired': char.notify_acquired,
80+
'notifying': char.notifying,
81+
'handle': char.handle,
82+
'MTU': char.mtu,
83+
}
84+
await print_props(char_properties, " ")
85+
flags: list[str] = await char.flags
86+
for flag in flags:
87+
if flag == 'read':
88+
value = await char.read_value({})
89+
print(f"read value: {str(value)}")
90+
91+
92+
async def print_service_props(dbus: SdBus, dev_path: str,
93+
service_name: str) -> None:
94+
path = dev_path + '/' + service_name
95+
print('=============================================')
96+
print(f"service: {service_name}")
97+
service = GattServiceInterfaceAsync()
98+
service._connect('org.bluez', path, bus=dbus)
99+
service_properties = {
100+
'UUID': service.uuid,
101+
'device': service.device_path,
102+
'primary': service.primary,
103+
'includes paths': service.includes_paths,
104+
'handle': service.handle,
105+
}
106+
await print_props(service_properties, "")
107+
108+
chars = await find_chars(dbus, path)
109+
if chars:
110+
for char in chars:
111+
print(f" {char}")
112+
await print_char_props(dbus, dev_path, service_name, char)
113+
return None
114+
115+
116+
async def find_chars(dbus: SdBus, path: str) -> list[str]:
117+
# do D-Bus introspect to the service path and get characteristic paths
118+
adapter_introspect = DbusInterfaceCommonAsync()
119+
adapter_introspect._connect('org.bluez', path, bus=dbus)
120+
s = await adapter_introspect.dbus_introspect()
121+
parser = ET.fromstring(s)
122+
nodes = parser.findall("./node")
123+
if not nodes:
124+
print("characteristic not found")
125+
return []
126+
127+
chars = []
128+
for node in nodes:
129+
chars.append(node.attrib['name'])
130+
return chars
131+
132+
133+
async def find_service(dbus: SdBus, dev_path: str) -> list[str]:
134+
# do D-Bus introspect to the device path and get service paths under it
135+
adapter_introspect = DbusInterfaceCommonAsync()
136+
adapter_introspect._connect('org.bluez', dev_path, bus=dbus)
137+
s = await adapter_introspect.dbus_introspect()
138+
parser = ET.fromstring(s)
139+
nodes = parser.findall("./node")
140+
if not nodes:
141+
print("service not found")
142+
return []
143+
144+
services = []
145+
for node in nodes:
146+
print(f" {node.attrib['name']}")
147+
services.append(node.attrib['name'])
148+
return services
149+
150+
151+
async def main(dev_path: str) -> None:
152+
153+
# connect to D-Bus
154+
dbus = sd_bus_open_system()
155+
device = DeviceInterfaceAsync()
156+
device._connect('org.bluez', dev_path, bus=dbus)
157+
158+
# discover the specified BLE device
159+
await discover(dbus)
160+
161+
# connect to the device
162+
print(f"Conneting to {dev_path}...")
163+
try:
164+
await device.connect()
165+
except NotImplementedError as e:
166+
print(e)
167+
return
168+
except Exception as e:
169+
print(e)
170+
print(f"failed to connect: {e}")
171+
return
172+
print("Conneted")
173+
174+
# Get services and characteristics and print their properties
175+
await sleep(2)
176+
for i in range(3):
177+
print("Find services...")
178+
services = await find_service(dbus, dev_path)
179+
if services:
180+
for service in services:
181+
await print_service_props(dbus, dev_path, service)
182+
break
183+
await sleep(1)
184+
185+
# Clean up
186+
await device.disconnect()
187+
188+
if __name__ == '__main__':
189+
if len(sys.argv) < 2:
190+
usage()
191+
dev_path = '/org/bluez/hci0/dev_' + sys.argv[1].replace(":", "_")
192+
run(main(dev_path))

0 commit comments

Comments
 (0)