Creating a Service¶
Attribute Flags¶
The behavior of a particular attribute is described by a set of flags. These flags are implemented using the CharacteristicFlags and DescriptorFlags enums. A single attribute may have multiple flags, in python you can combine these flags using the | operator (eg. CharacteristicFlags.READ | CharacteristicFlags.WRITE).
UUIDs¶
Hint
The Bluetooth SIG has reserved 16-bit UUIDs for standardised services. 128-bit UUIDs should be preferred to avoid conflicts and confusion.
BLE uses 128-bit Universally Unique Identifiers (UUIDs) to determine what each service, characteristic and descriptor refers to in addition to the type of every attribute. To minimize the amount of information that needs to be transmitted the Bluetooth SIG selected a base UUID of 0000XXXX-0000-1000-8000-00805F9B34FB. This allows a 16-bit number to be transmitted in place of the full 128-bit value in some cases. In bluez_peripheral 16-bit UUIDs are represented by the UUID16 class whilst 128-bit values are represented by uuid.UUID. In bluez_peripheral all user provided UUIDs are are parsed using UUID16.parse_uuid() meaning you can use these types interchangeably, UUID16s will automatically be used where possible.
Adding Attributes¶
The @characteristic and @descriptor decorators are designed to work identically to the built-in @property decorator. Attributes can be added to a service either manually or using decorators:
Warning
Attributes exceeding 48 bytes in length may take place across multiple accesses, using the options.offset parameter to select portions of the data. This is dependent upon the options.mtu.
from bluez_peripheral.gatt import Service
from bluez_peripheral.gatt import characteristic, CharacteristicFlags as CharFlags
from bluez_peripheral.gatt import descriptor, DescriptorFlags as DescFlags
class MyService(Service):
def __init__(self):
# You must call the super class constructor to register any decorated attributes.
super().__init__(uuid="BEED")
@characteristic("BEEE", CharFlags.READ | CharFlags.WRITE)
def my_characteristic(self, options):
# This is the getter for my_characteristic.
# All attribute functions must return bytes.
return bytes("Hello World!", "utf-8")
@my_characteristic.setter
def my_characteristic(self, value, options):
# This is the setter for my_characteristic.
# Value consists of some bytes.
self._my_char_value = value
# Descriptors work exactly the same way.
@descriptor("BEEF", my_characteristic, DescFlags.WRITE)
def my_writeonly_descriptor(self, options):
# This function is a manditory placeholder.
# In Python 3.9+ you don't need this function (See PEP 614).
pass
my_writeonly_descriptor.setter
def my_writeonly_descriptor(self, value, options):
self._my_desc_value = value
# Characteristic and Descriptor getters/ setters may also be asynchronous.
@characteristic("BEEB", CharFlags.READ)
async def my_async_characteristic(self, options):
return await my_awaitable()
from bluez_peripheral.gatt import Service
from bluez_peripheral.gatt import characteristic, CharacteristicFlags as CharFlags
from bluez_peripheral.gatt import descriptor, DescriptorFlags as DescFlags
# Create my_characteristic
my_char_value = None
def my_characteristic_getter(service, options):
return bytes("Hello World!", "utf-8")
def my_characteristic_setter(service, value, options):
my_char_value = value
# See characteristic.__call__()
my_characteristic = characteristic("BEEE", CharFlags.READ | CharFlags.WRITE)(
my_characteristic_getter, my_characteristic_setter
)
# Create my_descriptor
my_desc_value = None
def my_readonly_descriptor_setter(service, value, options):
my_desc_value = value
# See descriptor.__call__()
my_descriptor = descriptor("BEEF", my_characteristic, DescFlags.WRITE)(
None, my_readonly_descriptor_setter
)
async def my_async_characteristic_getter(self, options):
return await my_awaitable()
my_async_characteristic = characteristic("BEEE", CharFlags.READ)(
my_async_characteristic_getter
)
# Register my_descriptor with its parent characteristic and my_characteristic
# with its parent service.
my_service = Service("BEED")
my_characteristic.add_descriptor(my_descriptor)
my_service.add_characteristic(my_characteristic)
my_service.add_characteristic(my_async_characteristic)
Error Handling¶
Attribute getters/ setters may raise one of a set of legal exceptions to signal specific conditions to bluez. Avoid throwing custom exceptions in attribute assessors, since these will not be presented to a user and bluez will not know how to interpret them. Additionally any exceptions thrown must derive from dbus_fast.DBusError.
Legal Errors¶
Error |
Characteristic |
Descriptor |
||
|---|---|---|---|---|
Getter |
Setter |
Getter |
Setter |
|
✓ |
✓ |
✓ |
✓ |
|
✓ |
✓ |
✓ |
✓ |
|
✓ |
||||
✓ |
✓ |
|||
✓ |
✓ |
✓ |
✓ |
|
✓ |
✓ |
✓ |
✓ |
|
✓ |
✓ |
✓ |
✓ |
|
Registering a Service¶
Warning
Ensure that the thread used to register your service yields regularly. Client requests will not be served otherwise.
Hint
The “message bus” referred to here is a dbus_fast.aio.MessageBus.
Services can either be registered individually using a Service or as part of a ServiceCollection. For example following on from the earlier code:
from bluez_peripheral import get_message_bus
async def main():
my_service = Service()
bus = await get_message_bus()
# Register the service for bluez to access.
await my_service.register(bus)
# Yeild so that the service can handle requests.
await bus.wait_for_disconnect()
if __name__ == "__main__":
asyncio.run(main())
from bluez_peripheral import get_message_bus
from bluez_peripheral.gatt import ServiceCollection
async def main():
my_service_collection = ServiceCollection()
my_service_collection.add_service(my_service)
#my_service_collection.add_service(my_other_service)
bus = await get_message_bus()
# Register the service for bluez to access.
await my_service_collection.register(bus)
# Yeild so that the services can handle requests.
await bus.wait_for_disconnect()
if __name__ == "__main__":
asyncio.run(main())
Notification¶
Characteristics with the NOTIFY or INDICATE flags can update clients when their value changes. Indicate requires acknowledgment from the client whilst notify does not. For this to work the client must first call subscribe to the notification. The client can then be notified by calling characteristic.changed().
Warning
The characteristic.changed() function may only be called in the same thread that registered the service.
from bluez_peripheral import get_message_bus
from bluez_peripheral.gatt import Service
from bluez_peripheral.gatt import characteristic, CharacteristicFlags as CharFlags
class MyService(Service):
def __init__(self):
super().__init__(uuid="DEED")
@characteristic("DEEE", CharFlags.NOTIFY)
def my_notify_characteristic(self, options):
pass
async def main():
my_service = MyService()
bus = await get_message_bus()
await my_service.register(bus)
# Signal that the value of the characteristic has changed.
service.my_notify_characteristic.changed(bytes("My new value", "utf-8"))
# Yeild so that the service can handle requests and signal the change.
await bus.wait_for_disconnect()
if __name__ == "__main__":
asyncio.run(main())
See also
- Bluez Documentation
- Attribute Access Options
CharacteristicReadOptionsCharacteristicWriteOptionsDescriptorReadOptionsDescriptorWriteOptions