`aioble` --- Async BLE¶

aioble adalah pembungkus asyncio tingkat tinggi yang ramah di sekitar modul bluetooth. Ini menyediakan coroutine yang bersih untuk pemindaian, koneksi, periklanan, layanan GATT, dan kanal L2CAP.

Semua operasi jarak jauh (connect, disconnect, client read/write, server indicate, l2cap recv/send, pair) dapat di-await dan mendukung timeout.

Peran yang didukung:

Broadcaster (advertiser) --- menghasilkan payload periklanan dan scan-response untuk bidang umum, secara otomatis membagi payload antara periklanan dan scan response, mengiklankan tanpa batas atau untuk durasi tertentu.
Peripheral --- menunggu koneksi dari central, menunggu pertukaran MTU.
Observer (scanner) --- pemindaian pasif dan aktif, menggabungkan payload periklanan dan scan-response untuk perangkat yang sama, menguraikan bidang umum dari payload periklanan.
Central --- terhubung ke peripheral, memulai pertukaran MTU.
GATT Client --- menemukan layanan / karakteristik / deskriptor (opsional berdasarkan UUID); membaca / menulis / menulis-dengan-respons pada karakteristik dan deskriptor; berlangganan notifikasi dan indikasi (via CCCD); menunggu notifikasi dan indikasi.
GATT Server --- mendaftarkan layanan / karakteristik / deskriptor; menunggu penulisan pada karakteristik dan deskriptor; mencegat permintaan baca; mengirim notifikasi dan indikasi (dan menunggu respons).
L2CAP --- menerima dan menghubungkan kanal berorientasi koneksi L2CAP, mengelola kontrol aliran kanal.
Security --- manajemen kunci/rahasia berbasis JSON, memulai pairing, mengkueri status enkripsi / autentikasi.

Contoh¶

Pindai perangkat BLE terdekat dan cetak setiap perangkat saat ditemukan:

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

Terhubung ke peripheral yang mengiklankan layanan Heart Rate sebagai central dan berlangganan notifikasi pengukurannya:

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

Bertindak sebagai peripheral: daftarkan layanan GATT, iklankan, dan kirim notifikasi ke siapa pun yang terhubung:

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

Fungsi tingkat modul¶

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

Meneruskan ke bluetooth.BLE.config(), memastikan radio BLE aktif terlebih dahulu.

args: Nama parameter tunggal opsional untuk dikueri.
kwargs: Argumen kata kunci untuk mengatur nilai konfigurasi.

aioble.stop() → None¶: Menonaktifkan radio BLE yang mendasarinya dan menjalankan semua penangan shutdown sub-modul yang terdaftar. Setelah memanggil ini, pemindai, pengiklan, koneksi, dan kanal L2CAP semuanya dihentikan.

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

Mengembalikan scan async context-manager / async-iterator yang menghasilkan instance ScanResult untuk setiap perangkat unik yang ditemukan (atau untuk setiap data periklanan baru dari perangkat yang dikenal).

duration_ms: Berapa lama memindai, dalam milidetik. Berikan 0 untuk memindai tanpa batas hingga context manager keluar.
interval_us: Interval pemindaian dalam mikrodetik. Default 1.280.000.
window_us: Jendela pemindaian dalam mikrodetik (harus kurang dari atau sama dengan interval_us). Default 11.250.
active: Jika True, lakukan pemindaian aktif (minta data scan response). Default 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¶

Coroutine async yang mulai mengiklankan dan menunggu koneksi central masuk. Mengembalikan DeviceConnection yang mewakili central yang terhubung, atau memunculkan asyncio.TimeoutError jika timeout.

