1#!/usr/bin/env python3

2#

3# Copyright 2018 Canonical Ltd.

4# Written by:

5# Sylvain Pineau <sylvain.pineau@canonical.com>

6#

7# This is free software: you can redistribute it and/or modify

8# it under the terms of the GNU General Public License version 3,

9# as published by the Free Software Foundation.

10#

11# This file is distributed in the hope that it will be useful,

12# but WITHOUT ANY WARRANTY; without even the implied warranty of

13# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the

14# GNU General Public License for more details.

15#

16# You should have received a copy of the GNU General Public License

17# along with this file. If not, see <http://www.gnu.org/licenses/>.

18

19 import argparse

20 import logging

21 import os

22 import sys

23 import time

24

25 import dbus

26 import dbus.service

27 import dbus.mainloop.glib

28 from gi.repository import GObject

29

30 logger = logging.getLogger(__file__)

31 logger.addHandler(logging.StreamHandler(sys.stdout))

32

33 ADAPTER_INTERFACE = "org.bluez.Adapter1"

34 DEVICE_INTERFACE = "org.bluez.Device1"

35 PROP_INTERFACE = "org.freedesktop.DBus.Properties"

36 OM_INTERFACE = "org.freedesktop.DBus.ObjectManager"

37 GATT_SERVICE_INTERFACE = "org.bluez.GattService1"

38 GATT_CHRC_INTERFACE = "org.bluez.GattCharacteristic1"

39

40 dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)

41

42

43 class BtAdapter:

44 """Bluetooth LE Adapter class."""

45

46 def __init__(self, pattern):

47 self._pattern = os.path.basename(pattern)

48 self._bus = dbus.SystemBus()

49 self._manager = dbus.Interface(

50 self._bus.get_object("org.bluez", "/"), OM_INTERFACE

51 )

52 self._main_loop = GObject.MainLoop()

53 self._adapter = self._find_adapter()

54 self._path = self._adapter.object_path

55 self._props = dbus.Interface(self._adapter, PROP_INTERFACE)

56 self._name = self._props.Get(ADAPTER_INTERFACE, "Name")

57 self._addr = self._props.Get(ADAPTER_INTERFACE, "Address")

58 self._alias = self._props.Get(ADAPTER_INTERFACE, "Alias")

59 logger.info(

60 "Adapter found: [ {} ] {} - {}".format(

61 self._path, self._addr, self._alias

62 )

63 )

64

65 def _get_managed_objects(self):

66 return self._manager.GetManagedObjects()

67

68 def _find_adapter(self):

69 for path, ifaces in self._get_managed_objects().items():

70 adapter = ifaces.get(ADAPTER_INTERFACE)

71 if adapter is None:

72 continue

73 if self._pattern == adapter["Address"] or path.endswith(

74 self._pattern

75 ):

76 obj = self._bus.get_object("org.bluez", path)

77 return dbus.Interface(obj, ADAPTER_INTERFACE)

78 raise SystemExit("Bluetooth adapter not found!")

79

80 def ensure_powered(self):

81 """Turn the adapter on."""

82 self._props.Set(ADAPTER_INTERFACE, "Powered", dbus.Boolean(1))

83 logger.info("Adapter powered on")

84

85 def scan(self, timeout=10):

86 """Scan for BT devices."""

87 dbus.Interface(self._adapter, ADAPTER_INTERFACE).StartDiscovery()

88 logger.info("Adapter scan on ({}s)".format(timeout))

89 GObject.timeout_add_seconds(timeout, self._scan_timeout)

90 self._main_loop.run()

91

92 def _scan_timeout(self):

93 dbus.Interface(self._adapter, ADAPTER_INTERFACE).StopDiscovery()

94 logger.info("Adapter scan completed")

95 self._main_loop.quit()

96

97 def find_device_with_service(self, ADV_SVC_UUID):

98 """Find a device with a given remote service."""

99 for path, ifaces in self._get_managed_objects().items():

100 device = ifaces.get(DEVICE_INTERFACE)

101 if device is None:

102 continue

103 logger.debug(

104 "{} {} {}".format(path, device["Address"], device["Alias"])

105 )

106 if ADV_SVC_UUID in device["UUIDs"] and path.startswith(self._path):

107 obj = self._bus.get_object("org.bluez", path)

108 logger.info(

109 "Device found: [ {} ] {} - {}".format(

110 path, device["Name"], device["Address"]

111 )

112 )

113 return dbus.Interface(obj, DEVICE_INTERFACE)

114 raise SystemExit("Bluetooth device not found!")

115

116 def remove_device(self, device):

117 """Remove the remote device object at the given path."""

118 try:

119 self._adapter.RemoveDevice(device)

120 except dbus.exceptions.DBusException as msg:

121 logging.error(msg)

122 raise SystemExit(1)

123 logger.info("Device properly removed")

124

125

126 class BtGATTRemoteService:

127 """Bluetooth LE GATT Remote Service class."""

128

129 def __init__(self, SVC_UUID, adapter, device, max_notif):

130 self.SVC_UUID = SVC_UUID

