aioble — 비동기 BLE

aioble은 bluetooth 모듈을 감싼 고수준의 asyncio 친화적 래퍼입니다. 스캔, 연결, 광고, GATT 서비스, L2CAP 채널을 위한 깔끔한 코루틴을 제공합니다.

모든 원격 작업(연결, 연결 해제, 클라이언트 읽기/쓰기, 서버 indicate, l2cap recv/send, 페어링)은 await 가능하며 타임아웃을 지원합니다.

지원하는 역할:

  • 브로드캐스터(광고자) — 일반적인 필드에 대한 광고 및 스캔 응답 페이로드를 생성하고, 페이로드를 광고와 스캔 응답에 자동으로 분할하며, 무기한 또는 지정된 기간 동안 광고합니다.

  • 페리페럴(주변기기) — 센트럴로부터의 연결을 대기하고, MTU 교환을 대기합니다.

  • 옵저버(스캐너) — 수동 및 능동 스캔, 동일한 장치의 광고 및 스캔 응답 페이로드 결합, 광고 페이로드에서 일반적인 필드 파싱.

  • 센트럴 — 페리페럴에 연결하고 MTU 교환을 시작합니다.

  • GATT 클라이언트 — 서비스 / 특성 / 디스크립터를 검색(선택적으로 UUID로); 특성과 디스크립터에 대해 읽기 / 쓰기 / 응답 포함 쓰기; (CCCD를 통해) 알림 및 지시 구독; 알림 및 지시 대기.

  • GATT 서버 — 서비스 / 특성 / 디스크립터 등록; 특성과 디스크립터에 대한 쓰기 대기; 읽기 요청 가로채기; 알림 및 지시 전송(및 응답 대기).

  • L2CAP — L2CAP 연결 지향 채널의 수락 및 연결, 채널 흐름 제어 관리.

  • 보안 — JSON 기반 키/시크릿 관리, 페어링 시작, 암호화 / 인증 상태 조회.

예제

근처의 BLE 장치를 스캔하고 발견될 때마다 각각을 출력합니다:

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())

심박수 서비스를 광고하는 페리페럴에 센트럴로 연결하고 그 측정 알림을 구독합니다:

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())

페리페럴로 동작: GATT 서비스를 등록하고 광고한 다음, 연결하는 모든 대상에게 알림을 전송합니다:

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())

모듈 수준 함수

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

bluetooth.BLE.config()로 전달하며, 먼저 BLE 라디오가 활성화되어 있는지 확인합니다.

args

조회할 단일 매개변수 이름(선택 사항).

kwargs

구성 값을 설정하기 위한 키워드 인수.

aioble.stop() None

기반이 되는 BLE 라디오를 비활성화하고 등록된 모든 하위 모듈 종료 핸들러를 실행합니다. 이를 호출한 후에는 스캐너, 광고자, 연결 및 L2CAP 채널이 모두 해체됩니다.

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

발견된 각 고유 장치(또는 알려진 장치로부터의 새로운 광고 데이터 조각)에 대해 ScanResult 인스턴스를 산출하는 scan 비동기 컨텍스트 관리자 / 비동기 반복자를 반환합니다.

duration_ms

스캔 시간(밀리초). 컨텍스트 관리자가 종료될 때까지 무기한 스캔하려면 0을 전달합니다.

interval_us

스캔 간격(마이크로초). 기본값은 1,280,000입니다.

window_us

스캔 윈도우(마이크로초)(interval_us 이하여야 함). 기본값은 11,250입니다.

active

True이면 능동 스캔을 수행합니다(스캔 응답 데이터를 요청). 기본값은 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

광고를 시작하고 들어오는 센트럴 연결을 대기하는 비동기 코루틴입니다. 연결된 센트럴을 나타내는 DeviceConnection을 반환하거나, 타임아웃 시 asyncio.TimeoutError를 발생시킵니다.

interval_us

광고 간격(마이크로초).

adv_data

원시 광고 페이로드. 설정되지 않으면 adv_data는 나머지 키워드 인수로부터 빌드됩니다.