interval_us: Interval periklanan, dalam mikrodetik.
adv_data: Payload periklanan mentah. Jika tidak diset, adv_data dibangun dari argumen kata kunci yang tersisa.
resp_data: Payload scan response mentah. Diisi otomatis dari luapan adv_data jika diperlukan.
connectable: Jika True, ini adalah iklan yang dapat dihubungkan.
limited_disc: Gunakan flag limited-discoverable alih-alih general.
br_edr: Atur flag BR/EDR-supported.
name: Nama lokal lengkap opsional untuk disematkan.
services: Iterable dari bluetooth.UUID untuk diiklankan.
appearance: Nilai appearance 16-bit (lihat nomor yang ditetapkan Bluetooth).
manufacturer: Tuple dari (company_id, data_bytes) untuk diiklankan sebagai data spesifik produsen.
timeout_ms: Hentikan periklanan setelah sekian milidetik tanpa koneksi. None berarti mengiklankan sampai terhubung.

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

Daftarkan satu atau lebih objek Service (beserta karakteristik dan deskriptornya) ke server GATT. Harus dipanggil sekali sebelum memulai advertise. Panggilan berikutnya menggantikan pendaftaran sebelumnya.

services: Satu atau lebih instance Service.

Konstanta tingkat modul¶

aioble.ADDR_PUBLIC¶: Tipe alamat perangkat BLE publik (0).

aioble.ADDR_RANDOM¶: Tipe alamat perangkat BLE acak (1).

Pengecualian¶

exception aioble.GattError¶: Dimunculkan ketika operasi GATT jarak jauh (read / write / indicate) selesai dengan status bukan nol. Kode status tersedia di atribut _status.

exception aioble.DeviceDisconnectedError¶: Dimunculkan di dalam operasi async (misalnya read, write, notified) ketika koneksi yang mendasarinya putus saat menunggu.

exception aioble.L2CAPDisconnectedError¶: Dimunculkan ketika operasi send/recv/flush kanal L2CAP dicoba pada (atau diinterupsi oleh) kanal yang terputus.

exception aioble.L2CAPConnectionError¶: Dimunculkan oleh DeviceConnection.l2cap_connect ketika pembentukan kanal gagal. Kode status Bluetooth adalah argumen pertama.

Kelas¶

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

Mewakili perangkat BLE jarak jauh berdasarkan alamat. Dua instance Device dianggap sama jika addr_type dan addr keduanya cocok. Digunakan sebagai handle untuk memulai koneksi.

addr_type: Salah satu dari ADDR_PUBLIC atau ADDR_RANDOM.
addr: Alamat enam byte sebagai bytes, atau string hex yang dipisahkan titik dua (misalnya "aa:bb:cc:dd:ee:ff").

addr_type¶: Tipe alamat yang digunakan saat perangkat dibuat.

addr¶: Alamat perangkat enam byte mentah.

addr_hex() → str¶: Kembalikan alamat yang diformat sebagai string hex yang dipisahkan titik dua.

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]¶

Async. Mulai koneksi GAP ke perangkat ini dan kembalikan DeviceConnection yang dihasilkan. Membatalkan pemindaian yang sedang berlangsung.

timeout_ms: Berapa lama menunggu koneksi selesai.
scan_duration_ms: Durasi pemindaian awal sebelum menghubungkan (spesifik controller).
min_conn_interval_us / max_conn_interval_us: Batas interval koneksi opsional, dalam mikrodetik.

class aioble.DeviceConnection¶

Koneksi GAP aktif ke Device. Dikembalikan oleh Device.connect() atau advertise. Mendukung penggunaan sebagai context manager async with yang memutus koneksi secara otomatis saat keluar.

Jangan buat secara langsung.

device¶: Device yang mendasarinya.

encrypted¶: True setelah tautan dienkripsi (misalnya setelah pairing).

authenticated¶: True jika tautan diautentikasi (pair yang dilindungi MITM).

bonded¶: True jika pairing menghasilkan kunci bonding.

key_size¶: Ukuran kunci enkripsi yang dinegosiasikan dalam byte, atau False jika tidak dienkripsi.

mtu¶: ATT MTU yang dinegosiasikan setelah exchange_mtu, atau None sampai diset.

is_connected() → bool¶: Kembalikan apakah koneksi masih aktif.

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

