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.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_type과 addr이 모두 일치하면 동일한 것으로 비교됩니다. 연결을 시작하기 위한 핸들로 사용됩니다.- addr_type
ADDR_PUBLIC또는ADDR_RANDOM중 하나.- addr
bytes형식의 6바이트 주소 또는 콜론으로 구분된 16진수 문자열(예:"aa:bb:cc:dd:ee:ff").
- addr_type¶
장치가 생성될 때 사용된 주소 유형.
- addr¶
원시 6바이트 장치 주소.
- 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컨텍스트 관리자로 사용할 수 있습니다.직접 생성하지 마십시오.
- encrypted¶
링크가 암호화되면(예: 페어링 후)
True.
- authenticated¶
링크가 인증되었으면(MITM 보호 페어링)
True.
- bonded¶
페어링으로 본딩 키가 생성되었으면
True.
- key_size¶
협상된 암호화 키 크기(바이트), 암호화되지 않은 경우
False.
- mtu¶
exchange_mtu이후 협상된 ATT MTU, 설정되기 전까지는None.
- disconnect(timeout_ms: int = 2000) Awaitable[None]¶
비동기. 연결을 해제하고 연결 해제 IRQ를 대기합니다.
- timeout_ms
연결 해제를 대기할 최대 시간.
- disconnected(timeout_ms: int | None = None, disconnect: bool = False) Awaitable[None]¶
비동기. 양쪽 중 어느 한쪽에 의해 연결이 종료될 때까지 대기합니다. disconnect가
True이면 먼저 능동적으로 연결을 해제합니다.- 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중에 발견된 단일 장치. 새로운 광고 데이터가 도착하면 동일한 인스턴스가 다시 산출됩니다.직접 생성하지 마십시오.
- rssi¶
마지막으로 보고된 RSSI(dBm).
- adv_data¶
원시 광고 페이로드(
bytes또는None).
- resp_data¶
능동 스캔이 활성화된 경우 원시 스캔 응답 페이로드(
bytes또는None).
- connectable¶
가장 최근 광고가 연결 가능했으면
True.
- services() Iterator[bluetooth.UUID]¶
16/32/128비트 서비스 목록 필드에 광고된 각
bluetooth.UUID를 산출하는 제너레이터.
- 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객체의 목록.
- 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
쓰기 타임아웃(response가
True인 경우에만 관련됨).
- 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으로부터read와write를 상속합니다.직접 생성하지 마십시오.
- characteristic¶
소유하는
ClientCharacteristic.
- uuid¶
디스크립터 UUID.
- class aioble.L2CAPChannel¶
활성 L2CAP 연결 지향 채널.
DeviceConnection.l2cap_accept()또는DeviceConnection.l2cap_connect()에서 반환됩니다. 종료 시 자동으로 연결을 해제하는async with컨텍스트 관리자로 사용할 수 있습니다.직접 생성하지 마십시오.
- our_mtu¶
피어가 단일 SDU로 우리에게 전송할 수 있는 최대 크기(바이트).
- peer_mtu¶
우리가 단일 SDU로 피어에게 전송할 수 있는 최대 크기(바이트).
- 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
대기할 최대 시간.