resp_data

원시 스캔 응답 페이로드. 필요한 경우 adv_data의 오버플로로 자동 채워집니다.

connectable

True이면 연결 가능한 광고입니다.

limited_disc

일반 대신 제한적 검색 가능(limited-discoverable) 플래그를 사용합니다.

br_edr

BR/EDR 지원 플래그를 설정합니다.

name

포함할 전체 로컬 이름(선택 사항).

services

광고할 bluetooth.UUID의 이터러블.

appearance

16비트 appearance 값(Bluetooth 할당 번호 참조).

manufacturer

제조사별 데이터로 광고할 (company_id, data_bytes) 튜플.

timeout_ms

연결 없이 이만큼의 밀리초가 지나면 광고를 중지합니다. None은 연결될 때까지 광고함을 의미합니다.

aioble.register_services(*services: Service) None

하나 이상의 Service 객체(및 그 특성과 디스크립터)를 GATT 서버에 등록합니다. advertise를 시작하기 전에 한 번 호출해야 합니다. 이후 호출은 이전 등록을 대체합니다.

services

하나 이상의 Service 인스턴스.

모듈 수준 상수

aioble.ADDR_PUBLIC

공용(public) BLE 장치 주소 유형(0).

aioble.ADDR_RANDOM

랜덤(random) BLE 장치 주소 유형(1).

예외

exception aioble.GattError

원격 GATT 작업(읽기 / 쓰기 / indicate)이 0이 아닌 상태로 완료될 때 발생합니다. 상태 코드는 _status 속성에서 확인할 수 있습니다.

exception aioble.DeviceDisconnectedError

대기 중에 기반 연결이 끊어졌을 때 비동기 작업(예: read, write, notified) 내부에서 발생합니다.

exception aioble.L2CAPDisconnectedError

연결이 끊긴 채널에서(또는 그로 인해 중단되어) L2CAP 채널 send/recv/flush 작업이 시도될 때 발생합니다.

exception aioble.L2CAPConnectionError

채널을 설정하는 데 실패했을 때 DeviceConnection.l2cap_connect에 의해 발생합니다. Bluetooth 상태 코드가 첫 번째 인수입니다.

클래스

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

주소로 원격 BLE 장치를 나타냅니다. 두 Device 인스턴스는 addr_typeaddr이 모두 일치하면 동일한 것으로 비교됩니다. 연결을 시작하기 위한 핸들로 사용됩니다.

addr_type

ADDR_PUBLIC 또는 ADDR_RANDOM 중 하나.

addr

bytes 형식의 6바이트 주소 또는 콜론으로 구분된 16진수 문자열(예: "aa:bb:cc:dd:ee:ff").

addr_type

장치가 생성될 때 사용된 주소 유형.

addr

원시 6바이트 장치 주소.

addr_hex() str

주소를 콜론으로 구분된 16진수 문자열로 형식화하여 반환합니다.

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]

비동기. 이 장치에 대한 GAP 연결을 시작하고 그 결과 DeviceConnection을 반환합니다. 진행 중인 스캔을 취소합니다.

timeout_ms

연결이 완료될 때까지 대기할 시간.

scan_duration_ms

연결 전 초기 스캔 시간(컨트롤러에 따라 다름).

min_conn_interval_us / max_conn_interval_us

연결 간격 범위(선택 사항, 마이크로초).

class aioble.DeviceConnection

Device에 대한 활성 GAP 연결. Device.connect() 또는 advertise에서 반환됩니다. 종료 시 자동으로 연결을 해제하는 async with 컨텍스트 관리자로 사용할 수 있습니다.

직접 생성하지 마십시오.

device

기반이 되는 Device.

encrypted

링크가 암호화되면(예: 페어링 후) True.

authenticated

링크가 인증되었으면(MITM 보호 페어링) True.

bonded

페어링으로 본딩 키가 생성되었으면 True.

key_size

협상된 암호화 키 크기(바이트), 암호화되지 않은 경우 False.

mtu

