serverble.py

import sys, os, socket, string, random
import dbus, dbus.mainloop.glib
from gi.repository import GLib
import subprocess
from example_advertisement import Advertisement
from example_advertisement import register_ad_cb, register_ad_error_cb
from example_gatt_server import Service, Characteristic
from example_gatt_server import register_app_cb, register_app_error_cb
 
BLUEZ_SERVICE_NAME =           'org.bluez'
DBUS_OM_IFACE =                'org.freedesktop.DBus.ObjectManager'
LE_ADVERTISING_MANAGER_IFACE = 'org.bluez.LEAdvertisingManager1'
GATT_MANAGER_IFACE =           'org.bluez.GattManager1'
GATT_CHRC_IFACE =              'org.bluez.GattCharacteristic1'
UART_SERVICE_UUID =            '6e400001-b5a3-f393-e0a9-e50e24dcca9e'
UART_RX_CHARACTERISTIC_UUID =  '6e400002-b5a3-f393-e0a9-e50e24dcca9e'
UART_TX_CHARACTERISTIC_UUID =  '6e400003-b5a3-f393-e0a9-e50e24dcca9e'
mainloop = None
 
# class TxCharacteristic(Characteristic):
#     def __init__(self, bus, index, service):
#         Characteristic.__init__(self, bus, index, UART_TX_CHARACTERISTIC_UUID,
#                                 ['notify'], service)
#         self.notifying = False
#         GLib.io_add_watch(sys.stdin, GLib.IO_IN, self.on_console_input)
 
#     def on_console_input(self, fd, condition):
#         s = fd.readline()
#         if s.isspace():
#             pass
#         else:
#             self.send_tx(s)
#         return True
 
#     def send_tx(self, s):
#         if not self.notifying:
#             return
#         value = []
#         for c in s:
#             value.append(dbus.Byte(c.encode()))
#         self.PropertiesChanged(GATT_CHRC_IFACE, {'Value': value}, [])
 
#     def StartNotify(self):
#         if self.notifying:
#             return
#         self.notifying = True
 
#     def StopNotify(self):
#         if not self.notifying:
#             return
#         self.notifying = False
 
class RxCharacteristic(Characteristic):
    result = None
    def __init__(self, bus, index, service):
        Characteristic.__init__(self, bus, index, UART_RX_CHARACTERISTIC_UUID,
                                ['write', 'read'], service)
 
 # Sends message to device
    def WriteValue(self, value, options):
        print('remote: {}'.format(bytearray(value).decode()))
        inData = bytearray(value).decode()
        self.result = subprocess.run(inData.split(),capture_output=True)


    def ReadValue(self, options):
        response = ''
        if self.result:
            response = self.result.stdout
            print(response)
        
        return response
    
        # return response
 
class UartService(Service):
    def __init__(self, bus, index):
        Service.__init__(self, bus, index, UART_SERVICE_UUID, True)
        # self.add_characteristic(TxCharacteristic(bus, 0, self))
        self.add_characteristic(RxCharacteristic(bus, 0, self))
 
class Application(dbus.service.Object):
    def __init__(self, bus):
        self.path = '/'
        self.services = []
        dbus.service.Object.__init__(self, bus, self.path)
 
    def get_path(self):
        return dbus.ObjectPath(self.path)
 
    def add_service(self, service):
        self.services.append(service)
 
    @dbus.service.method(DBUS_OM_IFACE, out_signature='a{oa{sa{sv}}}')
    def GetManagedObjects(self):
        response = {}
        for service in self.services:
            response[service.get_path()] = service.get_properties()
            chrcs = service.get_characteristics()
            for chrc in chrcs:
                response[chrc.get_path()] = chrc.get_properties()
        return response
 
class UartApplication(Application):
    def __init__(self, bus):
        Application.__init__(self, bus)
        self.add_service(UartService(bus, 0))

class UartAdvertisement(Advertisement):
    def __init__(self, bus, index):
        Advertisement.__init__(self, bus, index, 'peripheral')
        self.add_service_uuid(UART_SERVICE_UUID)
        self.add_local_name(self.get_host_name())
        self.include_tx_power = True

    def get_host_name(self):
        bt_device_number = ''.join(random.sample((string.digits), 4))
        f = open("/etc/bluetooth-id", "w")
        f.write(bt_device_number)
        f.close()  
        return "%s-%s" % (socket.gethostname(), bt_device_number)
 
def find_adapter(bus):
    remote_om = dbus.Interface(bus.get_object(BLUEZ_SERVICE_NAME, '/'),
                               DBUS_OM_IFACE)
    objects = remote_om.GetManagedObjects()
    for o, props in objects.items():
        if LE_ADVERTISING_MANAGER_IFACE in props and GATT_MANAGER_IFACE in props:
            return o
        print('Skip adapter:', o)
    return None

def main():
    global mainloop
    dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)
    bus = dbus.SystemBus()
    adapter = find_adapter(bus)
    if not adapter:
        print('BLE adapter not found')
        return
 
    service_manager = dbus.Interface(
                                bus.get_object(BLUEZ_SERVICE_NAME, adapter),
                                GATT_MANAGER_IFACE)
    ad_manager = dbus.Interface(bus.get_object(BLUEZ_SERVICE_NAME, adapter),
                                LE_ADVERTISING_MANAGER_IFACE)
 
    app = UartApplication(bus)
    adv = UartAdvertisement(bus, 0)
 
    mainloop = GLib.MainLoop()
 
    service_manager.RegisterApplication(app.get_path(), {},
                                        reply_handler=register_app_cb,
                                        error_handler=register_app_error_cb)
    ad_manager.RegisterAdvertisement(adv.get_path(), {},
                                     reply_handler=register_ad_cb,
                                     error_handler=register_ad_error_cb)
    try:
        mainloop.run()
    except KeyboardInterrupt:
        adv.Release()
 
if __name__ == '__main__':
    main()