diff --git a/scripts/components.py b/scripts/components.py index 9caf9ab0c..73f438995 100644 --- a/scripts/components.py +++ b/scripts/components.py @@ -9,6 +9,7 @@ 'iotile_transport_bled112': ['iotile-transport-bled112', 'transport_plugins/bled112', True], 'iotile_transport_awsiot': ['iotile-transport-awsiot', 'transport_plugins/awsiot', False], 'iotile_transport_websocket': ['iotile-transport-websocket', 'transport_plugins/websocket', True], + 'iotile_transport_native_ble': ['iotile-transport-native-ble', 'transport_plugins/native_ble', True], 'iotile_transport_jlink': ['iotile-transport-jlink', 'transport_plugins/jlink', False], 'iotile_ext_cloud': ['iotile-ext-cloud', 'iotile_ext_cloud', True] } diff --git a/scripts/test.py b/scripts/test.py index faefd7c6e..e5b34e2e2 100644 --- a/scripts/test.py +++ b/scripts/test.py @@ -64,6 +64,8 @@ def do_test_all(self, subcmd, opts, *args): if status is None: print("SKIPPED ON PYTHON 3") + elif status == 5: + print("NO TESTS RAN (%.1f seconds)" % duration) elif status != 0: failed = True failed_outputs.append(output) @@ -82,6 +84,7 @@ def do_test_all(self, subcmd, opts, *args): return int(failed) + if __name__ == '__main__': proc = TestProcessor() sys.exit(proc.main()) diff --git a/tox.ini b/tox.ini index 158d5fb8d..eb2b6cd72 100644 --- a/tox.ini +++ b/tox.ini @@ -1,36 +1,14 @@ [tox] -envlist = py27, py36 +envlist = py{27,36}-{mac_windows,linux_only} skipsdist = True -[testenv:py27] -passenv=APPDATA TRAVIS -deps= - six - cmdln - pytest - pytest-logging - pytest-localserver - ./iotilecore - ./iotilebuild - ./iotiletest - ./iotilegateway - ./iotilesensorgraph - ./iotileship - ./transport_plugins/bled112 - ./transport_plugins/awsiot - ./transport_plugins/jlink - ./transport_plugins/websocket - ./iotile_ext_cloud - requests-mock - tornado>=4.4.0,<5.0.0 - futures - pycryptodome -commands= - python scripts/test.py test_all +[testenv] +platform = mac_windows: darwin|win32 + linux_only: linux|linux2 -[testenv:py36] -passenv=APPDATA TRAVIS -deps= +passenv = APPDATA TRAVIS + +deps = six cmdln pytest @@ -46,10 +24,11 @@ deps= ./transport_plugins/awsiot ./transport_plugins/jlink ./transport_plugins/websocket + linux_only: ./transport_plugins/native_ble ./iotile_ext_cloud requests-mock tornado>=4.4.0,<5.0.0 + py27: futures pycryptodome -commands= +commands = python scripts/test.py test_all - diff --git a/transport_plugins/native_ble/LICENSE.md b/transport_plugins/native_ble/LICENSE.md new file mode 100644 index 000000000..65c5ca88a --- /dev/null +++ b/transport_plugins/native_ble/LICENSE.md @@ -0,0 +1,165 @@ + GNU LESSER GENERAL PUBLIC LICENSE + Version 3, 29 June 2007 + + Copyright (C) 2007 Free Software Foundation, Inc. + Everyone is permitted to copy and distribute verbatim copies + of this license document, but changing it is not allowed. + + + This version of the GNU Lesser General Public License incorporates +the terms and conditions of version 3 of the GNU General Public +License, supplemented by the additional permissions listed below. + + 0. Additional Definitions. + + As used herein, "this License" refers to version 3 of the GNU Lesser +General Public License, and the "GNU GPL" refers to version 3 of the GNU +General Public License. + + "The Library" refers to a covered work governed by this License, +other than an Application or a Combined Work as defined below. + + An "Application" is any work that makes use of an interface provided +by the Library, but which is not otherwise based on the Library. +Defining a subclass of a class defined by the Library is deemed a mode +of using an interface provided by the Library. + + A "Combined Work" is a work produced by combining or linking an +Application with the Library. The particular version of the Library +with which the Combined Work was made is also called the "Linked +Version". + + The "Minimal Corresponding Source" for a Combined Work means the +Corresponding Source for the Combined Work, excluding any source code +for portions of the Combined Work that, considered in isolation, are +based on the Application, and not on the Linked Version. + + The "Corresponding Application Code" for a Combined Work means the +object code and/or source code for the Application, including any data +and utility programs needed for reproducing the Combined Work from the +Application, but excluding the System Libraries of the Combined Work. + + 1. Exception to Section 3 of the GNU GPL. + + You may convey a covered work under sections 3 and 4 of this License +without being bound by section 3 of the GNU GPL. + + 2. Conveying Modified Versions. + + If you modify a copy of the Library, and, in your modifications, a +facility refers to a function or data to be supplied by an Application +that uses the facility (other than as an argument passed when the +facility is invoked), then you may convey a copy of the modified +version: + + a) under this License, provided that you make a good faith effort to + ensure that, in the event an Application does not supply the + function or data, the facility still operates, and performs + whatever part of its purpose remains meaningful, or + + b) under the GNU GPL, with none of the additional permissions of + this License applicable to that copy. + + 3. Object Code Incorporating Material from Library Header Files. + + The object code form of an Application may incorporate material from +a header file that is part of the Library. You may convey such object +code under terms of your choice, provided that, if the incorporated +material is not limited to numerical parameters, data structure +layouts and accessors, or small macros, inline functions and templates +(ten or fewer lines in length), you do both of the following: + + a) Give prominent notice with each copy of the object code that the + Library is used in it and that the Library and its use are + covered by this License. + + b) Accompany the object code with a copy of the GNU GPL and this license + document. + + 4. Combined Works. + + You may convey a Combined Work under terms of your choice that, +taken together, effectively do not restrict modification of the +portions of the Library contained in the Combined Work and reverse +engineering for debugging such modifications, if you also do each of +the following: + + a) Give prominent notice with each copy of the Combined Work that + the Library is used in it and that the Library and its use are + covered by this License. + + b) Accompany the Combined Work with a copy of the GNU GPL and this license + document. + + c) For a Combined Work that displays copyright notices during + execution, include the copyright notice for the Library among + these notices, as well as a reference directing the user to the + copies of the GNU GPL and this license document. + + d) Do one of the following: + + 0) Convey the Minimal Corresponding Source under the terms of this + License, and the Corresponding Application Code in a form + suitable for, and under terms that permit, the user to + recombine or relink the Application with a modified version of + the Linked Version to produce a modified Combined Work, in the + manner specified by section 6 of the GNU GPL for conveying + Corresponding Source. + + 1) Use a suitable shared library mechanism for linking with the + Library. A suitable mechanism is one that (a) uses at run time + a copy of the Library already present on the user's computer + system, and (b) will operate properly with a modified version + of the Library that is interface-compatible with the Linked + Version. + + e) Provide Installation Information, but only if you would otherwise + be required to provide such information under section 6 of the + GNU GPL, and only to the extent that such information is + necessary to install and execute a modified version of the + Combined Work produced by recombining or relinking the + Application with a modified version of the Linked Version. (If + you use option 4d0, the Installation Information must accompany + the Minimal Corresponding Source and Corresponding Application + Code. If you use option 4d1, you must provide the Installation + Information in the manner specified by section 6 of the GNU GPL + for conveying Corresponding Source.) + + 5. Combined Libraries. + + You may place library facilities that are a work based on the +Library side by side in a single library together with other library +facilities that are not Applications and are not covered by this +License, and convey such a combined library under terms of your +choice, if you do both of the following: + + a) Accompany the combined library with a copy of the same work based + on the Library, uncombined with any other library facilities, + conveyed under the terms of this License. + + b) Give prominent notice with the combined library that part of it + is a work based on the Library, and explaining where to find the + accompanying uncombined form of the same work. + + 6. Revised Versions of the GNU Lesser General Public License. + + The Free Software Foundation may publish revised and/or new versions +of the GNU Lesser General Public License from time to time. Such new +versions will be similar in spirit to the present version, but may +differ in detail to address new problems or concerns. + + Each version is given a distinguishing version number. If the +Library as you received it specifies that a certain numbered version +of the GNU Lesser General Public License "or any later version" +applies to it, you have the option of following the terms and +conditions either of that published version or of any later version +published by the Free Software Foundation. If the Library as you +received it does not specify a version number of the GNU Lesser +General Public License, you may choose any version of the GNU Lesser +General Public License ever published by the Free Software Foundation. + + If the Library as you received it specifies that a proxy can decide +whether future versions of the GNU Lesser General Public License shall +apply, that proxy's public statement of acceptance of any version is +permanent authorization for you to choose that version for the +Library. diff --git a/transport_plugins/native_ble/MANIFEST.in b/transport_plugins/native_ble/MANIFEST.in new file mode 100644 index 000000000..738bd51b1 --- /dev/null +++ b/transport_plugins/native_ble/MANIFEST.in @@ -0,0 +1 @@ +include version.py \ No newline at end of file diff --git a/transport_plugins/native_ble/README.md b/transport_plugins/native_ble/README.md new file mode 100644 index 000000000..a41b46ac9 --- /dev/null +++ b/transport_plugins/native_ble/README.md @@ -0,0 +1,17 @@ +## IOTile Transport Native BLE + +The IOTile Transport Native BLE plugin allows to connect to use any Bluetooth Low Energy controller +to interact with IOTile devices. +It contains a NativeBLEDeviceAdapter, a NativeBLEVirtualInterface and some tools needed +to make the whole thing work. + +To have a cross-platform way to interact with Bluetooth controller, we use [baBLE](https://github.com/iotile/baBLE), +allowing use to send and receive HCI packets, without worrying about the OS (currently only working on Linux). + +### Linux requirements + +To use **baBLE** without using `sudo`, we need to set its capabilities. To do so, simply run this command after the +transport plugin installation: +```bash +$ bable --set-cap +``` diff --git a/transport_plugins/native_ble/RELEASE.md b/transport_plugins/native_ble/RELEASE.md new file mode 100644 index 000000000..5d06ac4fd --- /dev/null +++ b/transport_plugins/native_ble/RELEASE.md @@ -0,0 +1,7 @@ +# Release Notes + +All major changes in each released version of the native BLE transport plugin are listed here. + +## 1.0.0 + +- Initial public release (only works on Linux) \ No newline at end of file diff --git a/transport_plugins/native_ble/iotile_transport_native_ble/__init__.py b/transport_plugins/native_ble/iotile_transport_native_ble/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/transport_plugins/native_ble/iotile_transport_native_ble/config_variables.py b/transport_plugins/native_ble/iotile_transport_native_ble/config_variables.py new file mode 100644 index 000000000..6322e16f6 --- /dev/null +++ b/transport_plugins/native_ble/iotile_transport_native_ble/config_variables.py @@ -0,0 +1,13 @@ +"""Canonical list of config variables defined by iotile-transport-native-ble.""" + +from __future__ import unicode_literals, absolute_import, print_function + + +def get_variables(): + prefix = "ble" + + conf_vars = [ + ["active-scan", "bool", "Probe devices during scan for uptime, voltage and broadcast data", "false"] + ] + + return prefix, conf_vars diff --git a/transport_plugins/native_ble/iotile_transport_native_ble/connection_manager.py b/transport_plugins/native_ble/iotile_transport_native_ble/connection_manager.py new file mode 100644 index 000000000..ac0ee70d7 --- /dev/null +++ b/transport_plugins/native_ble/iotile_transport_native_ble/connection_manager.py @@ -0,0 +1,706 @@ +import threading +import queue +import logging +from monotonic import monotonic +from past.builtins import basestring +from builtins import int +from future.utils import iteritems +from iotile.core.exceptions import ArgumentError + + +class ConnectionAction(object): + """A generic action handled internally by ConnectionManager + + Args: + action (string): The action to take + data (dict): Any associated data + sync (bool): Whether the caller is synchronously waiting + for the result. + timeout (float): The maximum amount of time that should occur + before timing this action out. + """ + + def __init__(self, action, data, timeout=5.0, sync=False): + self.action = action + self.data = data + self.sync = sync + self.timeout = timeout + self.start_time = monotonic() + + if self.sync: + self.done = threading.Event() + else: + self.done = None + + def set_timeout(self, timeout): + self.timeout = timeout + self.start_time = monotonic() + + @property + def expired(self): + """Boolean property if this action has expired + """ + if self.timeout is None: + return False + + return monotonic() - self.start_time > self.timeout + + +class ConnectionManager(threading.Thread): + """A class that manages connection states and transitions. + + ConnectionManager presents a nonblocking interface that is designed + to work with DeviceAdapter. It handles maintaining an internal dictionary + of currently active connections and a worker thread that processes + requested changes to those connections. All work is synchronized + through requests to the worker thread. + + A connection can be in one of 4 macrostates: + disconnected: There is no connection + connecting: The connection has been started but has not yet entered + a fully connected state + idle: The connection is connected and idle + in_progress: An operation is in progress on the connection + disconnecting: The connection has started the disconnect process + but has not finished it yet. + + The user of ConnectionManager is free to create their own microstates if + the actions required to, e.g., connect to a device require a sequence + of actions. + + Each connection has a user managed context associated with it that can be used + to track data about the connection and an internal identifier that can be used + to retrieve a connection's user context along with an integer connection_id. + + ConnectionManager just enforces that state transitions only happen in + the following ways: + nonexistent -> connecting -> idle <--> in_progress <--> idle -> disconnecting -> nonexistant + + ConnectionManager will fail a request that does not follow the above pattern. + """ + + Disconnected = 0 + Connecting = 1 + Idle = 2 + InProgress = 3 + Disconnecting = 4 + + def __init__(self, adapter_id): + """Constructor. + + Args: + adapter_id (int): Since the ConnectionManager responds to callbacks on behalf + of a DeviceAdapter, it needs to know what adapter_id to send with the + callbacks. + """ + + super(ConnectionManager, self).__init__() + + self.id = adapter_id + self._stop_event = threading.Event() + self._actions = queue.Queue() + self._connections = {} + self._int_connections = {} + self._data_lock = threading.Lock() + + # Our thread should be a daemon so that we don't block exiting the program if we hang + self.daemon = True + + self._logger = logging.getLogger(__name__) + self._logger.addHandler(logging.NullHandler()) + self._logger.setLevel(logging.INFO) + + def run(self): + while True: + try: + if self._stop_event.is_set(): + break + + # Check if we should time anything out + self._check_timeouts() + + try: + action = self._actions.get(timeout=0.1) + except queue.Empty: + continue + + handler_name = '_{}_action'.format(action.action) + + if not hasattr(self, handler_name): + self._logger.error("Ignoring unknown action in ConnectionManager: %s", action.action) + continue + + handler = getattr(self, handler_name) + handler(action) + + if action.sync: + action.done.set() + except Exception: + self._logger.exception('Exception processing event in ConnectionManager') + + def stop(self): + try: + self._stop_event.set() + self.join(5.0) + except RuntimeError: + self._logger.warn("Could not stop connection manager thread, killing it on exit in a dirty fashion") + + def get_connections(self): + """Get a list of all open connections + + Note that these connections can close at any time, so this cannot + be relied upon to be valid at any point after this function returns + + Returns: + int[]: A list of integer connection ids + """ + + return self._connections.keys() + + def get_context(self, conn_or_internal_id): + """Get the context for a connection by either connection_id or internal_id + + Args: + conn_or_internal_id (int, string): The external integer connection id or + an internal string connection id + + Returns: + dict: The context data associated with that connection or None if it cannot + be found. + + Raises: + ArgumentError: When the key is not found in the list of active connections + or is invalid. + """ + + key = conn_or_internal_id + if isinstance(key, basestring): + table = self._int_connections + elif isinstance(key, int): + table = self._connections + else: + raise ArgumentError( + "You must supply either an int connection id or a string internal id to _get_connection_state", + id=key + ) + + try: + data = table[key] + except KeyError: + raise ArgumentError("Could not find connection by id", id=key) + + return data['context'] + + def get_connection_id(self, conn_or_internal_id): + """Get the connection id. + + Args: + conn_or_internal_id (int, string): The external integer connection id or + an internal string connection id + + Returns: + int: The connection id associated with that connection + + Raises: + ArgumentError: When the key is not found in the list of active connections + or is invalid. + """ + + key = conn_or_internal_id + if isinstance(key, basestring): + table = self._int_connections + elif isinstance(key, int): + table = self._connections + else: + raise ArgumentError( + "You must supply either an int connection id or a string internal id to _get_connection_state", + id=key + ) + + try: + data = table[key] + except KeyError: + raise ArgumentError("Could not find connection by id", id=key) + + return data['connection_id'] + + def _get_connection(self, conn_or_internal_id): + """Get the data for a connection by either connection_id or internal_id + + Args: + conn_or_internal_id (int, string): The external integer connection id or + and internal string connection id + + Returns: + dict: The context data associated with that connection or None if it cannot + be found. + """ + + key = conn_or_internal_id + if isinstance(key, basestring): + table = self._int_connections + elif isinstance(key, int): + table = self._connections + else: + return None + + try: + data = table[key] + except KeyError: + return None + + return data + + def _get_connection_state(self, conn_or_internal_id): + """Get a connection's state by either connection_id or internal_id + + This routine must only be called from the internal worker thread. + + Args: + conn_or_internal_id (int, string): The external integer connection id or + and internal string connection id + """ + + key = conn_or_internal_id + if isinstance(key, basestring): + table = self._int_connections + elif isinstance(key, int): + table = self._connections + else: + raise ArgumentError( + "You must supply either an int connection id or a string internal id to _get_connection_state", + id=key + ) + + if key not in table: + return self.Disconnected + + data = table[key] + return data['state'] + + def get_state(self, conn_or_internal_id): + state = self._get_connection_state(conn_or_internal_id) + + if state == self.Disconnected: + return "Disconnected" + if state == self.Connecting: + return "Connecting" + if state == self.Idle: + return "Idle" + if state == self.InProgress: + return "InProgress" + elif state == self.Disconnecting: + return "Disconnecting" + else: + return "Unknown state" + + def _check_timeouts(self): + """Check if any operations in progress need to be timed out + + Adds the corresponding finish action that fails the request due to a timeout. + """ + + for connection_id, data in iteritems(self._connections): + if 'action' in data and data['action'].expired: + if data['state'] == self.Connecting: + self.finish_connection(connection_id, False, 'Connection attempt timed out') + elif data['state'] == self.Disconnecting: + self.finish_disconnection(connection_id, False, 'Disconnection attempt timed out') + elif data['state'] == self.InProgress: + if data['microstate'] == 'rpc': + self.finish_operation(connection_id, False, 'RPC timed out without response', None, None) + elif data['microstate'] == 'open_interface': + self.finish_operation(connection_id, False, 'Open interface request timed out') + + def add_connection(self, connection_id, internal_id, context): + """Add an already created connection. Used to register devices connected before starting the device adapter. + + Args: + connection_id (int): The external connection id + internal_id (string): An internal identifier for the connection + context (dict): Additional information to associate with this context + """ + # Make sure we are not reusing an id that is currently connected to something + if self._get_connection_state(connection_id) != self.Disconnected: + return + if self._get_connection_state(internal_id) != self.Disconnected: + return + + conn_data = { + 'state': self.Idle, + 'microstate': None, + 'connection_id': connection_id, + 'internal_id': internal_id, + 'context': context + } + + self._connections[connection_id] = conn_data + self._int_connections[internal_id] = conn_data + + def begin_connection(self, connection_id, internal_id, callback, context, timeout): + """Asynchronously begin a connection attempt + + Args: + connection_id (int): The external connection id + internal_id (string): An internal identifier for the connection + callback (callable): The function to be called when the connection + attempt finishes + context (dict): Additional information to associate with this context + timeout (float): How long to allow this connection attempt to proceed + without timing it out + """ + + data = { + 'callback': callback, + 'connection_id': connection_id, + 'internal_id': internal_id, + 'context': context + } + + action = ConnectionAction('begin_connection', data, timeout=timeout, sync=False) + self._actions.put(action) + + def finish_connection(self, conn_or_internal_id, successful, failure_reason=None): + """Finish a connection attempt + + Args: + conn_or_internal_id (string, int): Either an integer connection id or a string + internal_id + successful (bool): Whether this connection attempt was successful + failure_reason (string): If this connection attempt failed, an optional reason + for the failure. + """ + + data = { + 'id': conn_or_internal_id, + 'success': successful, + 'failure_reason': failure_reason + } + + action = ConnectionAction('finish_connection', data, sync=False) + self._actions.put(action) + + def _begin_connection_action(self, action): + """Begin a connection attempt + + Args: + action (ConnectionAction): the action object describing what we are + connecting to + """ + + connection_id = action.data['connection_id'] + internal_id = action.data['internal_id'] + callback = action.data['callback'] + + # Make sure we are not reusing an id that is currently connected to something + if self._get_connection_state(connection_id) != self.Disconnected: + callback(connection_id, self.id, False, 'Connection ID is already in use for another connection') + return + + if self._get_connection_state(internal_id) != self.Disconnected: + callback(connection_id, self.id, False, 'Internal ID is already in use for another connection') + return + + conn_data = { + 'state': self.Connecting, + 'microstate': None, + 'connection_id': connection_id, + 'internal_id': internal_id, + 'action': action, + 'context': action.data['context'] + } + + self._connections[connection_id] = conn_data + self._int_connections[internal_id] = conn_data + + def _finish_connection_action(self, action): + """Finish a connection attempt + + Args: + action (ConnectionAction): the action object describing what we are + connecting to and what the result of the operation was + """ + + success = action.data['success'] + conn_key = action.data['id'] + + if self._get_connection_state(conn_key) != self.Connecting: + self._logger.error( + "Invalid finish_connection action on a connection whose state is not Connecting, conn_key={}" + .format(str(conn_key)) + ) + return + + # Cannot be None since we checked above to make sure it exists + data = self._get_connection(conn_key) + connection_id = data['connection_id'] + internal_id = data['internal_id'] + + last_action = data['action'] + callback = last_action.data['callback'] + + if success is False: + failure_reason = action.data['failure_reason'] + if failure_reason is None: + failure_reason = "No reason was given" + + del self._connections[connection_id] + del self._int_connections[internal_id] + callback(connection_id, self.id, False, failure_reason) + else: + data['state'] = self.Idle + data['microstate'] = None + del data['action'] + callback(connection_id, self.id, True, None) + + def unexpected_disconnect(self, conn_or_internal_id): + """Notify that there was an unexpected disconnection of the device. + + Any in progress operations are canceled cleanly and the device is transitioned + to a disconnected state. + + Args: + conn_or_internal_id (string, int): Either an integer connection id or a string + internal_id + """ + + data = { + 'id': conn_or_internal_id + } + + action = ConnectionAction('force_disconnect', data, sync=False) + self._actions.put(action) + + def begin_disconnection(self, conn_or_internal_id, callback, timeout): + """Begin a disconnection attempt + + Args: + conn_or_internal_id (string, int): Either an integer connection id or a string + internal_id + callback (callable): Callback to call when this disconnection attempt either + succeeds or fails + timeout (float): How long to allow this connection attempt to proceed + without timing it out (in seconds) + """ + + data = { + 'id': conn_or_internal_id, + 'callback': callback + } + + action = ConnectionAction('begin_disconnection', data, timeout=timeout, sync=False) + self._actions.put(action) + + def _force_disconnect_action(self, action): + """Forcibly disconnect a device. + + Args: + action (ConnectionAction): the action object describing what we are + forcibly disconnecting + """ + + conn_key = action.data['id'] + if self._get_connection_state(conn_key) == self.Disconnected: + return + + data = self._get_connection(conn_key) + + # If there are any operations in progress, cancel them cleanly + if data['state'] == self.Connecting: + callback = data['action'].data['callback'] + callback(data['connection_id'], self.id, False, 'Unexpected disconnection') + elif data['state'] == self.Disconnecting: + callback = data['action'].data['callback'] + callback(data['connection_id'], self.id, True, None) + elif data['state'] == self.InProgress: + callback = data['action'].data['callback'] + if data['microstate'] == 'rpc': + callback(False, 'Unexpected disconnection', 0xFF, None) + elif data['microstate'] == 'open_interface': + callback(False, 'Unexpected disconnection') + elif data['microstate'] == 'close_interface': + callback(False, 'Unexpected disconnection') + + connection_id = data['connection_id'] + internal_id = data['internal_id'] + del self._connections[connection_id] + del self._int_connections[internal_id] + + def _begin_disconnection_action(self, action): + """Begin a disconnection attempt + + Args: + action (ConnectionAction): the action object describing what we are + connecting to and what the result of the operation was + """ + + conn_key = action.data['id'] + callback = action.data['callback'] + + if self._get_connection_state(conn_key) != self.Idle: + callback(conn_key, self.id, False, 'Cannot start disconnection, connection is not idle') + return + + # Cannot be None since we checked above to make sure it exists + data = self._get_connection(conn_key) + data['state'] = self.Disconnecting + data['microstate'] = None + data['action'] = action + + def finish_disconnection(self, conn_or_internal_id, successful, failure_reason): + """Finish a disconnection attempt + + Args: + conn_or_internal_id (string, int): Either an integer connection id or a string + internal_id + successful (bool): Whether this connection attempt was successful + failure_reason (string): If this connection attempt failed, an optional reason + for the failure. + """ + + data = { + 'id': conn_or_internal_id, + 'success': successful, + 'failure_reason': failure_reason + } + + action = ConnectionAction('finish_disconnection', data, sync=False) + self._actions.put(action) + + def _finish_disconnection_action(self, action): + """Finish a disconnection attempt + + There are two possible outcomes: + - if we were successful at disconnecting, we transition to disconnected + - if we failed at disconnecting, we transition back to idle + + Args: + action (ConnectionAction): the action object describing what we are + disconnecting from and what the result of the operation was + """ + + success = action.data['success'] + conn_key = action.data['id'] + + if self._get_connection_state(conn_key) != self.Disconnecting: + self._logger.error( + "Invalid finish_disconnection action on a connection whose state is not Disconnecting, conn_key={}" + .format(str(conn_key)) + ) + return + + # Cannot be None since we checked above to make sure it exists + data = self._get_connection(conn_key) + connection_id = data['connection_id'] + internal_id = data['internal_id'] + + last_action = data['action'] + callback = last_action.data['callback'] + + if success is False: + failure_reason = action.data['failure_reason'] + if failure_reason is None: + failure_reason = "No reason was given" + + data['state'] = self.Idle + data['microstate'] = None + del data['action'] + callback(connection_id, self.id, False, failure_reason) + else: + del self._connections[connection_id] + del self._int_connections[internal_id] + callback(connection_id, self.id, True, None) + + def begin_operation(self, conn_or_internal_id, op_name, callback, timeout): + """Begin an operation on a connection + + Args: + conn_or_internal_id (string, int): Either an integer connection id or a string + internal_id + op_name (string): The name of the operation that we are starting (stored in + the connection's microstate) + callback (callable): Callback to call when this disconnection attempt either + succeeds or fails + timeout (float): How long to allow this connection attempt to proceed + without timing it out (in seconds) + """ + + data = { + 'id': conn_or_internal_id, + 'callback': callback, + 'operation_name': op_name + } + + action = ConnectionAction('begin_operation', data, timeout=timeout, sync=False) + self._actions.put(action) + + def _begin_operation_action(self, action): + """Begin an attempted operation. + + Args: + action (ConnectionAction): the action object describing what we are + operating on + """ + + conn_key = action.data['id'] + callback = action.data['callback'] + + if self._get_connection_state(conn_key) != self.Idle: + callback(conn_key, self.id, False, 'Cannot start operation, connection is not idle') + return + + data = self._get_connection(conn_key) + data['state'] = self.InProgress + data['microstate'] = action.data['operation_name'] + data['action'] = action + + def finish_operation(self, conn_or_internal_id, success, *args): + """Finish an operation on a connection. + + Args: + conn_or_internal_id (string, int): Either an integer connection id or a string + internal_id + success (bool): Whether the operation was successful + *args: Optional arguments for the callback + """ + + data = { + 'id': conn_or_internal_id, + 'success': success, + 'callback_args': args + } + + action = ConnectionAction('finish_operation', data, sync=False) + self._actions.put(action) + + def _finish_operation_action(self, action): + """Finish an attempted operation. + + Args: + action (ConnectionAction): the action object describing the result + of the operation that we are finishing + """ + + success = action.data['success'] + conn_key = action.data['id'] + + if self._get_connection_state(conn_key) != self.InProgress: + self._logger.error( + "Invalid finish_operation action on a connection whose state is not InProgress, conn_key={}" + .format(str(conn_key)) + ) + return + + # Cannot be None since we checked above to make sure it exists + data = self._get_connection(conn_key) + last_action = data['action'] + + callback = last_action.data['callback'] + connection_id = data['connection_id'] + args = action.data['callback_args'] + + data['state'] = self.Idle + data['microstate'] = None + del data['action'] + + callback(connection_id, self.id, success, *args) diff --git a/transport_plugins/native_ble/iotile_transport_native_ble/device_adapter.py b/transport_plugins/native_ble/iotile_transport_native_ble/device_adapter.py new file mode 100644 index 000000000..2d5885473 --- /dev/null +++ b/transport_plugins/native_ble/iotile_transport_native_ble/device_adapter.py @@ -0,0 +1,994 @@ +# This file is copyright Arch Systems, Inc. +# Except as otherwise provided in the relevant LICENSE file, all rights are reserved. + +import datetime +import logging +import threading +import time +import bable_interface +from iotile.core.dev.config import ConfigManager +from iotile.core.hw.reports import IOTileReportParser, IOTileReading, BroadcastReport +from iotile.core.hw.transport.adapter import DeviceAdapter +from iotile.core.utilities.packed import unpack +from iotile.core.exceptions import ArgumentError, ExternalError +from .connection_manager import ConnectionManager +from .tilebus import * + +# TODO: release bable interface 1.2.0 (modify READMES...) + + +class NativeBLEDeviceAdapter(DeviceAdapter): + """Device adapter for native BLE controllers, supporting multiple simultaneous connections.""" + + def __init__(self, port, on_scan=None, on_disconnect=None, active_scan=None, **kwargs): + super(NativeBLEDeviceAdapter, self).__init__() + + # Create logger + self._logger = logging.getLogger(__name__) + self._logger.addHandler(logging.NullHandler()) + + # Register configuration + self.set_config('minimum_scan_time', 2.0) # Time to accumulate device advertising packets first + self.set_config('default_timeout', 10.0) # Time before timeout an operation + self.set_config('expiration_time', 60.0) # Time before a scanned device expired + self.set_config('maximum_connections', 3) # Maximum number of simultaneous connections per controller + + # Create the baBLE interface to interact with BLE controllers + self.bable = bable_interface.BaBLEInterface() + + # Get the list of BLE controllers + self.bable.start(on_error=self._on_ble_error) + controllers = self._find_ble_controllers() + self.bable.stop() + + if len(controllers) == 0: + raise ExternalError("Could not find any BLE controller connected to this computer") + + # Parse port and check if it exists + if port is None or port == '': + self.controller_id = controllers[0].id + else: + self.controller_id = int(port) + if not any(controller.id == self.controller_id for controller in controllers): + raise ExternalError("Could not find a BLE controller with the given ID, controller_id=%s" + .format(self.controller_id)) + + # Restart baBLE with the selected controller id to prevent conflicts if multiple controllers + self.bable.start(on_error=self._on_ble_error, exit_on_sigint=False, controller_id=self.controller_id) + + # Register callbacks + if on_scan is not None: + self.add_callback('on_scan', on_scan) + if on_disconnect is not None: + self.add_callback('on_disconnect', on_disconnect) + + self.scanning = False + self.stopped = False + + if active_scan is not None: + self._active_scan = active_scan + else: + config = ConfigManager() + self._active_scan = config.get('ble:active-scan') + + # To register advertising packets waiting for a scan response (only if active scan) + self.partial_scan_responses = {} + + # To manage multiple connections + self.connections = ConnectionManager(self.id) + self.connections.start() + + # Notification callbacks + self.notification_callbacks_lock = threading.Lock() + self.notification_callbacks = {} + + try: + self._initialize_system_sync() + self.start_scan(active=self._active_scan) + except Exception: + self.stop_sync() + raise + + def _find_ble_controllers(self): + """Get a list of the available and powered BLE controllers""" + controllers = self.bable.list_controllers() + return [ctrl for ctrl in controllers if ctrl.powered and ctrl.low_energy] + + def _on_ble_error(self, status, message): + """Callback function called when a BLE error, not related to a request, is received. Just log it for now.""" + self._logger.error("BLE error (status=%s, message=%s)", status, message) + + def _initialize_system_sync(self): + """Initialize the device adapter by removing all active connections and resetting scan and advertising to have + a clean starting state.""" + connected_devices = self.bable.list_connected_devices() + for device in connected_devices: + context = { + 'connection_id': len(self.connections.get_connections()), + 'connection_handle': device.connection_handle, + 'connection_string': device.address + } + self.connections.add_connection(context['connection_id'], device.address, context) + self.disconnect_sync(context['connection_id']) + + self.stop_scan() + + try: + self.bable.set_advertising(enabled=False) + except bable_interface.BaBLEException: + # If advertising is already disabled + pass + + def can_connect(self): + """Check if this adapter can take another connection + + Returns: + bool: whether there is room for one more connection + """ + return len(self.connections.get_connections()) < int(self.get_config('maximum_connections')) + + def start_scan(self, active): + """Start a scan. Will call self._on_device_found for each device scanned. + Args: + active (bool): Indicate if it is an active scan (probing for scan response) or not. + """ + try: + self.bable.start_scan(self._on_device_found, active_scan=active, sync=True) + except bable_interface.BaBLEException as err: + # If we are already scanning, raise an error only we tried to change the active scan param + if self._active_scan != active: + raise err + + self._active_scan = active + self.scanning = True + + def _on_device_found(self, success, device, failure_reason): + """Callback function called when a device has been scanned. + It is executed in the baBLE working thread: should not be blocking. + + Args: + success (bool): A bool indicating that the operation is successful or not + device (dict): The scanned device information + - type (int): Indicates if it is an advertising report or a scan response + - uuid (uuid.UUID): The service uuid + - manufacturer_data (bytes): The manufacturer data + - address (str): The device BT address + - address_type (str): The device address type (either 'random' or 'public') + - rssi (int): The signal strength + failure_reason (any): An object indicating the reason why the operation is not successful (else None) + """ + if not success: + self._logger.error("on_device_found() callback called with error: ", failure_reason) + return + + # If it is an adverting report + if device['type'] in [0x00, 0x01, 0x02]: + # If it has the TileBusService + if device['uuid'] == TileBusService.uuid: + if len(device['manufacturer_data']) != 6: + self._logger.error("Received advertisement response with wrong manufacturer data length " + "(expected=6, received=%d)", len(device['manufacturer_data'])) + return + + device_uuid, flags = unpack(" format + connection_id (int): A unique integer set by the caller for referring to this connection once created + callback (callable): A callback function called when the connection has succeeded or failed + retries (int): The number of attempts to connect to this device that can end in early disconnect + before we give up and report that we could not connect. A retry count of 0 will mean that + we fail as soon as we receive the first early disconnect. + context (dict): If we are retrying to connect, passes the context to not considering it as a new connection. + """ + if context is None: + # It is the first attempt to connect: begin a new connection + context = { + 'connection_id': connection_id, + 'retries': retries, + 'retry_connect': False, + 'connection_string': connection_string, + 'connect_time': time.time(), + 'callback': callback + } + + self.connections.begin_connection( + connection_id, + connection_string, + callback, + context, + self.get_config('default_timeout') + ) + + # Don't scan while we attempt to connect to this device + if self.scanning: + self.stop_scan() + + address, address_type = connection_string.split(',') + + # First, cancel any pending connection to prevent errors when starting a new one + self.bable.cancel_connection(sync=False) + + # Send a connect request + self.bable.connect( + address=address, + address_type=address_type, + connection_interval=[7.5, 7.5], + on_connected=[self._on_connection_finished, context], + on_disconnected=[self._on_unexpected_disconnection, context] + ) + + def _on_connection_finished(self, success, result, failure_reason, context): + """Callback called when the connection attempt to a BLE device has finished. + It is executed in the baBLE working thread: should not be blocking. + + Args: + success (bool): A bool indicating that the operation is successful or not + result (dict): The connection information (if successful) + - connection_handle (int): The connection handle + - address (str): The device BT address + - address_type (str): The device address type (either 'random' or 'public') + failure_reason (any): An object indicating the reason why the operation is not successful (else None) + """ + connection_id = context['connection_id'] + + if not success: + self._logger.error("Error while connecting to the device err=%s", failure_reason) + + # If connection failed to be established, we just should retry to connect + if failure_reason.packet.native_class == 'HCI' and failure_reason.packet.native_status == 0x3e: + context['retry_connect'] = True + + self._on_connection_failed(connection_id, self.id, success, failure_reason) + return + + context['connection_handle'] = result['connection_handle'] + + # After connection has been done, probe GATT services + self.bable.probe_services( + connection_handle=context['connection_handle'], + on_services_probed=[self._on_services_probed, context] + ) + + def _on_services_probed(self, success, result, failure_reason, context): + """Callback called when the services has been probed. + It is executed in the baBLE working thread: should not be blocking. + + Args: + success (bool): A bool indicating that the operation is successful or not + result (dict): Information probed (if successful) + - services (list): The list of services probed (bable_interface.Service instances) + failure_reason (any): An object indicating the reason why the operation is not successful (else None) + """ + connection_id = context['connection_id'] + + if not success: + self._logger.error("Error while probing services to the device, err=%s", failure_reason) + context['failure_reason'] = "Error while probing services" + self.disconnect_async(connection_id, self._on_connection_failed) + return + + services = {service: {} for service in result['services']} + + # Validate that this is a proper IOTile device + if TileBusService not in services: + context['failure_reason'] = 'TileBus service not present in GATT services' + self.disconnect_async(connection_id, self._on_connection_failed) + return + + context['services'] = services + + # Finally, probe GATT characteristics + self.bable.probe_characteristics( + connection_handle=context['connection_handle'], + start_handle=TileBusService.handle, + end_handle=TileBusService.group_end_handle, + on_characteristics_probed=[self._on_characteristics_probed, context] + ) + + def _on_characteristics_probed(self, success, result, failure_reason, context): + """Callback called when the characteristics has been probed. + It is executed in the baBLE working thread: should not be blocking. + + Args: + success (bool): A bool indicating that the operation is successful or not + result (dict): Information probed (if successful) + - characteristics (list): The list of characteristics probed (bable_interface.Characteristic instances) + failure_reason (any): An object indicating the reason why the operation is not successful (else None) + """ + connection_id = context['connection_id'] + + if not success: + self._logger.error("Error while probing characteristics to the device, err=%s", failure_reason) + context['failure_reason'] = "Error while probing characteristics" + self.disconnect_async(connection_id, self._on_connection_failed) + return + + context['services'][TileBusService] = { + characteristic: characteristic for characteristic in result['characteristics'] + } + + total_time = time.time() - context['connect_time'] + self._logger.info("Total time to connect to device: %.3f", total_time) + + self.connections.finish_connection( + connection_id, + success, + failure_reason + ) + + def _on_connection_failed(self, connection_id, adapter_id, success, failure_reason): + """Callback function called when a connection has failed. + It is executed in the baBLE working thread: should not be blocking. + + Args: + connection_id (int): A unique identifier for this connection on the DeviceManager that owns this adapter. + adapter_id (int): A unique identifier for the DeviceManager + success (bool): A bool indicating that the operation is successful or not + failure_reason (any): An object indicating the reason why the operation is not successful (else None) + """ + self._logger.info("_on_connection_failed connection_id=%d, reason=%s", connection_id, failure_reason) + + try: + context = self.connections.get_context(connection_id) + except ArgumentError: + self._logger.info("Unable to obtain connection data on unknown connection %d", connection_id) + context = {} + + # Cancel the connection to be able to resend a connect request later (else the controller sends an error) + self.bable.cancel_connection(sync=False) + + if context.get('retry_connect') and context.get('retries') > 0: + context['retries'] -= 1 + self.connect_async( + connection_id, + context['connection_string'], + context['callback'], + context['retries'], + context + ) + else: + self.connections.finish_connection( + connection_id, + False, + context.get('failure_reason', failure_reason) + ) + + def disconnect_async(self, connection_id, callback): + """Asynchronously disconnect from a device that has previously been connected + + Args: + connection_id (int): A unique identifier for this connection on the DeviceManager that owns this adapter. + callback (callable): A function called as callback(connection_id, adapter_id, success, failure_reason) + when the disconnection finishes. Disconnection can only either succeed or timeout. + """ + + try: + context = self.connections.get_context(connection_id) + except ArgumentError: + callback(connection_id, self.id, False, "Could not find connection information") + return + + self.connections.begin_disconnection(connection_id, callback, self.get_config('default_timeout')) + + self.bable.disconnect( + connection_handle=context['connection_handle'], + on_disconnected=[self._on_disconnection_finished, context] + ) + + def _on_unexpected_disconnection(self, success, result, failure_reason, context): + """Callback function called when an unexpected disconnection occured (meaning that we didn't previously send + a `disconnect` request). + It is executed in the baBLE working thread: should not be blocking. + + Args: + success (bool): A bool indicating that the operation is successful or not + result (dict): Disconnection information (if successful) + - connection_handle (int): The connection handle that just disconnected + - code (int): The reason code + - reason (str): A message explaining the reason code in plain text + failure_reason (any): An object indicating the reason why the operation is not successful (else None) + context (dict): The connection context + """ + connection_id = context['connection_id'] + + self._logger.warn('Unexpected disconnection event, handle=%d, reason=0x%X, state=%s', + result['connection_handle'], + result['code'], + self.connections.get_state(connection_id)) + + self.connections.unexpected_disconnect(connection_id) + self._trigger_callback('on_disconnect', self.id, connection_id) + + def _on_disconnection_finished(self, success, result, failure_reason, context): + """Callback function called when a previously asked disconnection has been finished. + It is executed in the baBLE working thread: should not be blocking. + + Args: + success (bool): A bool indicating that the operation is successful or not + result (dict): Disconnection information (if successful) + - connection_handle (int): The connection handle that just disconnected + - code (int): The reason code + - reason (str): A message explaining the reason code in plain text + failure_reason (any): An object indicating the reason why the operation is not successful (else None) + context (dict): The connection context + """ + if 'connection_handle' in context: + # Remove all the notification callbacks registered for this connection + with self.notification_callbacks_lock: + for connection_handle, attribute_handle in list(self.notification_callbacks.keys()): + if connection_handle == context['connection_handle']: + del self.notification_callbacks[(connection_handle, attribute_handle)] + + self.connections.finish_disconnection( + context['connection_id'], + success, + failure_reason + ) + + def _open_rpc_interface(self, connection_id, callback): + """Enable RPC interface for this IOTile device + + Args: + connection_id (int): The unique identifier for the connection + callback (callback): Callback to be called when this command finishes + callback(conn_id, adapter_id, success, failure_reason) + """ + + try: + context = self.connections.get_context(connection_id) + except ArgumentError: + callback(connection_id, self.id, False, "Could not find connection information") + return + + self.connections.begin_operation(connection_id, 'open_interface', callback, self.get_config('default_timeout')) + + try: + service = context['services'][TileBusService] + header_characteristic = service[ReceiveHeaderChar] + payload_characteristic = service[ReceivePayloadChar] + except KeyError: + self.connections.finish_operation(connection_id, False, "Can't find characteristics to open rpc interface") + return + + # Enable notification from ReceiveHeaderChar characteristic (ReceivePayloadChar will be enable just after) + self.bable.set_notification( + enabled=True, + connection_handle=context['connection_handle'], + characteristic=header_characteristic, + on_notification_set=[self._on_interface_opened, context, payload_characteristic], + on_notification_received=self._on_notification_received, + sync=False + ) + + def _open_script_interface(self, connection_id, callback): + """Enable script streaming interface for this IOTile device + + Args: + connection_id (int): The unique identifier for the connection + callback (callback): Callback to be called when this command finishes + callback(conn_id, adapter_id, success, failure_reason) + """ + + try: + context = self.connections.get_context(connection_id) + except ArgumentError: + callback(connection_id, self.id, False, "Could not find connection information") + return + + success = HighSpeedChar in context['services'][TileBusService] + reason = None + if not success: + reason = 'Could not find high speed streaming characteristic' + + callback(connection_id, self.id, success, reason) + + def _open_streaming_interface(self, connection_id, callback): + """Enable streaming interface for this IOTile device + + Args: + connection_id (int): The unique identifier for the connection + callback (callback): Callback to be called when this command finishes + callback(conn_id, adapter_id, success, failure_reason) + """ + + try: + context = self.connections.get_context(connection_id) + except ArgumentError: + callback(connection_id, self.id, False, "Could not find connection information") + return + + self._logger.info("Attempting to enable streaming") + self.connections.begin_operation(connection_id, 'open_interface', callback, self.get_config('default_timeout')) + + try: + characteristic = context['services'][TileBusService][StreamingChar] + except KeyError: + self.connections.finish_operation( + connection_id, + False, + "Can't find characteristic to open streaming interface" + ) + return + + context['parser'] = IOTileReportParser(report_callback=self._on_report, error_callback=self._on_report_error) + context['parser'].context = connection_id + + def on_report_chunk_received(report_chunk): + """Callback function called when a report chunk has been received.""" + context['parser'].add_data(report_chunk) + + # Register our callback function in the notifications callbacks + self._register_notification_callback( + context['connection_handle'], + characteristic.value_handle, + on_report_chunk_received + ) + + self.bable.set_notification( + enabled=True, + connection_handle=context['connection_handle'], + characteristic=characteristic, + on_notification_set=[self._on_interface_opened, context], + on_notification_received=self._on_notification_received, + timeout=1.0, + sync=False + ) + + def _open_tracing_interface(self, connection_id, callback): + """Enable the tracing interface for this IOTile device + + Args: + connection_id (int): The unique identifier for the connection + callback (callback): Callback to be called when this command finishes + callback(conn_id, adapter_id, success, failure_reason) + """ + + try: + context = self.connections.get_context(connection_id) + except ArgumentError: + callback(connection_id, self.id, False, "Could not find connection information") + return + + self._logger.info("Attempting to enable tracing") + self.connections.begin_operation(connection_id, 'open_interface', callback, self.get_config('default_timeout')) + + try: + characteristic = context['services'][TileBusService][TracingChar] + except KeyError: + self.connections.finish_operation( + connection_id, + False, + "Can't find characteristic to open tracing interface" + ) + return + + # Register a callback function in the notifications callbacks, to trigger `on_trace` callback when a trace is + # notified. + self._register_notification_callback( + context['connection_handle'], + characteristic.value_handle, + lambda trace_chunk: self._trigger_callback('on_trace', connection_id, bytearray(trace_chunk)) + ) + + self.bable.set_notification( + enabled=True, + connection_handle=context['connection_handle'], + characteristic=characteristic, + on_notification_set=[self._on_interface_opened, context], + on_notification_received=self._on_notification_received, + timeout=1.0, + sync=False + ) + + def _on_interface_opened(self, success, result, failure_reason, context, next_characteristic=None): + """Callback function called when the notification related to an interface has been enabled. + It is executed in the baBLE working thread: should not be blocking. + + Args: + success (bool): A bool indicating that the operation is successful or not + result (dict): Information (if successful) + failure_reason (any): An object indicating the reason why the operation is not successful (else None) + context (dict): The connection context + next_characteristic (bable_interface.Characteristic): If not None, indicate another characteristic to enable + notification. + """ + if not success: + self.connections.finish_operation(context['connection_id'], False, failure_reason) + return + + if next_characteristic is not None: + self.bable.set_notification( + enabled=True, + connection_handle=context['connection_handle'], + characteristic=next_characteristic, + on_notification_set=[self._on_interface_opened, context], + on_notification_received=self._on_notification_received, + sync=False + ) + else: + self.connections.finish_operation(context['connection_id'], True, None) + + def _close_rpc_interface(self, connection_id, callback): + """Disable RPC interface for this IOTile device + + Args: + connection_id (int): The unique identifier for the connection + callback (callback): Callback to be called when this command finishes + callback(conn_id, adapter_id, success, failure_reason) + """ + + try: + context = self.connections.get_context(connection_id) + except ArgumentError: + callback(connection_id, self.id, False, "Could not find connection information") + return + + self.connections.begin_operation(connection_id, 'close_interface', callback, self.get_config('default_timeout')) + + try: + service = context['services'][TileBusService] + header_characteristic = service[ReceiveHeaderChar] + payload_characteristic = service[ReceivePayloadChar] + except KeyError: + self.connections.finish_operation(connection_id, False, "Can't find characteristics to open rpc interface") + return + + self.bable.set_notification( + enabled=False, + connection_handle=context['connection_handle'], + characteristic=header_characteristic, + on_notification_set=[self._on_interface_closed, context, payload_characteristic], + timeout=1.0 + ) + + def _on_interface_closed(self, success, result, failure_reason, context, next_characteristic=None): + """Callback function called when the notification related to an interface has been disabled. + It is executed in the baBLE working thread: should not be blocking. + + Args: + success (bool): A bool indicating that the operation is successful or not + result (dict): Information (if successful) + failure_reason (any): An object indicating the reason why the operation is not successful (else None) + context (dict): The connection context + next_characteristic (bable_interface.Characteristic): If not None, indicate another characteristic to + disable notification. + """ + if not success: + self.connections.finish_operation(context['connection_id'], False, failure_reason) + return + + if next_characteristic is not None: + self.bable.set_notification( + enabled=False, + connection_handle=context['connection_handle'], + characteristic=next_characteristic, + on_notification_set=[self._on_interface_closed, context], + timeout=1.0, + sync=False + ) + else: + self.connections.finish_operation(context['connection_id'], True, None) + + def _on_report(self, report, connection_id): + """Callback function called when a report has been processed. + + Args: + report (IOTileReport): The report object + connection_id (int): The connection id related to this report + + Returns: + - True to indicate that IOTileReportParser should also keep a copy of the report + or False to indicate it should delete it. + """ + self._logger.info('Received report: %s', str(report)) + self._trigger_callback('on_report', connection_id, report) + + return False + + def _on_report_error(self, code, message, connection_id): + """Callback function called if an error occured while parsing a report""" + self._logger.critical( + "Error receiving reports, no more reports will be processed on this adapter, code=%d, msg=%s", code, message + ) + + def send_rpc_async(self, connection_id, address, rpc_id, payload, timeout, callback): + """Asynchronously send an RPC to this IOTile device + + Args: + connection_id (int): A unique identifier that will refer to this connection + address (int): The address of the tile that we wish to send the RPC to + rpc_id (int): The 16-bit id of the RPC we want to call + payload (bytearray): The payload of the command + timeout (float): The number of seconds to wait for the RPC to execute + callback (callable): A callback for when we have finished the RPC. The callback will be called as + callback(connection_id, adapter_id, success, failure_reason, status, payload) + 'connection_id': The connection id + 'adapter_id': This adapter's id + 'success': A bool indicating whether we received a response to our attempted RPC + 'failure_reason': A string with the reason for the failure if success == False + 'status': The one byte status code returned for the RPC if success == True else None + 'payload': A bytearray with the payload returned by RPC if success == True else None + """ + try: + context = self.connections.get_context(connection_id) + except ArgumentError: + callback(connection_id, self.id, False, "Could not find connection information") + return + + connection_handle = context['connection_handle'] + + self.connections.begin_operation(connection_id, 'rpc', callback, timeout) + + try: + service = context['services'][TileBusService] + send_header_characteristic = service[SendHeaderChar] + send_payload_characteristic = service[SendPayloadChar] + receive_header_characteristic = service[ReceiveHeaderChar] + receive_payload_characteristic = service[ReceivePayloadChar] + except KeyError: + self.connections.finish_operation(connection_id, False, "Can't find characteristics to open rpc interface") + return + + length = len(payload) + if length < 20: + payload += b'\x00'*(20 - length) + if length > 20: + self.connections.finish_operation(connection_id, False, "Payload is too long, must be at most 20 bytes") + return + + header = bytearray([length, 0, rpc_id & 0xFF, (rpc_id >> 8) & 0xFF, address]) + result = {} + + def on_header_received(value): + """Callback function called when a notification has been received with the RPC header response.""" + result['status'] = value[0] + result['length'] = value[3] + + if result['length'] == 0: + # Simulate a empty payload received to end the RPC response + self._on_notification_received(True, { + 'connection_handle': connection_handle, + 'attribute_handle': receive_payload_characteristic.value_handle, + 'value': b'\x00'*20 + }, None) + + def on_payload_received(value): + """Callback function called when a notification has been received with the RPC payload response.""" + result['payload'] = value[:result['length']] + self.connections.finish_operation( + connection_id, + True, + None, + result['status'], + result['payload'] + ) + + # Register the header notification callback + self._register_notification_callback( + connection_handle, + receive_header_characteristic.value_handle, + on_header_received, + once=True + ) + + # Register the payload notification callback + self._register_notification_callback( + connection_handle, + receive_payload_characteristic.value_handle, + on_payload_received, + once=True + ) + + if length > 0: + # If payload is not empty, send it first + self.bable.write_without_response( + connection_handle=connection_handle, + attribute_handle=send_payload_characteristic.value_handle, + value=bytes(payload) + ) + + self.bable.write_without_response( + connection_handle=connection_handle, + attribute_handle=send_header_characteristic.value_handle, + value=bytes(header) + ) + + def send_script_async(self, connection_id, data, progress_callback, callback): + """Asynchronously send a a script to this IOTile device + + Args: + connection_id (int): A unique identifier that will refer to this connection + data (bytes): the script to send to the device + progress_callback (callable): A function to be called with status on our progress, called as: + progress_callback(done_count, total_count) + callback (callable): A callback for when we have finished sending the script. The callback will be called as + callback(connection_id, adapter_id, success, failure_reason) + 'connection_id': the connection id + 'adapter_id': this adapter's id + 'success': a bool indicating whether we received a response to our attempted RPC + 'failure_reason': a string with the reason for the failure if success == False + """ + + try: + context = self.connections.get_context(connection_id) + except ArgumentError: + callback(connection_id, self.id, False, "Could not find connection information") + return + + self.connections.begin_operation(connection_id, 'script', callback, self.get_config('default_timeout')) + mtu = int(self.get_config('mtu', 20)) # Split script payloads larger than this + + high_speed_char = context['services'][TileBusService][HighSpeedChar] + + # Count number of chunks to send + nb_chunks = 1 + if len(data) > mtu: + nb_chunks = len(data) // mtu + if len(data) % mtu != 0: + nb_chunks += 1 + + def send_script(): + """Function sending every chunks of the script. Executed in a separated thread.""" + for i in range(0, nb_chunks): + start = i * mtu + chunk = data[start: start + mtu] + sent = False + + while not sent: + try: + self.bable.write_without_response( + connection_handle=context['connection_handle'], + attribute_handle=high_speed_char.value_handle, + value=bytes(chunk) + ) + sent = True + except bable_interface.BaBLEException as err: + if err.packet.status == 'Rejected': # If we are streaming too fast, back off and try again + time.sleep(0.05) + else: + self.connections.finish_operation(connection_id, False, err.message) + return + + progress_callback(i, nb_chunks) + + self.connections.finish_operation(connection_id, True, None) + + # Start the thread to send the script asynchronously + send_script_thread = threading.Thread(target=send_script, name='SendScriptThread') + send_script_thread.daemon = True + send_script_thread.start() + + def _register_notification_callback(self, connection_handle, attribute_handle, callback, once=False): + """Register a callback as a notification callback. It will be called if a notification with the matching + connection_handle and attribute_handle is received. + + Args: + connection_handle (int): The connection handle to watch + attribute_handle (int): The attribute handle to watch + callback (func): The callback function to call once the notification has been received + once (bool): Should the callback only be called once (and then removed from the notification callbacks) + """ + notification_id = (connection_handle, attribute_handle) + with self.notification_callbacks_lock: + self.notification_callbacks[notification_id] = (callback, once) + + def _on_notification_received(self, success, result, failure_reason): + """Callback function called when a notification has been received. + It is executed in the baBLE working thread: should not be blocking. + + Args: + success (bool): A bool indicating that the operation is successful or not + result (dict): The notification information + - value (bytes): Data notified + failure_reason (any): An object indicating the reason why the operation is not successful (else None) + """ + if not success: + self._logger.info("Notification received with failure failure_reason=%s", failure_reason) + + notification_id = (result['connection_handle'], result['attribute_handle']) + + callback = None + with self.notification_callbacks_lock: + if notification_id in self.notification_callbacks: + callback, once = self.notification_callbacks[notification_id] + + if once: + del self.notification_callbacks[notification_id] + + if callback is not None: + callback(result['value']) + + def stop_sync(self): + """Safely stop this BLED112 instance without leaving it in a weird state""" + # Stop to scan + if self.scanning: + self.stop_scan() + + # Disconnect all connected devices + for connection_id in list(self.connections.get_connections()): + self.disconnect_sync(connection_id) + + # Stop the baBLE interface + self.bable.stop() + # Stop the connection manager + self.connections.stop() + + self.stopped = True + + def periodic_callback(self): + """Periodic cleanup tasks to maintain this adapter, should be called every second. """ + + if self.stopped: + return + + # Check if we should start scanning again + if not self.scanning and len(self.connections.get_connections()) == 0: + self._logger.info("Restarting scan for devices") + self.start_scan(self._active_scan) + self._logger.info("Finished restarting scan for devices") diff --git a/transport_plugins/native_ble/iotile_transport_native_ble/tilebus.py b/transport_plugins/native_ble/iotile_transport_native_ble/tilebus.py new file mode 100644 index 000000000..0e3c1167d --- /dev/null +++ b/transport_plugins/native_ble/iotile_transport_native_ble/tilebus.py @@ -0,0 +1,54 @@ +from bable_interface import Characteristic, Service +import uuid + +# Company ID +ArchManuID = 0x03C0 + +# GATT table +BLEService = Service(uuid='1800', handle=0x0001, group_end_handle=0x0005) +NameChar = Characteristic( + uuid='2a00', + handle=0x0002, value_handle=0x0003, const_value=b'V_IOTile ', + read=True) +AppearanceChar = Characteristic( + uuid='2a01', + handle=0x0004, value_handle=0x0005, const_value=b'\x80\x00', + read=True +) + +TileBusService = Service(uuid=uuid.UUID('00002000-3ff7-53ba-e611-132c0ff60f63'), handle=0x000B, group_end_handle=0xFFFF) +ReceiveHeaderChar = Characteristic( + uuid=uuid.UUID('00002001-0000-1000-8000-00805f9b34fb'), + handle=0x000C, value_handle=0x000D, config_handle=0x000E, + notify=True +) +ReceivePayloadChar = Characteristic( + uuid=uuid.UUID('00002002-0000-1000-8000-00805f9b34fb'), + handle=0x000F, value_handle=0x0010, config_handle=0x0011, + notify=True +) +SendHeaderChar = Characteristic( + uuid=uuid.UUID('00002003-0000-1000-8000-00805f9b34fb'), + handle=0x0012, value_handle=0x0013, + write=True +) +SendPayloadChar = Characteristic( + uuid=uuid.UUID('00002004-0000-1000-8000-00805f9b34fb'), + handle=0x0014, value_handle=0x0015, + write=True +) +StreamingChar = Characteristic( + uuid=uuid.UUID('00002005-0000-1000-8000-00805f9b34fb'), + handle=0x0016, value_handle=0x0017, config_handle=0x0018, + notify=True +) +HighSpeedChar = Characteristic( + uuid=uuid.UUID('00002006-0000-1000-8000-00805f9b34fb'), + handle=0x0019, value_handle=0x001A, + write=True +) +TracingChar = Characteristic( + uuid=uuid.UUID('00002007-0000-1000-8000-00805f9b34fb'), + handle=0x001B, value_handle=0x001C, config_handle=0x001D, + notify=True +) diff --git a/transport_plugins/native_ble/iotile_transport_native_ble/virtual_ble.py b/transport_plugins/native_ble/iotile_transport_native_ble/virtual_ble.py new file mode 100644 index 000000000..1cd55df47 --- /dev/null +++ b/transport_plugins/native_ble/iotile_transport_native_ble/virtual_ble.py @@ -0,0 +1,512 @@ +"""A VirtualInterface that provides access to a virtual IOTile device using native BLE""" + +# This file is copyright Arch Systems, Inc. +# Except as otherwise provided in the relevant LICENSE file, all rights are reserved. + +import struct +import bable_interface +import logging +import time +import binascii +from iotile.core.exceptions import ExternalError +from iotile.core.hw.virtual.virtualinterface import VirtualIOTileInterface +from iotile.core.hw.virtual.virtualdevice import RPCInvalidIDError, RPCNotFoundError, TileNotFoundError +from .tilebus import * + + +class NativeBLEVirtualInterface(VirtualIOTileInterface): + """Turn a BLE adapter into a virtual IOTile + + Args: + args (dict): A dictionary of arguments used to configure this interface. + Currently the only supported argument is 'port' which should be a + valid controller id (given by `sudo hcitool dev` on Linux, X in hciX) + """ + + def __init__(self, args): + super(NativeBLEVirtualInterface, self).__init__() + + # Create logger + self._logger = logging.getLogger(__name__) + self._logger.addHandler(logging.NullHandler()) + + # Create the baBLE interface to interact with BLE controllers + self.bable = bable_interface.BaBLEInterface() + + # Get the list of BLE controllers + self.bable.start(on_error=self._on_ble_error) + controllers = self._find_ble_controllers() + self.bable.stop() + + if len(controllers) == 0: + raise ExternalError("Could not find any BLE controller connected to this computer") + + # Parse args + port = None + if 'port' in args: + port = args['port'] + + if port is None or port == '': + self.controller_id = controllers[0].id + else: + self.controller_id = int(port) + if not any(controller.id == self.controller_id for controller in controllers): + raise ExternalError("Could not find a BLE controller with the given ID, controller_id=%s" + .format(self.controller_id)) + + if 'voltage' in args: + self.voltage = float(args['voltage']) + else: + self.voltage = 3.8 + + # Restart baBLE with the selected controller id to prevent conflicts if multiple controllers + self.bable.start(on_error=self._on_ble_error, exit_on_sigint=False, controller_id=self.controller_id) + # Register the callback function into baBLE + self.bable.on_write_request(self._on_write_request) + self.bable.on_connected(self._on_connected) + self.bable.on_disconnected(self._on_disconnected) + + # Initialize state + self.connected = False + self._connection_handle = 0 + + self.payload_notif = False + self.header_notif = False + self.streaming = False + self.tracing = False + + # Keep track of whether we've launched our state machine + # to stream or trace data so that when we find more data available + # in process() we know not to restart the streaming/tracing process + self._stream_sm_running = False + self._trace_sm_running = False + + self.rpc_payload = bytearray(20) + self.rpc_header = bytearray(20) + + try: + self._initialize_system_sync() + except Exception: + self.stop_sync() + raise + + def _find_ble_controllers(self): + """Get a list of the available and powered BLE controllers""" + controllers = self.bable.list_controllers() + return [ctrl for ctrl in controllers if ctrl.powered and ctrl.low_energy] + + def _on_ble_error(self, status, message): + """Callback function called when a BLE error, not related to a request, is received. Just log it for now.""" + self._logger.error("BLE error (status=%s, message=%s)", status, message) + + def _initialize_system_sync(self): + """Initialize the device adapter by removing all active connections and resetting scan and advertising to have + a clean starting state.""" + connected_devices = self.bable.list_connected_devices() + for device in connected_devices: + self.disconnect_sync(device.connection_handle) + + self.stop_scan() + + self.set_advertising(False) + + # Register the GATT table to send the right services and characteristics when probed (like an IOTile device) + self.register_gatt_table() + + def start(self, device): + """Start serving access to this VirtualIOTileDevice + + Args: + device (VirtualIOTileDevice): The device we will be providing access to + """ + + super(NativeBLEVirtualInterface, self).start(device) + self.set_advertising(True) + + def stop(self): + """Safely shut down this interface + """ + + super(NativeBLEVirtualInterface, self).stop() + + self.stop_sync() + + def register_gatt_table(self): + """Register the GATT table into baBLE.""" + services = [BLEService, TileBusService] + + characteristics = [ + NameChar, + AppearanceChar, + ReceiveHeaderChar, + ReceivePayloadChar, + SendHeaderChar, + SendPayloadChar, + StreamingChar, + HighSpeedChar, + TracingChar + ] + + self.bable.set_gatt_table(services, characteristics) + + def stop_scan(self): + """Stop to scan.""" + try: + self.bable.stop_scan(sync=True) + except bable_interface.BaBLEException: + # If we errored our it is because we were not currently scanning + pass + + def set_advertising(self, enabled): + """Toggle advertising.""" + if enabled: + self.bable.set_advertising( + enabled=True, + uuids=[TileBusService.uuid], + name="V_IOTile ", + company_id=ArchManuID, + advertising_data=self._advertisement(), + scan_response=self._scan_response(), + sync=True + ) + else: + try: + self.bable.set_advertising(enabled=False, sync=True) + except bable_interface.BaBLEException: + # If advertising is already disabled + pass + + def _advertisement(self): + """Create advertisement data.""" + # Flags are + # bit 0: whether we have pending data + # bit 1: whether we are in a low voltage state + # bit 2: whether another user is connected + # bit 3: whether we support robust reports + # bit 4: whether we allow fast writes + + flags = int(self.device.pending_data) | (0 << 1) | (0 << 2) | (1 << 3) | (1 << 4) + return struct.pack(" 0: + status |= (1 << 7) + except (RPCInvalidIDError, RPCNotFoundError): + status = 2 # FIXME: Insert the correct ID here + response = b'' + except TileNotFoundError: + status = 0xFF + response = b'' + except Exception: + status = 3 + response = b'' + self._logger.exception("Exception raise while calling rpc, header=%s, payload=%s", header, payload) + + self._audit( + "RPCReceived", + rpc_id=rpc_id, + address=address, + payload=binascii.hexlify(payload), + status=status, + response=binascii.hexlify(response) + ) + + resp_header = struct.pack(" 0: + self._send_rpc_response( + (ReceiveHeaderChar.value_handle, resp_header), + (ReceivePayloadChar.value_handle, response) + ) + else: + self._send_rpc_response((ReceiveHeaderChar.value_handle, resp_header)) + + def _send_notification(self, handle, payload): + """Send a notification over BLE + It is executed in the baBLE working thread: should not be blocking. + + Args: + handle (int): The handle to notify on + payload (bytearray): The value to notify + """ + + self.bable.notify( + connection_handle=self._connection_handle, + attribute_handle=handle, + value=payload + ) + + def _send_rpc_response(self, *packets): + """Send an RPC response. + It is executed in the baBLE working thread: should not be blocking. + + The RPC response is notified in one or two packets depending on whether or not + response data is included. If there is a temporary error sending one of the packets + it is retried automatically. If there is a permanent error, it is logged and the response + is abandoned. + """ + + if len(packets) == 0: + return + + handle, payload = packets[0] + + try: + self._send_notification(handle, payload) + except bable_interface.BaBLEException as err: + if err.packet.status == 'Rejected': # If we are streaming too fast, back off and try again + time.sleep(0.05) + self._defer(self._send_rpc_response, list(packets)) + else: + self._audit('ErrorSendingRPCResponse') + self._logger.exception("Error while sending RPC response, handle=%s, payload=%s", handle, payload) + + return + + if len(packets) > 1: + self._defer(self._send_rpc_response, list(packets[1:])) + + def _stream_data(self, chunk=None): + """Stream reports to the ble client in 20 byte chunks + + Args: + chunk (bytearray): A chunk that should be sent instead of requesting a + new chunk from the pending reports. + """ + + # If we failed to transmit a chunk, we will be requeued with an argument + self._stream_sm_running = True + + if chunk is None: + chunk = self._next_streaming_chunk(20) + + if chunk is None or len(chunk) == 0: + self._stream_sm_running = False + return + + try: + self._send_notification(StreamingChar.value_handle, chunk) + self._defer(self._stream_data) + except bable_interface.BaBLEException as err: + if err.packet.status == 'Rejected': # If we are streaming too fast, back off and try again + time.sleep(0.05) + self._defer(self._stream_data, [chunk]) + else: + self._audit('ErrorStreamingReport') # If there was an error, stop streaming but don't choke + self._logger.exception("Error while streaming data") + + def _send_trace(self, chunk=None): + """Stream tracing data to the ble client in 20 byte chunks + + Args: + chunk (bytearray): A chunk that should be sent instead of requesting a + new chunk from the pending reports. + """ + + self._trace_sm_running = True + # If we failed to transmit a chunk, we will be requeued with an argument + if chunk is None: + chunk = self._next_tracing_chunk(20) + + if chunk is None or len(chunk) == 0: + self._trace_sm_running = False + return + + try: + self._send_notification(TracingChar.value_handle, chunk) + self._defer(self._send_trace) + except bable_interface.BaBLEException as err: + if err.packet.status == 'Rejected': # If we are streaming too fast, back off and try again + time.sleep(0.05) + self._defer(self._send_trace, [chunk]) + else: + self._audit('ErrorStreamingTrace') # If there was an error, stop streaming but don't choke + self._logger.exception("Error while tracing data") + + def process(self): + """Periodic nonblocking processes""" + + super(NativeBLEVirtualInterface, self).process() + + if (not self._stream_sm_running) and (not self.reports.empty()): + self._stream_data() + + if (not self._trace_sm_running) and (not self.traces.empty()): + self._send_trace() diff --git a/transport_plugins/native_ble/setup.py b/transport_plugins/native_ble/setup.py new file mode 100644 index 000000000..1b24d3c2b --- /dev/null +++ b/transport_plugins/native_ble/setup.py @@ -0,0 +1,41 @@ +from setuptools import setup, find_packages + +import version + +setup( + name="iotile-transport-native-ble", + packages=find_packages(exclude=("test",)), + version=version.version, + license="LGPLv3", + install_requires=[ + "iotile-core>=3.6.2", + "monotonic", + "bable-interface>=1.2.0" + ], + entry_points={'iotile.device_adapter': ['ble = iotile_transport_native_ble.device_adapter:NativeBLEDeviceAdapter'], + 'iotile.virtual_interface': ['ble = iotile_transport_native_ble.virtual_ble:NativeBLEVirtualInterface'], + 'iotile.config_variables': ['ble = iotile_transport_native_ble.config_variables:get_variables']}, + description="IOTile Native BLE Transport Plugin", + author="Arch", + author_email="info@arch-iot.com", + url="https://github.com/iotile/coretools", + keywords=["iotile", "arch", "embedded", "hardware", "firmware", "ble", "bluetooth"], + classifiers=[ + "Programming Language :: Python", + "Programming Language :: Python :: 2", + "Programming Language :: Python :: 2.7", + "Programming Language :: Python :: 3", + "Programming Language :: Python :: 3.6", + "Development Status :: 3 - Alpha", + "License :: OSI Approved :: GNU Library or Lesser General Public License (LGPL)", + "Operating System :: Unix", # FIXME: change as soon as bable-interface will be deployed on Windows and Mac + "Topic :: Software Development :: Libraries :: Python Modules" + ], + long_description="""\ +IOTile Native BLE Transport Plugin +------------------------------- + +A python plugin for the IOTile framework that allows communication with IOTile devices over +Bluetooth Smart using the native Bluetooth controller, embedded in your computer. See https://www.arch-iot.com. +""" +) diff --git a/transport_plugins/native_ble/tests/__init__.py b/transport_plugins/native_ble/tests/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/transport_plugins/native_ble/tests/conftest.py b/transport_plugins/native_ble/tests/conftest.py new file mode 100644 index 000000000..8f40ced0e --- /dev/null +++ b/transport_plugins/native_ble/tests/conftest.py @@ -0,0 +1,95 @@ +# Pytest configuration file: will be run before tests to define fixtures +import os +import pytest +import sys +import threading +import time + +skip_imports = False +if sys.platform not in ['linux', 'linux2']: + collect_ignore = [f for f in os.listdir(os.path.dirname(__file__)) if f.startswith('test_')] + skip_imports = True + +if not skip_imports: + from iotile_transport_native_ble.device_adapter import NativeBLEDeviceAdapter + from iotile_transport_native_ble.virtual_ble import NativeBLEVirtualInterface + from iotile_transport_native_ble.tilebus import * + + +@pytest.fixture(scope='function') +def connected_device_adapter(mock_bable): + """Create an already connected NativeBLEDeviceAdapter, using the mocked BaBLEInterface.""" + device_address = '11:11:11:11:11:11' + device_address_type = 'random' + + callback_called = threading.Event() + + def on_connected(connection_id, adapter_id, success, failure_reason): + """Callback function called when a connection has been completed (succeeded or failed).""" + assert success is True + callback_called.set() + + device_adapter = NativeBLEDeviceAdapter(port=1) + device_adapter.set_config('default_timeout', 0.2) + + # Register the GATT table + services = [BLEService, TileBusService] + characteristics = [ + NameChar, + AppearanceChar, + ReceiveHeaderChar, + ReceivePayloadChar, + SendHeaderChar, + SendPayloadChar, + StreamingChar, + HighSpeedChar, + TracingChar + ] + mock_bable.set_gatt_table(services, characteristics) + + device_adapter.connect_async(0, '{},{}'.format(device_address, device_address_type), on_connected) + + time.sleep(0.1) # Wait for the connection manager to process the connection + + mock_bable.simulate_connected_event(device_adapter.controller_id, device_address) + callback_called.wait(timeout=10) + + return device_adapter, mock_bable + + +@pytest.fixture(scope="function") +def virtual_interface(mock_bable, request): + """Initialize the NativeBLEVirtualInterface with a given device.""" + port = request.param['port'] + config = { + 'port': port + } + + device = request.param['device'] + + interface = NativeBLEVirtualInterface(config) + interface.start(device) + + # Call the process function periodically (in a separated thread) + def _call_process(): + while True: + interface.process() + time.sleep(0.1) + + process_thread = threading.Thread(target=_call_process, name='VirtualInterfaceProcessThread') + process_thread.daemon = True + process_thread.start() + + return mock_bable, interface + + +@pytest.fixture(scope="function") +def connected_virtual_interface(virtual_interface): + """Create an already connected NativeBLEVirtualInterface.""" + mock_bable, interface = virtual_interface + device_address = '11:11:11:11:11:11' + + mock_bable.simulate_connected_event(interface.controller_id, device_address) + assert interface.connected is True + + return mock_bable, interface diff --git a/transport_plugins/native_ble/tests/devices_factory.py b/transport_plugins/native_ble/tests/devices_factory.py new file mode 100644 index 000000000..2024df3f5 --- /dev/null +++ b/transport_plugins/native_ble/tests/devices_factory.py @@ -0,0 +1,33 @@ +import json +import os +from iotile.mock.devices import ReportTestDevice, TracingTestDevice + + +# ====== Report test device ====== + +def build_report_device(): + with open(os.path.join(os.path.dirname(__file__), 'report_device_config.json'), "rb") as conf_file: + config = json.load(conf_file) + + return ReportTestDevice(config['device']) + + +def get_report_device_string(): + config_path = os.path.join(os.path.dirname(__file__), 'report_device_config.json') + + return 'report_test@{}'.format(config_path) + + +# ====== Tracing test device ====== + +def build_tracing_device(): + with open(os.path.join(os.path.dirname(__file__), 'tracing_device_config.json'), "rb") as conf_file: + config = json.load(conf_file) + + return TracingTestDevice(config['device']) + + +def get_tracing_device_string(): + config_path = os.path.join(os.path.dirname(__file__), 'tracing_device_config.json') + + return 'tracing_test@{}'.format(config_path) diff --git a/transport_plugins/native_ble/tests/report_device_config.json b/transport_plugins/native_ble/tests/report_device_config.json new file mode 100644 index 000000000..fa50d9639 --- /dev/null +++ b/transport_plugins/native_ble/tests/report_device_config.json @@ -0,0 +1,9 @@ +{ + "device": { + "iotile_id": "0x10", + "num_readings": 4, + "format": "signed_list", + "report_length": 2, + "stream_id": "0x2000" + } +} diff --git a/transport_plugins/native_ble/tests/test_device_adapter.py b/transport_plugins/native_ble/tests/test_device_adapter.py new file mode 100644 index 000000000..b1aa09dad --- /dev/null +++ b/transport_plugins/native_ble/tests/test_device_adapter.py @@ -0,0 +1,624 @@ +""" +Unit tests for native BLE transport plugin - tests of the NativeBLEDeviceAdapter. +""" + +import pytest +import struct +import threading +import time +from bable_interface import Controller +from iotile.core.exceptions import ExternalError +from iotile.core.hw.hwmanager import HardwareManager +from iotile.core.hw.reports import BroadcastReport, IOTileReading, IndividualReadingReport +from iotile_transport_native_ble.device_adapter import NativeBLEDeviceAdapter +from iotile_transport_native_ble.tilebus import * + + +def test_start_stop(mock_bable): + """Test if device adapter starts and stops correctly.""" + device_adapter = NativeBLEDeviceAdapter(port=None) + + # Test if it has been started + assert mock_bable.counters['start'] > 0 + assert device_adapter.stopped is False + + num_stop_bable = mock_bable.counters['stop'] # To count number of calls to bable_interface stop function + + device_adapter.stop_sync() + # Test if it has been stopped + assert mock_bable.counters['stop'] == num_stop_bable + 1 + assert device_adapter.stopped is True + + +def test_find_ble_controllers(mock_bable_no_ctrl): + """Test if the function to get the controllers available works.""" + # Test with no controllers + with pytest.raises(ExternalError): + NativeBLEDeviceAdapter(port=None) + + # Test with one controller (valid) + controller_valid = [Controller(1, '22:33:44:55:66:11', '#1', settings={'powered': True, 'low_energy': True})] + mock_bable_no_ctrl.set_controllers(controller_valid) + device_adapter = NativeBLEDeviceAdapter(port=None) + assert device_adapter.controller_id == 1 + + # Test with one controller (not valid) + controller_not_valid = [Controller(0, '11:22:33:44:55:66', '#0')] + mock_bable_no_ctrl.set_controllers(controller_not_valid) + with pytest.raises(ExternalError): + NativeBLEDeviceAdapter(port=None) + + # Test with multiple controllers + controller_only_powered = [Controller(2, '33:44:55:66:11:22', '#2', settings={'powered': True})] + mock_bable_no_ctrl.set_controllers(controller_valid + controller_not_valid + controller_only_powered) + device_adapter = NativeBLEDeviceAdapter(port=None) + assert device_adapter.controller_id == 1 # Because it is the only one who is both powered and low_energy + + # Test with given port (controller id) (valid) + controller_valid2 = [Controller(3, '44:55:66:11:22:33', '#3', settings={'powered': True, 'low_energy': True})] + mock_bable_no_ctrl.set_controllers(controller_valid + controller_not_valid + controller_valid2) + device_adapter = NativeBLEDeviceAdapter(port=3) + assert device_adapter.controller_id == 3 + + # Test with given port (controller id) (not valid) + with pytest.raises(ExternalError): + NativeBLEDeviceAdapter(port=4) + + # Test with wrong format port (controller id): must be an int + with pytest.raises(ValueError): + NativeBLEDeviceAdapter(port='hci0') + NativeBLEDeviceAdapter(port='hci0') + + +def test_scan(mock_bable): + """Test to start and stop scans.""" + # Flag to know if `on_scan` callback has been called + on_scan_called = [False] + + # Device information + iotile_id = 0x7777 + pending_data = False + low_voltage = False + user_connected = False + flags = pending_data | (low_voltage << 1) | (user_connected << 2) | (1 << 3) | (1 << 4) + advertisement = { + 'controller_id': 1, + 'type': 0x00, + 'address': '12:23:34:45:56:67', + 'address_type': 'public', + 'rssi': -60, + 'uuid': TileBusService.uuid, + 'company_id': ArchManuID, + 'device_name': 'Test', + 'manufacturer_data': struct.pack(" should fail) + device_adapter.connect_async(0, '{},{}'.format(device_address, device_address_type), on_connected) + assert mock_bable.counters['cancel_connection'] > 0 # Before connecting, we cancel all previous connections + assert mock_bable.counters['connect'] > 0 + assert device_address in controller_state['connecting'] # Verify that the device is in "connecting" state + + time.sleep(0.1) # Have to wait for connection manager to register the connection + mock_bable.simulate_connected_event(device_adapter.controller_id, device_address) # Simulate the connected event + assert device_address not in controller_state['connecting'] + + # On connect, device adapter should probe services and characteristics + connection_handle = list(controller_state['connected'])[0] + assert controller_state['connected'][connection_handle]['device'].address == device_address + assert mock_bable.counters['probe_services'] > 0 + assert mock_bable.counters['probe_characteristics'] == 0 + assert mock_bable.counters['disconnect'] == 1 # Because we do not have the TileBusService in our GATT table + + # Simulate the disconnected event to end the disconnection + time.sleep(0.1) + mock_bable.simulate_disconnected_event(device_adapter.controller_id, connection_handle) + + # Register the proper GATT table in our mock + services = [BLEService, TileBusService] + characteristics = [ + NameChar, + AppearanceChar, + ReceiveHeaderChar, + ReceivePayloadChar, + SendHeaderChar, + SendPayloadChar, + StreamingChar, + HighSpeedChar, + TracingChar + ] + mock_bable.set_gatt_table(services, characteristics) + + # Retry to connect + device_adapter.connect_async(0, '{},{}'.format(device_address, device_address_type), on_connected) + assert device_address in controller_state['connecting'] + + time.sleep(0.1) # Have to wait for connection manager to register the connection + mock_bable.simulate_connected_event(device_adapter.controller_id, device_address) + assert device_address not in controller_state['connecting'] + + connection_handle = list(controller_state['connected'])[0] + assert controller_state['connected'][connection_handle]['device'].address == device_address + assert mock_bable.counters['probe_services'] > 0 + assert mock_bable.counters['probe_characteristics'] > 0 + + # It should not disconnect now + assert mock_bable.counters['disconnect'] == 1 # Counter did not changed because now we have the TileBusService + + # Test if stopping with a connected device works + device_adapter.stop_sync() + assert mock_bable.counters['disconnect'] == 2 # Disconnection on stop + + +def test_connect_with_no_device(mock_bable): + """Test to connect to a non-existing device.""" + device_address = '11:11:11:11:11:11' + device_address_type = 'random' + + callback_called = threading.Event() + + def on_connected(connection_id, adapter_id, success, failure_reason): + """Callback function called when a device has been connected or connection failed.""" + assert success is False + callback_called.set() + + device_adapter = NativeBLEDeviceAdapter(port=1) + device_adapter.set_config('default_timeout', 0.2) # Set default timeout to not wait for 10 seconds + + # Try to connect + device_adapter.connect_async(0, '{},{}'.format(device_address, device_address_type), on_connected) + flag = callback_called.wait(timeout=device_adapter.get_config('default_timeout') + 1) + assert flag is True # Connection should have timed out after 0.2s and called the `on_connected` callback + + +def test_disconnect_not_connected_device(mock_bable): + """Test to disconnect while not connected.""" + callback_called = threading.Event() + + def on_disconnected(connection_id, adapter_id, success, failure_reason): + """Callback function called when the device has been disconnected or disconnection failed.""" + assert success is False + callback_called.set() + + device_adapter = NativeBLEDeviceAdapter(port=1) + device_adapter.set_config('default_timeout', 0.2) # Set default timeout to not wait for 10 seconds + + # Try to disconnect + device_adapter.disconnect_async(0, on_disconnected) + flag = callback_called.wait(timeout=device_adapter.get_config('default_timeout') + 1) + assert flag is True # Disconnection should have timed out after 0.2s and called the `on_disconnected` callback + + +def test_rpc(connected_device_adapter): + """Test to open RPC interface and send RPC.""" + device_adapter, mock_bable = connected_device_adapter + + # --- Opening RPC interface --- # + packets_sent = { + 'header': False, + 'payload': False + } + callback_called = threading.Event() + + def on_write_request__open_rpc_interface(request, *params): + """Callback function called when a write/write_without_response request has been sent to the mock_bable.""" + assert request['controller_id'] == 1 + assert request['attribute_handle'] in [ReceiveHeaderChar.config_handle, ReceivePayloadChar.config_handle] + assert request['value'] == b'\x00\x01' + + if request['attribute_handle'] == ReceiveHeaderChar.config_handle: + # Write request containing the header to open RPC interface + assert packets_sent['header'] is False + packets_sent['header'] = True + else: + # Write request containing the payload to open RPC interface + assert packets_sent['payload'] is False + packets_sent['payload'] = True + + def on_rpc_interface_opened(connection_id, adapter_id, success, *args): + """Callback function called when the RPC interface has been opened.""" + assert success is True + callback_called.set() + + # Register our on_write_request callback function in mock_bable + mock_bable.on_write_request(on_write_request__open_rpc_interface) + + # Open the RPC interface + device_adapter.open_interface_async(0, 'rpc', on_rpc_interface_opened) + callback_called.wait(timeout=5.0) + assert packets_sent['header'] is True + assert packets_sent['payload'] is True + + # --- Sending RPC --- # + packets_sent = { + 'header': False, + 'payload': False + } + callback_called.clear() + + def on_write_request__send_rpc(request, *params): + """Callback function called when a write/write_without_response request has been sent to the mock_bable.""" + assert request['controller_id'] == 1 + assert request['attribute_handle'] in [SendHeaderChar.value_handle, SendPayloadChar.value_handle] + + if request['attribute_handle'] == SendHeaderChar.value_handle: + # Write request containing the RPC header to send + assert packets_sent['header'] is False + packets_sent['header'] = True + else: + # Write request containing the RPC payload to send + assert packets_sent['payload'] is False + packets_sent['payload'] = True + + def on_rpc_response_received(connection_id, adapter_id, success, failure_reason, status, payload): + """Callback function called when an RPC response has been received.""" + assert connection_id == 0 + assert success is True + assert status == 0xFF + assert payload == b'' + callback_called.set() + + # Register our new on_write_request callback function in mock_bable + mock_bable.on_write_request(on_write_request__send_rpc) + + # Send the RPC request + device_adapter.send_rpc_async(0, 120, 0xFFFF, bytearray([]), 1.0, on_rpc_response_received) + + # Simulate the RPC response + mock_bable.notify( + connection_handle=1, + attribute_handle=ReceiveHeaderChar.value_handle, + value=b'\xFF\x00\x00\x00', + ) + flag = callback_called.wait(timeout=5.0) + assert flag is True + + +def test_streaming(connected_device_adapter): + """Test to open streaming interface and to receive reports.""" + device_adapter, mock_bable = connected_device_adapter + + # --- Opening streaming interface --- # + callback_called = threading.Event() + + def on_write_request__open_streaming_interface(request, *params): + """Callback function called when a write/write_without_response request has been sent to the mock_bable.""" + assert request['controller_id'] == 1 + assert request['attribute_handle'] == StreamingChar.config_handle + assert request['value'] == b'\x00\x01' + + def on_streaming_interface_opened(connection_id, adapter_id, success, *args): + """Callback function called when the streaming interface has been opened.""" + assert success is True + callback_called.set() + + # Register our on_write_request callback function in mock_bable + mock_bable.on_write_request(on_write_request__open_streaming_interface) + + # Open the streaming interface + device_adapter.open_interface_async(0, 'streaming', on_streaming_interface_opened) + flag = callback_called.wait(timeout=5.0) + assert flag is True + + # --- Streaming reports --- # + callback_called.clear() + reports = [] # Will contain the reports received + + def on_write_request__stream_reports(request, *params): + """Callback function called when a write/write_without_response request has been sent to the mock_bable.""" + assert request['controller_id'] == 1 + assert request['attribute_handle'] == StreamingChar.value_handle + + def on_report_callback(connection_id, report): + """Callback function called when a report has been processed.""" + assert connection_id == 0 + reports.append(report) + callback_called.set() + + # Register our report callback function to the `on_report` event + device_adapter.add_callback('on_report', on_report_callback) + + # Register our new on_write_request callback function in mock_bable + mock_bable.on_write_request(on_write_request__stream_reports) + + # Simulate a report sent by the device + mock_bable.notify( + connection_handle=1, + attribute_handle=StreamingChar.value_handle, + value=IndividualReadingReport.FromReadings(100, [IOTileReading(0, 1, 2)]).encode(), + ) + flag = callback_called.wait(timeout=5.0) + assert flag is True + + # We should have received exactly 1 report + assert len(reports) == 1 + + +def test_tracing(connected_device_adapter): + """Test to open tracing interface and to receive traces.""" + device_adapter, mock_bable = connected_device_adapter + + # --- Opening tracing interface --- # + callback_called = threading.Event() + + def on_write_request__open_tracing_interface(request, *params): + """Callback function called when a write/write_without_response request has been sent to the mock_bable.""" + assert request['controller_id'] == 1 + assert request['attribute_handle'] == TracingChar.config_handle + assert request['value'] == b'\x00\x01' + + def on_tracing_interface_opened(connection_id, adapter_id, success, *args): + """Callback function called when the tracing interface has been opened.""" + assert success is True + callback_called.set() + + # Register our on_write_request callback function in mock_bable + mock_bable.on_write_request(on_write_request__open_tracing_interface) + + # Open the tracing interface + device_adapter.open_interface_async(0, 'tracing', on_tracing_interface_opened) + flag = callback_called.wait(timeout=5.0) + assert flag is True + + # --- Receving traces --- # + callback_called.clear() + trace_chunks = [] + + def on_write_request__receive_traces(request, *params): + """Callback function called when a write/write_without_response request has been sent to the mock_bable.""" + assert request['controller_id'] == 1 + assert request['attribute_handle'] == TracingChar.value_handle + + def on_trace_received(connection_id, trace_chunk): + """Callback function called when a chunk trace has been received.""" + assert connection_id == 0 + trace_chunks.append(trace_chunk) + callback_called.set() + + # Register our trace received callback to the `on_trace` event + device_adapter.add_callback('on_trace', on_trace_received) + + # Register our new on_write_request callback function in mock_bable + mock_bable.on_write_request(on_write_request__receive_traces) + + # Simulate some traces + trace_sent = b'\x00\x11\x22\x33\x44\x55\x66\x88\x99\x00\x11\x22\x33\x44\x55\x66\x88\x99' # 20 bytes + mock_bable.notify( + connection_handle=1, + attribute_handle=TracingChar.value_handle, + value=trace_sent, + ) + flag = callback_called.wait(timeout=5.0) + assert flag is True + + assert len(trace_chunks) == 1 # We should have received only 1 chunk because we sent exactly 20 bytes + assert trace_chunks[0] == trace_sent # Verify that we received the trace sent + + +def test_send_script(connected_device_adapter): + """Test to open script interface and to send a script.""" + device_adapter, mock_bable = connected_device_adapter + + # --- Opening script interface --- # + callback_called = threading.Event() + + def on_write_request__open_script_interface(request, *params): + """Callback function called when a write/write_without_response request has been sent to the mock_bable.""" + assert request['controller_id'] == 1 + assert request['attribute_handle'] == HighSpeedChar.config_handle + assert request['value'] == b'\x00\x01' + + def on_script_interface_opened(connection_id, adapter_id, success, *args): + """Callback function called when the script interface has been opened.""" + assert success is True + callback_called.set() + + # Register our on_write_request callback function in mock_bable + mock_bable.on_write_request(on_write_request__open_script_interface) + + # Open the script interface + device_adapter.open_interface_async(0, 'script', on_script_interface_opened) + flag = callback_called.wait(timeout=5.0) + assert flag is True + + # --- Sending script --- # + callback_called.clear() + progress = {'done': 0, 'total': None} + + def on_write_request__send_script(request, *params): + """Callback function called when a write/write_without_response request has been sent to the mock_bable.""" + assert request['controller_id'] == 1 + assert request['attribute_handle'] == HighSpeedChar.value_handle + + def on_progress_callback(done_count, total_count): + """Callback function called when a script chunk has been sent to indicate progress.""" + if progress['total'] is not None: + assert progress['total'] == total_count + + assert done_count >= progress['done'] + assert done_count <= total_count + + progress['total'] = total_count + progress['done'] = done_count + + def on_script_sent(connection_id, adapter_id, success, failure_reason): + """Callback function called when all the script has been sent.""" + assert success is True + callback_called.set() + + # Register our new on_write_request callback function in mock_bable + mock_bable.on_write_request(on_write_request__send_script) + + # Send the script + script = bytes(b'ab')*100 + device_adapter.send_script_async(0, script, on_progress_callback, on_script_sent) + flag = callback_called.wait(timeout=5.0) + assert flag is True + assert progress['done'] == progress['total'] - 1 + + +def test_broadcast(mock_bable): + """Test to broadcast a report in scan response.""" + # Create some reading + broadcast_reading = IOTileReading(1, 0x1000, 100) + + # Device information + iotile_id = 0x7777 + pending_data = False + low_voltage = False + user_connected = False + flags = pending_data | (low_voltage << 1) | (user_connected << 2) | (1 << 3) | (1 << 4) + advertisement = { + 'controller_id': 1, + 'type': 0x00, + 'address': '12:23:34:45:56:67', + 'address_type': 'public', + 'rssi': -60, + 'uuid': TileBusService.uuid, + 'company_id': ArchManuID, + 'device_name': 'Test', + 'manufacturer_data': struct.pack("> 8) & 0xFF, address]) + mock_bable.write_without_response(connection_handle, SendHeaderChar.value_handle, header) + flag = response_received.wait(timeout=5.0) + assert flag is True + + +@pytest.mark.parametrize('virtual_interface', [{'device': build_report_device(), 'port': 1}], indirect=True) +def test_stream(connected_virtual_interface): + """Test to stream reports after streaming interface has been opened.""" + # Get the device configuration from the configuration file. + with open(get_report_device_string().split('@')[-1], "rb") as conf_file: + config = json.load(conf_file) + num_reports = int(config['device']['num_readings']) / int(config['device']['report_length']) + + mock_bable, interface = connected_virtual_interface + connection_handle = interface._connection_handle + + reports_received = threading.Event() + reports = [] + + def on_report(report, connection_id): + """Callback function called when a report has been processed.""" + reports.append(report) + if len(reports) == num_reports: # If all the reports have been received, we are done + reports_received.set() + + # Create a report parser and register our on_report callback + parser = IOTileReportParser(report_callback=on_report) + + def on_notification_received(success, result, failure_reason): + """Callback function called when a notification has been received (the virtual interface sends a notification + to stream reports)""" + assert success is True + assert result['controller_id'] == interface.controller_id + assert result['connection_handle'] == connection_handle + assert result['attribute_handle'] == StreamingChar.value_handle + + parser.add_data(result['value']) # Add the received report chunk to the report parser + + # Register our notification received callback into the mock_bable. + controller_state = mock_bable.controllers_state[interface.controller_id] + controller_state['connected'][connection_handle]['on_notification_received'] = (on_notification_received, []) + + # Open the streaming interface + assert interface.streaming is False + mock_bable.write_without_response(connection_handle, StreamingChar.config_handle, b'\x01\x00') + assert interface.streaming is True + + flag = reports_received.wait(timeout=5.0) + assert flag is True + + +@pytest.mark.parametrize('virtual_interface', [{'device': build_tracing_device(), 'port': 1}], indirect=True) +def test_tracing(connected_virtual_interface): + """Test to send traces after the tracing interface has been opened.""" + # Get the device configuration from the configuration file. + with open(get_tracing_device_string().split('@')[-1], "rb") as conf_file: + config = json.load(conf_file) + traces_sent = config['device']['ascii_data'] + + mock_bable, interface = connected_virtual_interface + connection_handle = interface._connection_handle + + all_received = threading.Event() + traces_received = [bytearray()] + + def on_notification_received(success, result, failure_reason): + """Callback function called when a notification has been received (the virtual interface sends a notification + to send traces)""" + assert success is True + assert result['controller_id'] == interface.controller_id + assert result['connection_handle'] == connection_handle + assert result['attribute_handle'] == TracingChar.value_handle + + traces_received[0] += result['value'] # Collect the traces + if len(traces_received[0]) == len(traces_sent): # If we have received all the traces, we are done + all_received.set() + + # Register our notification received callback into the mock_bable. + controller_state = mock_bable.controllers_state[interface.controller_id] + controller_state['connected'][connection_handle]['on_notification_received'] = (on_notification_received, []) + + # Open the tracing interface + assert interface.tracing is False + mock_bable.write_without_response(connection_handle, TracingChar.config_handle, b'\x01\x00') + assert interface.tracing is True + + flag = all_received.wait(timeout=5.0) + assert flag is True + assert traces_received[0].decode() == traces_sent # Verify that we received the right traces diff --git a/transport_plugins/native_ble/tests/tracing_device_config.json b/transport_plugins/native_ble/tests/tracing_device_config.json new file mode 100644 index 000000000..6883c4a65 --- /dev/null +++ b/transport_plugins/native_ble/tests/tracing_device_config.json @@ -0,0 +1,6 @@ +{ + "device": { + "iotile_id": "0x11", + "ascii_data": "Hello world! How are you? Lorem ipsum dolor sit amet, consectetur adipiscing elit. Integer lobortis egestas mollis. Donec a sapien sed quam pulvinar faucibus. Nullam sed finibus urna. Quisque vehicula eros sit amet lacus rutrum. Duis non laoreet est. Fusce gravida suscipit quam nec lacinia. Aliquam eget condimentum orci. Aliquam at convallis ipsum. Nulla at lacus quis metus luctus laoreet id vel est. Good bye!" + } +} diff --git a/transport_plugins/native_ble/tox.ini b/transport_plugins/native_ble/tox.ini new file mode 100644 index 000000000..954b06601 --- /dev/null +++ b/transport_plugins/native_ble/tox.ini @@ -0,0 +1,14 @@ +[tox] +envlist = py{27,36}-{linux_only} + +[testenv] +passenv=APPDATA + +platform = mac_windows: darwin|win32 + linux_only: linux|linux2 + +deps= + pytest + ../../iotilecore + ../../iotiletest +commands=py.test {posargs} diff --git a/transport_plugins/native_ble/version.py b/transport_plugins/native_ble/version.py new file mode 100644 index 000000000..11a716ec1 --- /dev/null +++ b/transport_plugins/native_ble/version.py @@ -0,0 +1 @@ +version = "1.0.0"