`aioble` — BLE asíncrono¶

aioble es un envoltorio de alto nivel compatible con asyncio alrededor del módulo bluetooth. Proporciona corrutinas limpias para escanear, conectar, anunciar, servicios GATT y canales L2CAP.

Todas las operaciones remotas (conectar, desconectar, lectura/escritura del cliente, indicación del servidor, recepción/envío de l2cap, emparejamiento) son awaitable y admiten tiempos de espera.

Roles admitidos:

Broadcaster (anunciante) — genera cargas útiles de anuncio y de respuesta de escaneo para campos comunes, divide automáticamente la carga útil entre el anuncio y la respuesta de escaneo, anuncia indefinidamente o durante una duración fija.
Peripheral — espera la conexión de un central, espera el intercambio de MTU.
Observer (escáner) — escaneo pasivo y activo, combina las cargas útiles de anuncio y de respuesta de escaneo del mismo dispositivo, analiza los campos comunes de las cargas útiles de anuncio.
Central — se conecta a un periférico, inicia el intercambio de MTU.
Cliente GATT — descubre servicios / características / descriptores (opcionalmente por UUID); lee / escribe / escribe-con-respuesta en características y descriptores; se suscribe a notificaciones e indicaciones (a través del CCCD); espera notificaciones e indicaciones.
Servidor GATT — registra servicios / características / descriptores; espera escrituras en características y descriptores; intercepta solicitudes de lectura; envía notificaciones e indicaciones (y espera la respuesta).
L2CAP — acepta y conecta canales L2CAP orientados a conexión, gestiona el control de flujo del canal.
Seguridad — gestión de claves/secretos respaldada por JSON, inicia el emparejamiento, consulta el estado de cifrado / autenticación.

Ejemplos¶

Escanea dispositivos BLE cercanos e imprime cada uno a medida que se detecta:

import aioble
import asyncio

async def find_devices():
    async with aioble.scan(duration_ms=5000, active=True) as scanner:
        async for result in scanner:
            print(result.device.addr_hex(), result.rssi, result.name())

asyncio.run(find_devices())

Conéctate como central a un periférico que anuncia el servicio Heart Rate y suscríbete a sus notificaciones de medición:

import aioble
import asyncio
import bluetooth

_HR_SERVICE = bluetooth.UUID(0x180D)
_HR_MEASUREMENT = bluetooth.UUID(0x2A37)

async def connect_and_read():
    device = None
    async with aioble.scan(duration_ms=5000, active=True) as scanner:
        async for result in scanner:
            if _HR_SERVICE in result.services():
                device = result.device
                break
    if device is None:
        return

    async with await device.connect() as conn:
        service = await conn.service(_HR_SERVICE)
        char = await service.characteristic(_HR_MEASUREMENT)
        await char.subscribe(notify=True)
        while True:
            data = await char.notified()
            print("notify:", data)

asyncio.run(connect_and_read())

Actúa como peripheral: registra un servicio GATT, anúncialo y envía notificaciones a quien se conecte:

import aioble
import asyncio
import bluetooth
import struct

_ENV_SERVICE = bluetooth.UUID(0x181A)
_TEMP_CHAR = bluetooth.UUID(0x2A6E)

def encode_temperature(deg_c):
    # Bluetooth Temperature (0x2A6E) is sint16 little-endian, 0.01 degC units.
    return struct.pack("<h", round(deg_c * 100))

service = aioble.Service(_ENV_SERVICE)
temp_char = aioble.Characteristic(service, _TEMP_CHAR, read=True, notify=True)
aioble.register_services(service)

async def peripheral_task():
    while True:
        connection = await aioble.advertise(
            interval_us=250000,
            name="openmv-sensor",
            services=[_ENV_SERVICE],
            appearance=0x0300,
        )
        print("connected:", connection.device.addr_hex())
        async with connection:
            while connection.is_connected():
                temp_char.write(encode_temperature(23.68), send_update=True)
                await asyncio.sleep(1)

asyncio.run(peripheral_task())

Funciones a nivel de módulo¶

aioble.config(*args, **kwargs) → Any¶