Async. Putuskan koneksi dan tunggu IRQ pemutusan koneksi.

timeout_ms: Waktu maksimum untuk menunggu pemutusan koneksi.

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

Async. Tunggu hingga koneksi diakhiri oleh salah satu pihak. Jika disconnect adalah True, ia memutus koneksi secara aktif terlebih dahulu.

timeout_ms: Waktu maksimum untuk menunggu. None berarti tunggu selamanya.
disconnect: Jika True, mulai pemutusan koneksi.

timeout(timeout_ms: int | None) → DeviceTimeout¶

Kembalikan context manager yang membatalkan isinya jika timeout berlalu (memunculkan asyncio.TimeoutError) atau perangkat terputus (memunculkan DeviceDisconnectedError).

timeout_ms: Timeout dalam milidetik, atau None tanpa timeout.

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

Async. Mulai pertukaran ATT MTU dan kembalikan MTU yang dinegosiasikan.

mtu: MTU pilihan opsional yang diset pada antarmuka BLE yang mendasarinya sebelum pertukaran.
timeout_ms: Timeout untuk pertukaran.

service(uuid: bluetooth.UUID, timeout_ms: int = 2000) → Awaitable[ClientService | None]¶: Async. Temukan satu layanan jarak jauh yang cocok dengan uuid, atau None jika tidak ditemukan.

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

Kembalikan iterator async dari objek ClientService jarak jauh. Gunakan dengan async for dan jalankan loop hingga selesai.

uuid: Filter UUID opsional. None mengembalikan setiap layanan.
timeout_ms: Timeout per penemuan.

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

Async. Mulai pairing pada koneksi ini. Memperbarui atribut encrypted / authenticated / bonded / key_size saat selesai.

bond: Simpan kunci pairing.
le_secure: Gunakan LE Secure Connections.
mitm: Memerlukan perlindungan man-in-the-middle.
io: Konstanta kemampuan IO (misalnya 3 untuk tidak ada input/output).
timeout_ms: Timeout pairing.

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

Async. Dengarkan pada PSM yang diberikan dan kembalikan L2CAPChannel setelah remote membukanya.

psm: Protocol/Service Multiplexer untuk didengarkan.
mtu: Ukuran penerimaan maksimum, dalam byte.
timeout_ms: Waktu maksimum untuk menunggu remote terhubung.

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

Async. Buka kanal L2CAP ke remote pada PSM yang diberikan.

psm: Protocol/Service Multiplexer untuk dihubungkan.
mtu: Ukuran penerimaan maksimum, dalam byte.
timeout_ms: Timeout koneksi.

class aioble.ScanResult¶

Satu perangkat yang ditemukan selama scan. Instance yang sama dihasilkan ulang saat data periklanan baru tiba.

Jangan buat secara langsung.

device¶: Device yang mendasarinya.

rssi¶: RSSI terakhir yang dilaporkan, dalam dBm.

adv_data¶: Payload periklanan mentah (bytes atau None).

resp_data¶: Payload scan response mentah (bytes atau None), jika pemindaian aktif diaktifkan.

connectable¶: True jika iklan terbaru dapat dihubungkan.

name() → str | None¶: Dekode nama lokal lengkap (atau dipersingkat) yang diiklankan dari payload, atau None jika tidak ada.

services() → Iterator[bluetooth.UUID]¶: Generator yang menghasilkan setiap bluetooth.UUID yang diiklankan dalam bidang daftar layanan 16/32/128-bit.

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

Generator yang menghasilkan tuple (company_id, data) dari bidang periklanan spesifik produsen.

filter: Jika diberikan, hanya hasilkan entri yang company ID-nya cocok.

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

Layanan GATT lokal. Bangun layanan dengan satu atau lebih instance Characteristic, lalu teruskan ke register_services.

uuid: UUID layanan.

uuid¶: UUID layanan.

characteristics¶: Daftar objek Characteristic yang terikat ke layanan ini.

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

Karakteristik GATT lokal. Membangunnya secara otomatis menambahkannya ke service.

