mirror of
https://github.com/espressif/esp-idf.git
synced 2024-10-05 20:47:46 -04:00
297 lines
12 KiB
Python
297 lines
12 KiB
Python
# Copyright 2018 Espressif Systems (Shanghai) PTE LTD
|
|
#
|
|
# Licensed under the Apache License, Version 2.0 (the "License");
|
|
# you may not use this file except in compliance with the License.
|
|
# You may obtain a copy of the License at
|
|
#
|
|
# http://www.apache.org/licenses/LICENSE-2.0
|
|
#
|
|
# Unless required by applicable law or agreed to in writing, software
|
|
# distributed under the License is distributed on an "AS IS" BASIS,
|
|
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
# See the License for the specific language governing permissions and
|
|
# limitations under the License.
|
|
#
|
|
|
|
from __future__ import print_function
|
|
|
|
import platform
|
|
from builtins import input
|
|
|
|
import utils
|
|
from future.utils import iteritems
|
|
|
|
fallback = True
|
|
|
|
|
|
# Check if platform is Linux and required packages are installed
|
|
# else fallback to console mode
|
|
if platform.system() == 'Linux':
|
|
try:
|
|
import time
|
|
|
|
import dbus
|
|
import dbus.mainloop.glib
|
|
fallback = False
|
|
except ImportError:
|
|
pass
|
|
|
|
|
|
# --------------------------------------------------------------------
|
|
|
|
|
|
# BLE client (Linux Only) using Bluez and DBus
|
|
class BLE_Bluez_Client:
|
|
def connect(self, devname, iface, chrc_names, fallback_srv_uuid):
|
|
self.devname = devname
|
|
self.srv_uuid_fallback = fallback_srv_uuid
|
|
self.chrc_names = [name.lower() for name in chrc_names]
|
|
self.device = None
|
|
self.adapter = None
|
|
self.adapter_props = None
|
|
self.services = None
|
|
self.nu_lookup = None
|
|
self.characteristics = dict()
|
|
self.srv_uuid_adv = None
|
|
|
|
dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)
|
|
bus = dbus.SystemBus()
|
|
manager = dbus.Interface(bus.get_object('org.bluez', '/'), 'org.freedesktop.DBus.ObjectManager')
|
|
objects = manager.GetManagedObjects()
|
|
|
|
for path, interfaces in iteritems(objects):
|
|
adapter = interfaces.get('org.bluez.Adapter1')
|
|
if adapter is not None:
|
|
if path.endswith(iface):
|
|
self.adapter = dbus.Interface(bus.get_object('org.bluez', path), 'org.bluez.Adapter1')
|
|
self.adapter_props = dbus.Interface(bus.get_object('org.bluez', path), 'org.freedesktop.DBus.Properties')
|
|
break
|
|
|
|
if self.adapter is None:
|
|
raise RuntimeError('Bluetooth adapter not found')
|
|
|
|
self.adapter_props.Set('org.bluez.Adapter1', 'Powered', dbus.Boolean(1))
|
|
self.adapter.StartDiscovery()
|
|
|
|
retry = 10
|
|
while (retry > 0):
|
|
try:
|
|
if self.device is None:
|
|
print('Connecting...')
|
|
# Wait for device to be discovered
|
|
time.sleep(5)
|
|
self._connect_()
|
|
print('Connected')
|
|
print('Getting Services...')
|
|
# Wait for services to be discovered
|
|
time.sleep(5)
|
|
self._get_services_()
|
|
return True
|
|
except Exception as e:
|
|
print(e)
|
|
retry -= 1
|
|
print('Retries left', retry)
|
|
continue
|
|
self.adapter.StopDiscovery()
|
|
return False
|
|
|
|
def _connect_(self):
|
|
bus = dbus.SystemBus()
|
|
manager = dbus.Interface(bus.get_object('org.bluez', '/'), 'org.freedesktop.DBus.ObjectManager')
|
|
objects = manager.GetManagedObjects()
|
|
dev_path = None
|
|
for path, interfaces in iteritems(objects):
|
|
if 'org.bluez.Device1' not in interfaces:
|
|
continue
|
|
if interfaces['org.bluez.Device1'].get('Name') == self.devname:
|
|
dev_path = path
|
|
break
|
|
|
|
if dev_path is None:
|
|
raise RuntimeError('BLE device not found')
|
|
|
|
try:
|
|
self.device = bus.get_object('org.bluez', dev_path)
|
|
try:
|
|
uuids = self.device.Get('org.bluez.Device1', 'UUIDs',
|
|
dbus_interface='org.freedesktop.DBus.Properties')
|
|
# There should be 1 service UUID in advertising data
|
|
# If bluez had cached an old version of the advertisement data
|
|
# the list of uuids may be incorrect, in which case connection
|
|
# or service discovery may fail the first time. If that happens
|
|
# the cache will be refreshed before next retry
|
|
if len(uuids) == 1:
|
|
self.srv_uuid_adv = uuids[0]
|
|
except dbus.exceptions.DBusException as e:
|
|
print(e)
|
|
|
|
self.device.Connect(dbus_interface='org.bluez.Device1')
|
|
except Exception as e:
|
|
print(e)
|
|
self.device = None
|
|
raise RuntimeError('BLE device could not connect')
|
|
|
|
def _get_services_(self):
|
|
bus = dbus.SystemBus()
|
|
manager = dbus.Interface(bus.get_object('org.bluez', '/'), 'org.freedesktop.DBus.ObjectManager')
|
|
objects = manager.GetManagedObjects()
|
|
service_found = False
|
|
for srv_path, srv_interfaces in iteritems(objects):
|
|
if 'org.bluez.GattService1' not in srv_interfaces:
|
|
continue
|
|
if not srv_path.startswith(self.device.object_path):
|
|
continue
|
|
service = bus.get_object('org.bluez', srv_path)
|
|
srv_uuid = service.Get('org.bluez.GattService1', 'UUID',
|
|
dbus_interface='org.freedesktop.DBus.Properties')
|
|
|
|
# If service UUID doesn't match the one found in advertisement data
|
|
# then also check if it matches the fallback UUID
|
|
if srv_uuid not in [self.srv_uuid_adv, self.srv_uuid_fallback]:
|
|
continue
|
|
|
|
nu_lookup = dict()
|
|
characteristics = dict()
|
|
for chrc_path, chrc_interfaces in iteritems(objects):
|
|
if 'org.bluez.GattCharacteristic1' not in chrc_interfaces:
|
|
continue
|
|
if not chrc_path.startswith(service.object_path):
|
|
continue
|
|
chrc = bus.get_object('org.bluez', chrc_path)
|
|
uuid = chrc.Get('org.bluez.GattCharacteristic1', 'UUID',
|
|
dbus_interface='org.freedesktop.DBus.Properties')
|
|
characteristics[uuid] = chrc
|
|
for desc_path, desc_interfaces in iteritems(objects):
|
|
if 'org.bluez.GattDescriptor1' not in desc_interfaces:
|
|
continue
|
|
if not desc_path.startswith(chrc.object_path):
|
|
continue
|
|
desc = bus.get_object('org.bluez', desc_path)
|
|
desc_uuid = desc.Get('org.bluez.GattDescriptor1', 'UUID',
|
|
dbus_interface='org.freedesktop.DBus.Properties')
|
|
if desc_uuid[4:8] != '2901':
|
|
continue
|
|
try:
|
|
readval = desc.ReadValue({}, dbus_interface='org.bluez.GattDescriptor1')
|
|
except dbus.exceptions.DBusException:
|
|
break
|
|
found_name = ''.join(chr(b) for b in readval).lower()
|
|
nu_lookup[found_name] = uuid
|
|
break
|
|
|
|
match_found = True
|
|
for name in self.chrc_names:
|
|
if name not in nu_lookup:
|
|
# Endpoint name not present
|
|
match_found = False
|
|
break
|
|
|
|
# Create lookup table only if all endpoint names found
|
|
self.nu_lookup = [None, nu_lookup][match_found]
|
|
self.characteristics = characteristics
|
|
service_found = True
|
|
|
|
# If the service UUID matches that in the advertisement
|
|
# we can stop the search now. If it doesn't match, we
|
|
# have found the service corresponding to the fallback
|
|
# UUID, in which case don't break and keep searching
|
|
# for the advertised service
|
|
if srv_uuid == self.srv_uuid_adv:
|
|
break
|
|
|
|
if not service_found:
|
|
self.device.Disconnect(dbus_interface='org.bluez.Device1')
|
|
if self.adapter:
|
|
self.adapter.RemoveDevice(self.device)
|
|
self.device = None
|
|
self.nu_lookup = None
|
|
self.characteristics = dict()
|
|
raise RuntimeError('Provisioning service not found')
|
|
|
|
def get_nu_lookup(self):
|
|
return self.nu_lookup
|
|
|
|
def has_characteristic(self, uuid):
|
|
if uuid in self.characteristics:
|
|
return True
|
|
return False
|
|
|
|
def disconnect(self):
|
|
if self.device:
|
|
self.device.Disconnect(dbus_interface='org.bluez.Device1')
|
|
if self.adapter:
|
|
self.adapter.RemoveDevice(self.device)
|
|
self.device = None
|
|
self.nu_lookup = None
|
|
self.characteristics = dict()
|
|
if self.adapter_props:
|
|
self.adapter_props.Set('org.bluez.Adapter1', 'Powered', dbus.Boolean(0))
|
|
|
|
def send_data(self, characteristic_uuid, data):
|
|
try:
|
|
path = self.characteristics[characteristic_uuid]
|
|
except KeyError:
|
|
raise RuntimeError('Invalid characteristic : ' + characteristic_uuid)
|
|
|
|
try:
|
|
path.WriteValue([ord(c) for c in data], {}, dbus_interface='org.bluez.GattCharacteristic1')
|
|
except TypeError: # python3 compatible
|
|
path.WriteValue([c for c in data], {}, dbus_interface='org.bluez.GattCharacteristic1')
|
|
except dbus.exceptions.DBusException as e:
|
|
raise RuntimeError('Failed to write value to characteristic ' + characteristic_uuid + ': ' + str(e))
|
|
|
|
try:
|
|
readval = path.ReadValue({}, dbus_interface='org.bluez.GattCharacteristic1')
|
|
except dbus.exceptions.DBusException as e:
|
|
raise RuntimeError('Failed to read value from characteristic ' + characteristic_uuid + ': ' + str(e))
|
|
return ''.join(chr(b) for b in readval)
|
|
|
|
|
|
# --------------------------------------------------------------------
|
|
|
|
|
|
# Console based BLE client for Cross Platform support
|
|
class BLE_Console_Client:
|
|
def connect(self, devname, iface, chrc_names, fallback_srv_uuid):
|
|
print('BLE client is running in console mode')
|
|
print('\tThis could be due to your platform not being supported or dependencies not being met')
|
|
print('\tPlease ensure all pre-requisites are met to run the full fledged client')
|
|
print('BLECLI >> Please connect to BLE device `' + devname + '` manually using your tool of choice')
|
|
resp = input('BLECLI >> Was the device connected successfully? [y/n] ')
|
|
if resp != 'Y' and resp != 'y':
|
|
return False
|
|
print('BLECLI >> List available attributes of the connected device')
|
|
resp = input("BLECLI >> Is the service UUID '" + fallback_srv_uuid + "' listed among available attributes? [y/n] ")
|
|
if resp != 'Y' and resp != 'y':
|
|
return False
|
|
return True
|
|
|
|
def get_nu_lookup(self):
|
|
return None
|
|
|
|
def has_characteristic(self, uuid):
|
|
resp = input("BLECLI >> Is the characteristic UUID '" + uuid + "' listed among available attributes? [y/n] ")
|
|
if resp != 'Y' and resp != 'y':
|
|
return False
|
|
return True
|
|
|
|
def disconnect(self):
|
|
pass
|
|
|
|
def send_data(self, characteristic_uuid, data):
|
|
print("BLECLI >> Write following data to characteristic with UUID '" + characteristic_uuid + "' :")
|
|
print('\t>> ' + utils.str_to_hexstr(data))
|
|
print('BLECLI >> Enter data read from characteristic (in hex) :')
|
|
resp = input('\t<< ')
|
|
return utils.hexstr_to_str(resp)
|
|
|
|
|
|
# --------------------------------------------------------------------
|
|
|
|
|
|
# Function to get client instance depending upon platform
|
|
def get_client():
|
|
if fallback:
|
|
return BLE_Console_Client()
|
|
return BLE_Bluez_Client()
|