Reenvía a bluetooth.BLE.config(), asegurándose primero de que la radio BLE esté activa.

args: Nombre de un único parámetro opcional a consultar.
kwargs: Argumentos de palabra clave para establecer valores de configuración.

aioble.stop() → None¶: Desactiva la radio BLE subyacente y ejecuta los manejadores de apagado de submódulos registrados. Después de llamar a esto, los escáneres, anunciantes, conexiones y canales L2CAP se desmontan todos.

aioble.scan(duration_ms: int, interval_us: int | None = None, window_us: int | None = None, active: bool = False) → scan¶

Devuelve un gestor de contexto asíncrono / iterador asíncrono scan que produce instancias de ScanResult por cada dispositivo único descubierto (o por cada nueva pieza de datos de anuncio de un dispositivo conocido).

duration_ms: Cuánto tiempo escanear, en milisegundos. Pasa 0 para escanear indefinidamente hasta que el gestor de contexto salga.
interval_us: Intervalo de escaneo en microsegundos. El valor predeterminado es 1.280.000.
window_us: Ventana de escaneo en microsegundos (debe ser menor o igual que interval_us). El valor predeterminado es 11.250.
active: Si es True, realiza un escaneo activo (solicita datos de respuesta de escaneo). El valor predeterminado es False.

aioble.advertise(interval_us: int, adv_data: bytes | None = None, resp_data: bytes | None = None, connectable: bool = True, limited_disc: bool = False, br_edr: bool = False, name: str | None = None, services: list | None = None, appearance: int = 0, manufacturer: tuple | None = None, timeout_ms: int | None = None) → DeviceConnection¶

Corrutina asíncrona que comienza a anunciar y espera una conexión central entrante. Devuelve una DeviceConnection que representa el central conectado, o lanza asyncio.TimeoutError si se agota el tiempo de espera.

interval_us: Intervalo de anuncio, en microsegundos.
adv_data: Carga útil de anuncio en bruto. Si no se establece, adv_data se construye a partir de los argumentos de palabra clave restantes.
resp_data: Carga útil de respuesta de escaneo en bruto. Se rellena automáticamente con el desbordamiento de adv_data si es necesario.
connectable: Si es True, este es un anuncio conectable.
limited_disc: Usa la bandera de descubrimiento limitado en lugar de la general.
br_edr: Establece la bandera de soporte de BR/EDR.
name: Nombre local completo opcional a incrustar.
services: Iterable de bluetooth.UUID a anunciar.
appearance: Valor de apariencia de 16 bits (consulta los números asignados de Bluetooth).
manufacturer: Tupla de (company_id, data_bytes) a anunciar como datos específicos del fabricante.
timeout_ms: Deja de anunciar después de tantos milisegundos sin una conexión. None significa anunciar hasta que se conecte.

aioble.register_services(*services: Service) → None¶

Registra uno o más objetos Service (y sus características y descriptores) con el servidor GATT. Debe llamarse una vez antes de iniciar advertise. Las llamadas posteriores reemplazan el registro anterior.

services: Una o más instancias de Service.

Constantes a nivel de módulo¶

aioble.ADDR_PUBLIC¶: Tipo de dirección de dispositivo BLE pública (0).

aioble.ADDR_RANDOM¶: Tipo de dirección de dispositivo BLE aleatoria (1).

Excepciones¶

exception aioble.GattError¶: Se lanza cuando una operación GATT remota (lectura / escritura / indicación) se completa con un estado distinto de cero. El código de estado está disponible en el atributo _status.

exception aioble.DeviceDisconnectedError¶: Se lanza dentro de una operación asíncrona (p. ej. read, write, notified) cuando la conexión subyacente se cae mientras se espera.

exception aioble.L2CAPDisconnectedError¶: Se lanza cuando se intenta una operación de envío/recepción/vaciado de un canal L2CAP sobre (o es interrumpida por) un canal desconectado.

exception aioble.L2CAPConnectionError¶: La lanza DeviceConnection.l2cap_connect cuando falla el establecimiento del canal. El código de estado de Bluetooth es el primer argumento.