service: Service yang memilikinya.
uuid: UUID karakteristik.
read, write, write_no_response, notify, indicate: Boolean yang memilih operasi GATT yang didukung.
initial: Nilai awal opsional (bytes).
capture: Jika True, nilai yang ditulis diantrekan (hingga 10 dalam) sehingga penulisan berturut-turut yang cepat tidak hilang. Setiap panggilan written kemudian mengembalikan tuple (connection, data).

uuid¶: UUID karakteristik.

flags¶: Bitmask dari flag properti GATT yang dibangun dari konstruktor.

descriptors¶: Daftar objek Descriptor yang terikat ke karakteristik ini.

read() → bytes¶: Baca nilai saat ini dari database GATT lokal.

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

Perbarui nilai dalam database GATT lokal.

data: Byte nilai baru.
send_update: Jika True, juga notify/indicate setiap koneksi yang berlangganan.

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

Kirim GATT Notify ke connection.

connection: Koneksi client target.
data: Payload yang akan dikirim. Jika None, nilai lokal saat ini yang dikirim.

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

Async. Kirim GATT Indicate ke connection dan tunggu konfirmasi client. Memunculkan GattError pada status bukan nol.

connection: Koneksi client target.
data: Payload yang akan diindikasikan, atau None untuk mengirim nilai lokal.
timeout_ms: Waktu maksimum untuk menunggu konfirmasi.

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

Async. Tunggu penulisan jarak jauh. Mengembalikan DeviceConnection yang menulis, atau (connection, data) jika karakteristik dibuat dengan capture=True.

timeout_ms: Waktu maksimum untuk menunggu. None menunggu selamanya.

on_read(connection: DeviceConnection) → int¶: Hook override yang dipanggil secara sinkron ketika bacaan jarak jauh diterima. Kembalikan 0 untuk mengizinkan bacaan atau kode kesalahan ATT bukan nol untuk menolaknya. Implementasi default mengembalikan 0.

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

Characteristic yang buffer GATT-nya dapat dikonfigurasi. Berguna untuk menerima nilai yang lebih besar dari ukuran atribut default, atau untuk mengantrekan penulisan berturut-turut.

max_len: Ukuran buffer, dalam byte.
append: Jika True, penulisan sekuensial ditambahkan ke buffer alih-alih menimpa.

Argumen lain diteruskan ke Characteristic.

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

Deskriptor GATT lokal. Membangunnya secara otomatis menambahkannya ke characteristic. Mewarisi read, write dan written dari Characteristic.

characteristic: Characteristic yang memilikinya.
uuid: UUID deskriptor.
read, write: Boolean yang memilih operasi GATT yang didukung.
initial: Nilai awal opsional (bytes).

class aioble.ClientService¶

Layanan GATT jarak jauh yang ditemukan pada peer. Dikembalikan oleh DeviceConnection.service() atau diiterasi dari DeviceConnection.services().

Jangan buat secara langsung.

connection¶: DeviceConnection yang memilikinya.

uuid¶: UUID layanan jarak jauh.

characteristic(uuid: bluetooth.UUID, timeout_ms: int = 2000) → Awaitable[ClientCharacteristic | None]¶: Async. Temukan satu karakteristik berdasarkan UUID, atau None jika tidak ditemukan.

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

Kembalikan iterator async dari objek ClientCharacteristic. Gunakan dengan async for dan jalankan loop hingga selesai.

uuid: Filter UUID opsional.
timeout_ms: Timeout per penemuan.

class aioble.ClientCharacteristic¶

Karakteristik GATT jarak jauh yang ditemukan pada peer. Dikembalikan oleh ClientService.characteristic() atau diiterasi dari ClientService.characteristics().

Jangan buat secara langsung.

service¶: ClientService yang memilikinya.

uuid¶: UUID karakteristik.

properties¶: Bitmask operasi GATT yang didukung sebagaimana dilaporkan oleh peer.

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