exchange_mtu 이후 협상된 ATT MTU, 설정되기 전까지는 None.

is_connected() bool

연결이 여전히 활성 상태인지 여부를 반환합니다.

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

비동기. 연결을 해제하고 연결 해제 IRQ를 대기합니다.

timeout_ms

연결 해제를 대기할 최대 시간.

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

비동기. 양쪽 중 어느 한쪽에 의해 연결이 종료될 때까지 대기합니다. disconnectTrue이면 먼저 능동적으로 연결을 해제합니다.

timeout_ms

대기할 최대 시간. None은 영원히 대기함을 의미합니다.

disconnect

True이면 연결 해제를 시작합니다.

timeout(timeout_ms: int | None) DeviceTimeout

타임아웃이 경과하거나(asyncio.TimeoutError 발생) 장치가 연결을 해제하면(DeviceDisconnectedError 발생) 본문을 취소하는 컨텍스트 관리자를 반환합니다.

timeout_ms

타임아웃(밀리초) 또는 타임아웃 없음을 위한 None.

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

비동기. ATT MTU 교환을 시작하고 협상된 MTU를 반환합니다.

mtu

교환 전에 기반 BLE 인터페이스에 설정할 선호 MTU(선택 사항).

timeout_ms

교환에 대한 타임아웃.

service(uuid: bluetooth.UUID, timeout_ms: int = 2000) Awaitable[ClientService | None]

비동기. uuid와 일치하는 단일 원격 서비스를 검색하거나, 찾을 수 없으면 None을 반환합니다.

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

원격 ClientService 객체의 비동기 반복자를 반환합니다. async for와 함께 사용하고 루프를 끝까지 실행하십시오.

uuid

UUID 필터(선택 사항). None은 모든 서비스를 반환합니다.

timeout_ms

검색당 타임아웃.

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

비동기. 이 연결에서 페어링을 시작합니다. 완료되면 encrypted / authenticated / bonded / key_size 속성을 업데이트합니다.

bond

페어링 키를 영구 저장합니다.

le_secure

LE Secure Connections를 사용합니다.

mitm

중간자(man-in-the-middle) 보호를 요구합니다.

io

IO 기능 상수(예: 입력/출력 없음의 경우 3).

timeout_ms

페어링 타임아웃.

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

비동기. 지정된 PSM에서 수신 대기하고 원격이 채널을 열면 L2CAPChannel을 반환합니다.

psm

수신 대기할 Protocol/Service Multiplexer.

mtu

최대 수신 크기(바이트).

timeout_ms

원격이 연결될 때까지 대기할 최대 시간.

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

비동기. 지정된 PSM에서 원격으로의 L2CAP 채널을 엽니다.

psm

연결할 Protocol/Service Multiplexer.

mtu

최대 수신 크기(바이트).

timeout_ms

연결 타임아웃.

class aioble.ScanResult

scan 중에 발견된 단일 장치. 새로운 광고 데이터가 도착하면 동일한 인스턴스가 다시 산출됩니다.

직접 생성하지 마십시오.

device

기반이 되는 Device.

rssi

마지막으로 보고된 RSSI(dBm).

adv_data

원시 광고 페이로드(bytes 또는 None).

resp_data

능동 스캔이 활성화된 경우 원시 스캔 응답 페이로드(bytes 또는 None).

connectable

가장 최근 광고가 연결 가능했으면 True.

name() str | None

페이로드에서 광고된 전체(또는 단축된) 로컬 이름을 디코딩하거나, 존재하지 않으면 None을 반환합니다.

services() Iterator[bluetooth.UUID]

16/32/128비트 서비스 목록 필드에 광고된 각 bluetooth.UUID를 산출하는 제너레이터.

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

제조사별 광고 필드로부터 (company_id, data) 튜플을 산출하는 제너레이터.

filter

지정된 경우, company ID가 일치하는 항목만 산출합니다.

class aioble.Service(uuid: bluetooth.UUID)

로컬 GATT 서비스. 하나 이상의 Characteristic 인스턴스로 서비스를 빌드한 다음 register_services에 전달합니다.