Clases¶

class aioble.Device(addr_type: int, addr: bytes | str)¶

Representa un dispositivo BLE remoto por dirección. Dos instancias de Device se consideran iguales si coinciden tanto addr_type como addr. Se usa como manejador para iniciar conexiones.

addr_type: ADDR_PUBLIC o ADDR_RANDOM.
addr: Dirección de seis bytes como bytes, o una cadena hexadecimal separada por dos puntos (p. ej. "aa:bb:cc:dd:ee:ff").

addr_type¶: El tipo de dirección con el que se construyó el dispositivo.

addr¶: La dirección de dispositivo en bruto de seis bytes.

addr_hex() → str¶: Devuelve la dirección formateada como una cadena hexadecimal separada por dos puntos.

connect(timeout_ms: int = 10000, scan_duration_ms: int | None = None, min_conn_interval_us: int | None = None, max_conn_interval_us: int | None = None) → Awaitable[DeviceConnection]¶

Asíncrono. Inicia una conexión GAP a este dispositivo y devuelve la DeviceConnection resultante. Cancela cualquier escaneo en progreso.

timeout_ms: Cuánto tiempo esperar a que se complete la conexión.
scan_duration_ms: Duración inicial del escaneo antes de conectar (específico del controlador).
min_conn_interval_us / max_conn_interval_us: Límites opcionales del intervalo de conexión, en microsegundos.

class aioble.DeviceConnection¶

Una conexión GAP activa a un Device. La devuelven Device.connect() o advertise. Admite su uso como gestor de contexto async with que se desconecta automáticamente al salir.

No la construyas directamente.

device¶: El Device subyacente.

encrypted¶: True una vez que el enlace está cifrado (p. ej. después del emparejamiento).

authenticated¶: True si el enlace fue autenticado (emparejamiento protegido contra MITM).

bonded¶: True si el emparejamiento produjo claves de vinculación (bonding).

key_size¶: Tamaño de la clave de cifrado negociada en bytes, o False si no está cifrada.

mtu¶: MTU de ATT negociada después de exchange_mtu, o None hasta que se establezca.

is_connected() → bool¶: Devuelve si la conexión sigue activa.

disconnect(timeout_ms: int = 2000) → Awaitable[None]¶

Asíncrono. Desconecta y espera la IRQ de desconexión.

timeout_ms: Tiempo máximo a esperar para la desconexión.

disconnected(timeout_ms: int | None = None, disconnect: bool = False) → Awaitable[None]¶

Asíncrono. Espera a que la conexión sea terminada por cualquiera de las partes. Si disconnect es True, desconecta activamente primero.

timeout_ms: Tiempo máximo a esperar. None significa esperar para siempre.
disconnect: Si es True, inicia la desconexión.

timeout(timeout_ms: int | None) → DeviceTimeout¶

Devuelve un gestor de contexto que cancela su cuerpo si transcurre el tiempo de espera (lanzando asyncio.TimeoutError) o si el dispositivo se desconecta (lanzando DeviceDisconnectedError).

timeout_ms: Tiempo de espera en milisegundos, o None para sin tiempo de espera.

exchange_mtu(mtu: int | None = None, timeout_ms: int = 1000) → Awaitable[int]¶

Asíncrono. Inicia un intercambio de MTU de ATT y devuelve la MTU negociada.

mtu: MTU preferida opcional a establecer en la interfaz BLE subyacente antes del intercambio.
timeout_ms: Tiempo de espera para el intercambio.

service(uuid: bluetooth.UUID, timeout_ms: int = 2000) → Awaitable[ClientService | None]¶: Asíncrono. Descubre un único servicio remoto que coincida con uuid, o None si no se encuentra.

services(uuid: bluetooth.UUID | None = None, timeout_ms: int = 2000) → ClientDiscover¶

Devuelve un iterador asíncrono de objetos ClientService remotos. Úsalo con async for y ejecuta el bucle hasta completarlo.

uuid: Filtro de UUID opcional. None devuelve todos los servicios.
timeout_ms: Tiempo de espera por descubrimiento.