Async. Keluarkan GATT Read dan kembalikan nilainya. Memunculkan GattError pada status bukan nol.

timeout_ms: Timeout baca.

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

Async. Keluarkan GATT Write.

data: Nilai yang akan ditulis.
response: True untuk memerlukan write-response (dan memunculkan GattError pada kegagalan). False untuk write-without-response. None (default) memilih secara otomatis berdasarkan apa yang diiklankan peer.
timeout_ms: Timeout tulis (hanya relevan jika response adalah True).

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

Async. Tunggu notifikasi berikutnya pada karakteristik ini dan kembalikan payloadnya. Langsung mengembalikan jika notifikasi sudah diantrekan.

timeout_ms: Waktu maksimum untuk menunggu. None menunggu selamanya.

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

Async. Tunggu indikasi berikutnya pada karakteristik ini dan kembalikan payloadnya.

timeout_ms: Waktu maksimum untuk menunggu.

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

Async. Tulis Client Characteristic Configuration Descriptor (CCCD) untuk berlangganan (atau berhenti berlangganan) untuk notifikasi dan/atau indikasi.

notify: Aktifkan notifikasi.
indicate: Aktifkan indikasi.

descriptor(uuid: bluetooth.UUID, timeout_ms: int = 2000) → Awaitable[ClientDescriptor | None]¶: Async. Temukan satu deskriptor berdasarkan UUID, atau None jika tidak ditemukan.

descriptors(timeout_ms: int = 2000) → ClientDiscover¶: Kembalikan iterator async dari objek ClientDescriptor. Gunakan dengan async for dan jalankan loop hingga selesai.

class aioble.ClientDescriptor¶

Deskriptor GATT jarak jauh yang ditemukan pada peer. Mewarisi read dan write dari ClientCharacteristic.

Jangan buat secara langsung.

characteristic¶: ClientCharacteristic yang memilikinya.

uuid¶: UUID deskriptor.

class aioble.L2CAPChannel¶

Kanal L2CAP berorientasi koneksi yang aktif. Dikembalikan oleh DeviceConnection.l2cap_accept() atau DeviceConnection.l2cap_connect(). Mendukung penggunaan sebagai context manager async with yang memutus koneksi secara otomatis saat keluar.

Jangan buat secara langsung.

our_mtu¶: Ukuran maksimum, dalam byte, yang dapat dikirim peer ke kita dalam satu SDU.

peer_mtu¶: Ukuran maksimum, dalam byte, yang dapat kita kirim ke peer dalam satu SDU.

available() → bool¶: Kembalikan True secara sinkron jika data penerimaan yang di-buffer siap (yaitu recvinto tidak akan memblokir).

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

Async. Terima ke dalam buf, mengembalikan jumlah byte yang dibaca. Menunggu data baru jika kanal kosong.

buf: Buffer yang telah dialokasikan sebelumnya untuk diisi.
timeout_ms: Waktu maksimum untuk menunggu. None menunggu selamanya.

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

Async. Kirim buf pada kanal, memfragmentasi payload yang lebih besar menjadi potongan berukuran MTU. Menunggu kredit flow-control sesuai kebutuhan.

buf: Objek bytes-like untuk dikirim.
timeout_ms: Waktu maksimum untuk menunggu per potongan.
chunk_size: Override opsional untuk ukuran potongan per panggilan. Dibatasi ke min(our_mtu * 2, peer_mtu).

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

Async. Tunggu hingga send yang terhenti telah dikuras oleh controller.

timeout_ms: Waktu maksimum untuk menunggu.

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

Async. Putuskan kanal dan tunggu IRQ pemutusan koneksi.

timeout_ms: Waktu maksimum untuk menunggu.

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

Async. Tunggu hingga kanal diputuskan oleh salah satu pihak.

timeout_ms: Waktu maksimum untuk menunggu.

aioble --- Async BLE¶

Contoh¶

Fungsi tingkat modul¶

Konstanta tingkat modul¶

Pengecualian¶

Kelas¶

`aioble` --- Async BLE¶