Added BLE terminal

examples/read_firmware_version.py

@@ -29,4 +29,5 @@ def characteristic_value_updated(self, characteristic, value):
 device = AnyDevice(manager=manager, mac_address=args.mac_address)
 device.connect()
 
+
 manager.run()

gatt/gatt_linux.py

@@ -63,6 +63,7 @@ def run(self):
         This call blocks until you call `stop()` to stop the main loop.
         """
 
+
         if self._main_loop:
             return
 
@@ -86,7 +87,7 @@ def disconnect_signals():
             self._properties_changed_signal.remove()
             self._interface_added_signal.remove()
 
-        self._main_loop = GObject.MainLoop()
+        self._main_loop = GObject.MainLoop.new(None, False)  # modification to launch run in a thread
         try:
             self._main_loop.run()
             disconnect_signals()
@@ -605,6 +606,7 @@ def write_value(self, value, offset=0):
         :param value: array of bytes to be written
         :param offset: offset from where to start writing the bytes (defaults to 0)
         """
+        print("Sent : ", value)
         bytes = [dbus.Byte(b) for b in value]
 
         try:
@@ -640,6 +642,7 @@ def enable_notifications(self, enabled=True):
         Each time when the device notifies a new value, `characteristic_value_updated()` of the related
         device will be called.
         """
+        print("enable_notifications called")
         try:
             if enabled:
                 self._object.StartNotify(

gattctl.py

@@ -1,7 +1,12 @@
 #!/usr/bin/env python3
 
 from argparse import ArgumentParser
+from threading import Thread
+import keyboard
+from termios import tcflush, TCIFLUSH
 import gatt
+import time
+import sys
 
 device_manager = None
 
@@ -11,13 +16,28 @@ class AnyDeviceManager(gatt.DeviceManager):
     An implementation of ``gatt.DeviceManager`` that discovers any GATT device
     and prints all discovered devices.
     """
+    def __init__(self, adapter_name):
+        super().__init__(adapter_name)
+        self.nb_device_connected = 0
+        self.discover_filter = 'True'
 
     def device_discovered(self, device):
-        print("[%s] Discovered, alias = %s" % (device.mac_address, device.alias()))
+        if self.discover_filter == 'True':
+            print("[%s] Discovered, alias = %s" % (device.mac_address, device.alias()))
+        elif self.discover_filter in device.alias() or self.discover_filter in device.mac_address:
+            print("[%s] Discovered, alias = %s" % (device.mac_address, device.alias()))
 
     def make_device(self, mac_address):
         return AnyDevice(mac_address=mac_address, manager=self)
 