pair(bond: bool = True, le_secure: bool = True, mitm: bool = False, io: int = 3, timeout_ms: int = 20000) → Awaitable[None]¶

Asíncrono. Inicia el emparejamiento en esta conexión. Actualiza los atributos encrypted / authenticated / bonded / key_size cuando se completa.

bond: Conserva las claves de emparejamiento.
le_secure: Usa LE Secure Connections.
mitm: Requiere protección contra ataques de intermediario (man-in-the-middle).
io: Constante de capacidad de E/S (p. ej. 3 para sin entrada/salida).
timeout_ms: Tiempo de espera del emparejamiento.

l2cap_accept(psm: int, mtu: int, timeout_ms: int | None = None) → Awaitable[L2CAPChannel]¶

Asíncrono. Escucha en el PSM dado y devuelve un L2CAPChannel una vez que el remoto lo abre.

psm: Multiplexor de Protocolo/Servicio (PSM) en el que escuchar.
mtu: Tamaño máximo de recepción, en bytes.
timeout_ms: Tiempo máximo a esperar a que el remoto se conecte.

l2cap_connect(psm: int, mtu: int, timeout_ms: int = 1000) → Awaitable[L2CAPChannel]¶

Asíncrono. Abre un canal L2CAP al remoto en el PSM dado.

psm: Multiplexor de Protocolo/Servicio (PSM) al que conectar.
mtu: Tamaño máximo de recepción, en bytes.
timeout_ms: Tiempo de espera de la conexión.

class aioble.ScanResult¶

Un único dispositivo descubierto durante scan. La misma instancia se vuelve a producir a medida que llegan nuevos datos de anuncio.

No la construyas directamente.

device¶: El Device subyacente.

rssi¶: Último RSSI reportado, en dBm.

adv_data¶: Carga útil de anuncio en bruto (bytes o None).

resp_data¶: Carga útil de respuesta de escaneo en bruto (bytes o None), si el escaneo activo está habilitado.

connectable¶: True si el anuncio más reciente era conectable.

name() → str | None¶: Decodifica el nombre local anunciado completo (o abreviado) de la carga útil, o None si no está presente.

services() → Iterator[bluetooth.UUID]¶: Generador que produce cada bluetooth.UUID anunciado en los campos de lista de servicios de 16/32/128 bits.

manufacturer(filter: int | None = None) → Iterator[tuple[int, bytes]]¶

Generador que produce tuplas (company_id, data) de los campos de anuncio específicos del fabricante.

filter: Si se proporciona, solo produce entradas cuyo ID de empresa coincida.

class aioble.Service(uuid: bluetooth.UUID)¶

Un servicio GATT local. Construye un servicio con una o más instancias de Characteristic, luego pásalo a register_services.

uuid: El UUID del servicio.

uuid¶: El UUID del servicio.

characteristics¶: Lista de objetos Characteristic vinculados a este servicio.

class aioble.Characteristic(service: Service, uuid: bluetooth.UUID, read: bool = False, write: bool = False, write_no_response: bool = False, notify: bool = False, indicate: bool = False, initial: bytes | None = None, capture: bool = False)¶

Una característica GATT local. Construir una la añade automáticamente a service.

service: El Service propietario.
uuid: El UUID de la característica.
read, write, write_no_response, notify, indicate: Booleanos que seleccionan las operaciones GATT admitidas.
initial: Valor inicial opcional (bytes).
capture: Si es True, los valores escritos se ponen en cola (hasta 10 de profundidad) para que las escrituras rápidas consecutivas no se pierdan. Cada llamada a written devuelve entonces una tupla (connection, data).

uuid¶: El UUID de la característica.

flags¶: Máscara de bits de las banderas de propiedad GATT construidas a partir del constructor.

descriptors¶: Lista de objetos Descriptor vinculados a esta característica.

read() → bytes¶: Lee el valor actual de la base de datos GATT local.

write(data: bytes, send_update: bool = False) → None¶

Actualiza el valor en la base de datos GATT local.

data: Bytes del nuevo valor.
send_update: Si es True, también notifica/indica a cada conexión suscrita.