131 self._adapter = adapter

132 self.device = device

133 self._max_notif = max_notif

134 self._notifications = 0

135 self._bus = dbus.SystemBus()

136 self._manager = dbus.Interface(

137 self._bus.get_object("org.bluez", "/"), OM_INTERFACE

138 )

139 self._main_loop = GObject.MainLoop()

140 self._service = self._find_service()

141 self._path = self._service.object_path

142

143 def _get_managed_objects(self):

144 return self._manager.GetManagedObjects()

145

146 def _find_service(self):

147 for path, ifaces in self._get_managed_objects().items():

148 if GATT_SERVICE_INTERFACE not in ifaces.keys():

149 continue

150 service = self._bus.get_object("org.bluez", path)

151 props = dbus.Interface(service, PROP_INTERFACE)

152 if props.Get(GATT_SERVICE_INTERFACE, "UUID") == self.SVC_UUID:

153 logger.info("Service found: {}".format(path))

154 return service

155 self._adapter.remove_device(self._device)

156 raise SystemExit("Bluetooth Service not found!")

157

158 def find_chrc(self, MSRMT_UUID):

159 for path, ifaces in self._get_managed_objects().items():

160 if GATT_CHRC_INTERFACE not in ifaces.keys():

161 continue

162 chrc = self._bus.get_object("org.bluez", path)

163 props = dbus.Interface(chrc, PROP_INTERFACE)

164 if props.Get(GATT_CHRC_INTERFACE, "UUID") == MSRMT_UUID:

165 logger.info("Characteristic found: {}".format(path))

166 return chrc

167 self._adapter.remove_device(self._device)

168 raise SystemExit("Bluetooth Characteristic not found!")

169

170 def _generic_error_cb(self, error):

171 self._adapter.remove_device(self._device)

172 self._main_loop.quit()

173 raise SystemExit("D-Bus call failed: " + str(error))

174

175 def _start_notify_cb(self):

176 logger.info("Notifications enabled")

177

178 def _notify_timeout(self):

179 self._adapter.remove_device(self._device)

180 self._main_loop.quit()

181 raise SystemExit("Notification test failed")

182

183 def _changed_cb(self, iface, changed_props, invalidated_props):

184 if iface != GATT_CHRC_INTERFACE:

185 return

186 if not len(changed_props):

187 return

188 value = changed_props.get("Value", None)

189 if not value:

190 return

191 logger.debug("New Notification")

192 self._notifications += 1

193 if self._notifications >= self._max_notif:

194 logger.info("Notification test succeeded")

195 self._main_loop.quit()

196

197 def check_notification(self, chrc, timeout=20):

198 # Listen to PropertiesChanged signals from the BLE Measurement

199 # Characteristic.

200 prop_iface = dbus.Interface(chrc, PROP_INTERFACE)

201 prop_iface.connect_to_signal("PropertiesChanged", self._changed_cb)

202

203 # Subscribe to BLE Measurement notifications.

204 chrc.StartNotify(

205 reply_handler=self._start_notify_cb,

206 error_handler=self._generic_error_cb,

207 dbus_interface=GATT_CHRC_INTERFACE,

208 )

209 GObject.timeout_add_seconds(timeout, self._notify_timeout)

210 self._main_loop.run()

211

212

213 def main():

214 logger.setLevel(logging.DEBUG)

215 parser = argparse.ArgumentParser()

216 parser.add_argument(

217 "id", help="Address, udev path or name (hciX) of the BT adapter"

218 )

219 parser.add_argument(

220 "ADV_SVC_UUID", help="Beacon Gatt configuration service UUID"

221 )

222 parser.add_argument(

223 "SVC_UUID", help="Beacon Gatt notification service UUID"

224 )

225 parser.add_argument("MSRMT_UUID", help="Beacon Gatt measurement UUID")

226 parser.add_argument(

227 "--max-notif",

228 "-m",

229 type=int,

230 default=5,

231 help="Maximum notification threshold",

232 )

233 args = parser.parse_args()

234 adapter = BtAdapter(args.id)

235 adapter.ensure_powered()

236 adapter.scan()

237 device = adapter.find_device_with_service(args.ADV_SVC_UUID)

238 try:

239 device.Connect()

240 except dbus.exceptions.DBusException as msg:

241 logging.error(msg)

242 adapter.remove_device(device)

243 raise SystemExit(1)

244 logger.info("Device connected, waiting 10s for services to be available")

245 time.sleep(10) # Let all the services to broadcast their UUIDs

246 service = BtGATTRemoteService(

247 args.SVC_UUID, adapter, device, args.max_notif

248 )

249 chrc = service.find_chrc(args.MSRMT_UUID)

250 service.check_notification(chrc)

251 try:

252 device.Disconnect()

253 except dbus.exceptions.DBusException as msg:

254 logging.error(msg)

255 adapter.remove_device(device)

256 raise SystemExit(1)

257 logger.info("Device properly disconnected")

258 adapter.remove_device(device)

259

260

261 if __name__ == "__main__":

262 sys.exit(main())