gatt_test.py

import gatt
import threading

manager = gatt.DeviceManager(adapter_name='hci0')

class AnyDevice(gatt.Device):
    def services_resolved(self):
        super().services_resolved()

        device_proprietary_service = next(
            s for s in self.services
            if s.uuid == 'f815e810-456c-6761-746f-4d756e696368')

        color_characteristic = next(
            c for c in device_proprietary_service.characteristics
            if c.uuid == 'f815e811-456c-6761-746f-4d756e696368')

        color_characteristic.write_value(bytearray(b'\x35\xF4\x01\x0A\x00\x00\x80\x00\x30\x00\x20\x00\x10'))


    def characteristic_value_updated(self, characteristic, value):
        print(str(characteristic)+", value: "+value.decode("utf-8"))

    def characteristic_write_value_succeeded(self, characteristic):
        print("Write successful.")

    def characteristic_write_value_failed(self, characteristic, error):
        print("Write failed. "+str(error))


class myThread (threading.Thread):
   def __init__(self, manager):
      threading.Thread.__init__(self)
      self.manager=manager
   def run(self):
      self.manager.run()
   def quit(self):
      self.manager.stop()  

device = AnyDevice(mac_address='78:A5:04:8E:12:44', manager=manager)
device.connect()


manager_thread = myThread(manager)
manager_thread.daemon = True
manager_thread.start()


print("Started Thread.")

try:
    while True:
        pass
except KeyboardInterrupt:
    print('KeyboardInterrupt!')
    manager_thread.quit()
    manager.stop()
    manager_thread.join(1)