notify(connection: DeviceConnection, data: bytes | None = None) → None¶

Envía una Notify GATT a connection.

connection: La conexión de cliente de destino.
data: Carga útil a enviar. Si es None, se envía el valor local actual.

indicate(connection: DeviceConnection, data: bytes | None = None, timeout_ms: int = 1000) → Awaitable[None]¶

Asíncrono. Envía una Indicate GATT a connection y espera la confirmación del cliente. Lanza GattError ante un estado distinto de cero.

connection: La conexión de cliente de destino.
data: Carga útil a indicar, o None para enviar el valor local.
timeout_ms: Tiempo máximo a esperar la confirmación.

written(timeout_ms: int | None = None) → Awaitable[DeviceConnection | tuple[DeviceConnection, bytes]]¶

Asíncrono. Espera una escritura remota. Devuelve la DeviceConnection que escribe, o (connection, data) si la característica se creó con capture=True.

timeout_ms: Tiempo máximo a esperar. None espera para siempre.

on_read(connection: DeviceConnection) → int¶: Hook sobrescribible invocado de forma síncrona cuando se recibe una lectura remota. Devuelve 0 para permitir la lectura o un código de error ATT distinto de cero para rechazarla. La implementación predeterminada devuelve 0.

class aioble.BufferedCharacteristic(service: Service, uuid: bluetooth.UUID, max_len: int = 20, append: bool = False, **kwargs)¶

Una Characteristic cuyo búfer GATT de respaldo puede configurarse. Útil para recibir valores más grandes que el tamaño de atributo predeterminado, o para poner en cola escrituras consecutivas.

max_len: Tamaño del búfer, en bytes.
append: Si es True, las escrituras secuenciales se añaden al búfer en lugar de sobrescribirlo.

Otros argumentos se reenvían a Characteristic.

class aioble.Descriptor(characteristic: Characteristic, uuid: bluetooth.UUID, read: bool = False, write: bool = False, initial: bytes | None = None)¶

Un descriptor GATT local. Construir uno lo añade automáticamente a characteristic. Hereda read, write y written de Characteristic.

characteristic: La Characteristic propietaria.
uuid: El UUID del descriptor.
read, write: Booleanos que seleccionan las operaciones GATT admitidas.
initial: Valor inicial opcional (bytes).

class aioble.ClientService¶

Un servicio GATT remoto descubierto en un par. Lo devuelve DeviceConnection.service() o se itera desde DeviceConnection.services().

No la construyas directamente.

connection¶: La DeviceConnection propietaria.

uuid¶: El UUID del servicio remoto.

characteristic(uuid: bluetooth.UUID, timeout_ms: int = 2000) → Awaitable[ClientCharacteristic | None]¶: Asíncrono. Descubre una única característica por UUID, o None si no se encuentra.

characteristics(uuid: bluetooth.UUID | None = None, timeout_ms: int = 2000) → ClientDiscover¶

Devuelve un iterador asíncrono de objetos ClientCharacteristic. Úsalo con async for y ejecuta el bucle hasta completarlo.

uuid: Filtro de UUID opcional.
timeout_ms: Tiempo de espera por descubrimiento.

class aioble.ClientCharacteristic¶

Una característica GATT remota descubierta en un par. La devuelve ClientService.characteristic() o se itera desde ClientService.characteristics().

No la construyas directamente.

service¶: El ClientService propietario.

uuid¶: El UUID de la característica.

properties¶: Máscara de bits de las operaciones GATT admitidas según las reporta el par.

read(timeout_ms: int = 1000) → Awaitable[bytes]¶

Asíncrono. Emite una Read GATT y devuelve el valor. Lanza GattError ante un estado distinto de cero.

timeout_ms: Tiempo de espera de lectura.

write(data: bytes, response: bool | None = None, timeout_ms: int = 1000) → Awaitable[None]¶

Asíncrono. Emite una Write GATT.

