Receive device GATT notification

Receive device Notification/Indication data through the API.

Refer Receive Notification and Indication.

Example

import asyncio
import cassiablue


async def notifier():
    print("start recv notification data")
    ok, _ = await cassiablue.start_recv_notify()
    print(type(ok), ok)
    if not ok:
        return

    async for evt in cassiablue.notify_result():
        print(type(evt), evt)


async def connector():
    """
    The handle and value in the examples relate to the device’s GATT service;
    refer to the device-specific implementation protocol for details.
    """
    print("start connect")

    params = '{"type": "random"}'
    addr = "20:00:00:00:00:01"
    ok, ret = await cassiablue.connect(addr, params)
    print("connect done:", type(ok), ok, type(ret), ret)

    await asyncio.sleep(3)

    # open notification
    handle = "45"
    value = "0100"
    ok, ret = await cassiablue.gatt_write(addr, handle, value)
    print("write done:", type(ok), ok, type(ret), ret)

    # write cmd trigger notification
    handle = "42"
    value = "20020001"
    ok, ret = await cassiablue.gatt_write(addr, handle, value)
    print("write done:", type(ok), ok, type(ret), ret)

    await asyncio.sleep(5)
    print("sleep ok")

    await cassiablue.disconnect(addr)
    print("disconnect ok")


async def main():
    print("start")
    await asyncio.gather(notifier(), connector())


print("hello")
asyncio.run(main())

API

async cassiablue.start_recv_notify() → tuple[ok: bool, result: str]

Enable receiving device GATT notification data. Currently, only one subscription can be active at a time.

Returns:

• ok: Whether the operation was successful

• result: Execution result. For notification data, refer to BLENotifyResult

async cassiablue.notify_result() → BLENotifyResult

Receive Device GATT Notification Data

Returns:

class BLENotifyResult

class BLENotifyResult

An asynchronous iterable object used to receive BLE notification data in a streaming manner.

async __aiter__() → BLENotifyResult

Returns itself, allowing the instance to be used directly in an async for loop.

async __anext__() → dict

Asynchronously retrieves the next notification data from the global notify_sse_client.queue and returns it.

If the queue is empty, it waits until new data becomes available.

Data Field Description

Field	Type	Description	Example
id	str	Device MAC	20:00:00:00:00:01
dataType	str	Data Type	notification indication
handle	int	Handle	44
value	str	Value	A0EE00…