uuid

서비스 UUID.

uuid

서비스 UUID.

characteristics

이 서비스에 바인딩된 Characteristic 객체의 목록.

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)

로컬 GATT 특성. 하나를 생성하면 자동으로 service에 추가됩니다.

service

소유하는 Service.

uuid

특성 UUID.

read, write, write_no_response, notify, indicate

지원되는 GATT 작업을 선택하는 불리언 값.

initial

초기 값(bytes, 선택 사항).

capture

True이면 기록된 값이 (최대 10개까지) 큐에 저장되어 빠른 연속 쓰기가 손실되지 않습니다. 그러면 각 written 호출은 (connection, data) 튜플을 반환합니다.

uuid

특성 UUID.

flags

생성자로부터 빌드된 GATT 속성 플래그의 비트마스크.

descriptors

이 특성에 바인딩된 Descriptor 객체의 목록.

read() bytes

로컬 GATT 데이터베이스에서 현재 값을 읽습니다.

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

로컬 GATT 데이터베이스의 값을 업데이트합니다.

data

새 값 바이트.

send_update

True이면 구독한 모든 연결에 알림/지시도 전송합니다.

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

connection에 GATT Notify를 전송합니다.

connection

대상 클라이언트 연결.

data

전송할 페이로드. None이면 현재 로컬 값이 전송됩니다.

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

비동기. connection에 GATT Indicate를 전송하고 클라이언트 확인을 대기합니다. 0이 아닌 상태에서 GattError를 발생시킵니다.

connection

대상 클라이언트 연결.

data

지시할 페이로드, 또는 로컬 값을 전송하려면 None.

timeout_ms

확인을 대기할 최대 시간.

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

비동기. 원격 쓰기를 대기합니다. 쓰기를 수행한 DeviceConnection을 반환하거나, 특성이 capture=True로 생성된 경우 (connection, data)를 반환합니다.

timeout_ms

대기할 최대 시간. None은 영원히 대기합니다.

on_read(connection: DeviceConnection) int

원격 읽기가 수신될 때 동기적으로 호출되는 재정의 훅. 읽기를 허용하려면 0을, 거부하려면 0이 아닌 ATT 오류 코드를 반환합니다. 기본 구현은 0을 반환합니다.

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

기반 GATT 버퍼를 구성할 수 있는 Characteristic. 기본 속성 크기보다 큰 값을 수신하거나 연속 쓰기를 큐에 저장하는 데 유용합니다.

max_len

버퍼 크기(바이트).

append

True이면 순차적 쓰기가 덮어쓰는 대신 버퍼에 추가됩니다.

다른 인수는 Characteristic으로 전달됩니다.

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

로컬 GATT 디스크립터. 하나를 생성하면 자동으로 characteristic에 추가됩니다. Characteristic으로부터 read, write, written을 상속합니다.

characteristic

소유하는 Characteristic.

uuid

디스크립터 UUID.

read, write

지원되는 GATT 작업을 선택하는 불리언 값.

initial

초기 값(bytes, 선택 사항).

class aioble.ClientService

피어에서 검색된 원격 GATT 서비스. DeviceConnection.service()에서 반환되거나 DeviceConnection.services()에서 반복됩니다.

직접 생성하지 마십시오.

connection

소유하는 DeviceConnection.

uuid

원격 서비스 UUID.

characteristic(uuid: bluetooth.UUID, timeout_ms: int = 2000) Awaitable[ClientCharacteristic | None]

비동기. UUID로 단일 특성을 검색하거나, 찾을 수 없으면 None을 반환합니다.

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

ClientCharacteristic 객체의 비동기 반복자를 반환합니다. async for와 함께 사용하고 루프를 끝까지 실행하십시오.

uuid

UUID 필터(선택 사항).

timeout_ms

검색당 타임아웃.

class aioble.ClientCharacteristic

피어에서 검색된 원격 GATT 특성. ClientService.characteristic()에서 반환되거나 ClientService.characteristics()에서 반복됩니다.

직접 생성하지 마십시오.