data: Valor a escribir.
response: True para requerir una respuesta de escritura (y lanzar GattError en caso de fallo). False para escritura-sin-respuesta. None (predeterminado) selecciona automáticamente según lo que anuncie el par.
timeout_ms: Tiempo de espera de escritura (solo relevante si response es True).

notified(timeout_ms: int | None = None) → Awaitable[bytes]¶

Asíncrono. Espera la siguiente notificación en esta característica y devuelve su carga útil. Regresa inmediatamente si ya hay una notificación en cola.

timeout_ms: Tiempo máximo a esperar. None espera para siempre.

indicated(timeout_ms: int | None = None) → Awaitable[bytes]¶

Asíncrono. Espera la siguiente indicación en esta característica y devuelve su carga útil.

timeout_ms: Tiempo máximo a esperar.

subscribe(notify: bool = True, indicate: bool = False) → Awaitable[None]¶

Asíncrono. Escribe el Client Characteristic Configuration Descriptor (CCCD) para suscribirse (o cancelar la suscripción) a notificaciones y/o indicaciones.

notify: Habilita las notificaciones.
indicate: Habilita las indicaciones.

descriptor(uuid: bluetooth.UUID, timeout_ms: int = 2000) → Awaitable[ClientDescriptor | None]¶: Asíncrono. Descubre un único descriptor por UUID, o None si no se encuentra.

descriptors(timeout_ms: int = 2000) → ClientDiscover¶: Devuelve un iterador asíncrono de objetos ClientDescriptor. Úsalo con async for y ejecuta el bucle hasta completarlo.

class aioble.ClientDescriptor¶

Un descriptor GATT remoto descubierto en un par. Hereda read y write de ClientCharacteristic.

No la construyas directamente.

characteristic¶: La ClientCharacteristic propietaria.

uuid¶: El UUID del descriptor.

class aioble.L2CAPChannel¶

Un canal L2CAP activo orientado a conexión. Lo devuelve DeviceConnection.l2cap_accept() o DeviceConnection.l2cap_connect(). Admite su uso como gestor de contexto async with que se desconecta automáticamente al salir.

No la construyas directamente.

our_mtu¶: Tamaño máximo, en bytes, que el par puede enviarnos en un solo SDU.

peer_mtu¶: Tamaño máximo, en bytes, que podemos enviar al par en un solo SDU.

available() → bool¶: Devuelve de forma síncrona True si hay datos de recepción almacenados en búfer listos (es decir, recvinto no se bloqueará).

recvinto(buf: bytearray, timeout_ms: int | None = None) → Awaitable[int]¶

Asíncrono. Recibe en buf, devolviendo el número de bytes leídos. Espera nuevos datos si el canal está vacío.

buf: Búfer preasignado a rellenar.
timeout_ms: Tiempo máximo a esperar. None espera para siempre.

send(buf: bytes, timeout_ms: int | None = None, chunk_size: int | None = None) → Awaitable[None]¶

Asíncrono. Envía buf por el canal, fragmentando las cargas útiles más grandes en trozos del tamaño de la MTU. Espera créditos de control de flujo según sea necesario.

buf: Objeto de tipo bytes a enviar.
timeout_ms: Tiempo máximo a esperar por trozo.
chunk_size: Anulación opcional para el tamaño de trozo por llamada. Limitado a min(our_mtu * 2, peer_mtu).

flush(timeout_ms: int | None = None) → Awaitable[None]¶

Asíncrono. Espera hasta que cualquier send estancado haya sido drenado por el controlador.

timeout_ms: Tiempo máximo a esperar.

disconnect(timeout_ms: int = 1000) → Awaitable[None]¶

Asíncrono. Desconecta el canal y espera la IRQ de desconexión.

timeout_ms: Tiempo máximo a esperar.

disconnected(timeout_ms: int = 1000) → Awaitable[None]¶

Asíncrono. Espera hasta que el canal sea desconectado por cualquiera de las partes.

timeout_ms: Tiempo máximo a esperar.

aioble — BLE asíncrono¶

Ejemplos¶

Funciones a nivel de módulo¶

Constantes a nivel de módulo¶

Excepciones¶

Clases¶

`aioble` — BLE asíncrono¶