How can I subscribe a characteristic?

[Squirre17]

I want to subscribe a characteristic like following:

Characteristic(handle=0x003C, uuid=0000AAAA-0000-2222-8888-00805F9B1111, READ|NOTIFY|INDICATE)

I want to handle data which sent by remote gatt server through this characteristic.

[zxzxwu]

You should be able to discover characteristics, find the one you want and subscribe them.

https://github.com/google/bumble/blob/main/examples/run_gatt_client.py

https://github.com/google/bumble/blob/main/bumble/gatt_client.py#L232

https://github.com/google/bumble/blob/main/examples/battery_client.py

[zxzxwu]

Or you may use console app to subscribe everything without any code.

[Squirre17]

You should be able to discover characteristics, find the one you want and subscribe them.

https://github.com/google/bumble/blob/main/examples/run_gatt_client.py

https://github.com/google/bumble/blob/main/bumble/gatt_client.py#L232

https://github.com/google/bumble/blob/main/examples/battery_client.py

thx, i found my way to subscribe the characteristic:

connection.gatt_client.notification_subscribers[handle] = set([recv])

[barbibulle]

Directly accessing gatt_client.notification_subscribers is not recommended.
Subscribing to a characteristic should be done by using the .subscribe method of a CharactertisticProxy object.
See examples/battery_client.py for an example.

[Squirre17]

Directly accessing gatt_client.notification_subscribers is not recommended. Subscribing to a characteristic should be done by using the .subscribe method of a CharactertisticProxy object. See examples/battery_client.py for an example.

modified version according to your advise:

import asyncio
from bumble.device import Connection
from bumble.gatt_client import CharacteristicProxy

class AsyncSpscFifo:

    def __init__(self):
        self.que = asyncio.Queue() # type: ignore[var-annotated]
        self.connection = None

    @staticmethod
    async def create(connection: Connection, chara: CharacteristicProxy):

        self = AsyncSpscFifo()
        # self.que = asyncio.Queue() # type: ignore[var-annotated]
        self.connection = connection
        
        def recv(data: bytes):
            self.que.put_nowait(data)
            
        # connection.gatt_client.notification_subscribers[chara.handle] = set([recv])
        await connection.gatt_client.subscribe(
            chara, recv
        )
        return self

    async def get(self):
        return await self.que.get()