service

소유하는 ClientService.

uuid

특성 UUID.

properties

피어가 보고한 지원 GATT 작업의 비트마스크.

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

비동기. GATT Read를 실행하고 값을 반환합니다. 0이 아닌 상태에서 GattError를 발생시킵니다.

timeout_ms

읽기 타임아웃.

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

비동기. GATT Write를 실행합니다.

data

쓸 값.

response

True이면 쓰기 응답을 요구합니다(실패 시 GattError를 발생시킴). False이면 응답 없는 쓰기를 수행합니다. None(기본값)이면 피어(peer)가 광고하는 내용에 따라 자동 선택됩니다.

timeout_ms

쓰기 타임아웃(responseTrue인 경우에만 관련됨).

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

비동기. 이 특성에 대한 다음 알림을 대기하고 그 페이로드를 반환합니다. 이미 알림이 큐에 있으면 즉시 반환합니다.

timeout_ms

대기할 최대 시간. None은 영원히 대기합니다.

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

비동기. 이 특성에 대한 다음 지시를 대기하고 그 페이로드를 반환합니다.

timeout_ms

대기할 최대 시간.

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

비동기. 알림 및/또는 지시를 구독(또는 구독 해제)하기 위해 Client Characteristic Configuration Descriptor(CCCD)를 씁니다.

notify

알림을 활성화합니다.

indicate

지시를 활성화합니다.

descriptor(uuid: bluetooth.UUID, timeout_ms: int = 2000) Awaitable[ClientDescriptor | None]

비동기. UUID로 단일 디스크립터를 검색하거나, 찾을 수 없으면 None을 반환합니다.

descriptors(timeout_ms: int = 2000) ClientDiscover

ClientDescriptor 객체의 비동기 반복자를 반환합니다. async for와 함께 사용하고 루프를 끝까지 실행하십시오.

class aioble.ClientDescriptor

피어에서 검색된 원격 GATT 디스크립터. ClientCharacteristic으로부터 readwrite를 상속합니다.

직접 생성하지 마십시오.

characteristic

소유하는 ClientCharacteristic.

uuid

디스크립터 UUID.

class aioble.L2CAPChannel

활성 L2CAP 연결 지향 채널. DeviceConnection.l2cap_accept() 또는 DeviceConnection.l2cap_connect()에서 반환됩니다. 종료 시 자동으로 연결을 해제하는 async with 컨텍스트 관리자로 사용할 수 있습니다.

직접 생성하지 마십시오.

our_mtu

피어가 단일 SDU로 우리에게 전송할 수 있는 최대 크기(바이트).

peer_mtu

우리가 단일 SDU로 피어에게 전송할 수 있는 최대 크기(바이트).

available() bool

버퍼링된 수신 데이터가 준비되어 있으면(즉, recvinto가 블록되지 않으면) 동기적으로 True를 반환합니다.

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

비동기. buf로 수신하여 읽은 바이트 수를 반환합니다. 채널이 비어 있으면 새 데이터를 await합니다.

buf

채울 미리 할당된 버퍼.

timeout_ms

대기할 최대 시간. None은 영원히 대기합니다.

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

비동기. 채널에서 buf를 전송하며, 더 큰 페이로드는 MTU 크기의 청크로 단편화합니다. 필요에 따라 흐름 제어 크레딧을 await합니다.

buf

전송할 바이트 유사 객체.

timeout_ms

청크당 대기할 최대 시간.

chunk_size

호출당 청크 크기에 대한 재정의(선택 사항). min(our_mtu * 2, peer_mtu)로 제한됩니다.

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

비동기. 멈춰 있던 send가 컨트롤러에 의해 비워질 때까지 대기합니다.

timeout_ms

대기할 최대 시간.

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

비동기. 채널의 연결을 해제하고 연결 해제 IRQ를 대기합니다.

timeout_ms

대기할 최대 시간.

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

비동기. 양쪽 중 어느 한쪽에 의해 채널의 연결이 해제될 때까지 대기합니다.

timeout_ms

대기할 최대 시간.