+    def run(self):
+        print("Running Manager")
+        super().run()
+
+    def quit(self):
+        print("Stopping Manager")
+        super().stop()
+
 
 class AnyDevice(gatt.Device):
     """
@@ -28,20 +48,26 @@ class AnyDevice(gatt.Device):
     def __init__(self, mac_address, manager, auto_reconnect=False):
         super().__init__(mac_address=mac_address, manager=manager)
         self.auto_reconnect = auto_reconnect
+        self.message = ''
+        self.transmition = ""
+        self.reception = ""
 
     def connect(self):
         print("Connecting...")
         super().connect()
 
     def connect_succeeded(self):
         super().connect_succeeded()
+        device_manager.nb_device_connected += 1
         print("[%s] Connected" % (self.mac_address))
 
     def connect_failed(self, error):
         super().connect_failed(error)
         print("[%s] Connection failed: %s" % (self.mac_address, str(error)))
 
     def disconnect_succeeded(self):
+        if device_manager.nb_device_connected:
+            device_manager.nb_device_connected -= 1
         super().disconnect_succeeded()
 
         print("[%s] Disconnected" % (self.mac_address))
@@ -55,7 +81,48 @@ def services_resolved(self):
         for service in self.services:
             print("[%s]  Service [%s]" % (self.mac_address, service.uuid))
             for characteristic in service.characteristics:
+                if characteristic.uuid == '6e400003-b5a3-f393-e0a9-e50e24dcca9e':
+                    characteristic.enable_notifications()
+
                 print("[%s]    Characteristic [%s]" % (self.mac_address, characteristic.uuid))
+        print("[%s] Resolved services" % (self.mac_address))
+
+        device_service = next(
+            s for s in self.services
+            if s.uuid == '6e400001-b5a3-f393-e0a9-e50e24dcca9e')
+
+        self.reception = next(
+            c for c in device_service.characteristics
+            if c.uuid == '6e400003-b5a3-f393-e0a9-e50e24dcca9e')
+
+        self.transmition = next(
+            c for c in device_service.characteristics
+            if c.uuid == '6e400002-b5a3-f393-e0a9-e50e24dcca9e')
+
+        if self.message:
+            bar = bytearray(self.message.encode())
+            self.transmition.write_value(bar)
+
+        time.sleep(2)
+        self.reception.read_value()
+
+    def characteristic_value_updated(self, characteristic, value):
+        print("RX:", (value.decode("utf-8")).replace('\n', ''))
+
+    def characteristic_read_value_failed(self, characteristic, error):
+        print("RX fail : ", error)
+
+    def characteristic_enable_notifications_succeeded(self, characteristic):
+        print("enable notification succeeded for characteristic ", characteristic.uuid)
+
+    def characteristic_enable_notifications_failed(self, characteristic, error):
+        print("enable notification succeeded for characteristic ", characteristic.uuid, ", error, ", error)
+
+    def characteristic_write_value_succeeded(self, characteristic):
+        print("write value succeeded for characteristic ", characteristic.uuid)
+
+    def characteristic_write_value_failed(self, characteristic, error):
+        print("write value fail for characteristic ", characteristic.uuid, ", error, ", error)
 
 
 def main():
@@ -64,6 +131,11 @@ def main():
         '--adapter',
         default='hci0',
         help="Name of Bluetooth adapter, defaults to 'hci0'")
+    arg_parser.add_argument(
+        '--message',
+        metavar='string',
+        type=str,
+        help="String send by ble, usable with --connect only. All 'n' of the message will be remplace by '\ n' ")
     arg_commands_group = arg_parser.add_mutually_exclusive_group(required=True)
     arg_commands_group.add_argument(
         '--power-on',
@@ -81,6 +153,11 @@ def main():
         '--discover',
         action='store_true',
         help="Lists all nearby GATT devices")
+    arg_commands_group.add_argument(
+        '--discover-filter',
+        metavar='filter',
+        type=str,
+        help="Lists all nearby GATT devices containing filter in their MAC address or alias")
     arg_commands_group.add_argument(
         '--connect',
         metavar='address',
@@ -114,8 +191,14 @@ def main():
         return
     if args.discover:
         device_manager.start_discovery()
+    if args.discover_filter:
+        print("discover filter : ", args.discover_filter)
+        device_manager.discover_filter = args.discover_filter
+        device_manager.start_discovery()
     elif args.connect:
         device = AnyDevice(mac_address=args.connect, manager=device_manager)
+        if args.message:
+            device.message = (args.message.replace('%n','\n'))
         device.connect()
     elif args.auto:
         device = AnyDevice(mac_address=args.auto, manager=device_manager, auto_reconnect=True)
@@ -125,12 +208,28 @@ def main():
         device.disconnect()
         return
 
+    thread = Thread(target=device_manager.run)
+    thread.start()
+
     print("Terminate with Ctrl+C")
+    print("Started Thread.")
+
     try:
-        device_manager.run()
+        while True:
+            if device_manager.nb_device_connected:
+                # If ESCAPE key is pressed you can start to write the message, ENTER to send it to the device
+                if keyboard.is_pressed(chr(27)):
+                    tcflush(sys.stdin, TCIFLUSH)
+                    message = bytearray(input(">> ").replace('%n','\n').encode())
+                    device.transmition.write_value(message)
+
+            pass
     except KeyboardInterrupt:
-        pass
-
+        print('KeyboardInterrupt!')
+        if device_manager.nb_device_connected:
+            device.disconnect()
+        device_manager.quit()
+    thread.join()
 
 if __name__ == '__main__':
     main()