`aioble` --- Async BLE¶

aioble เป็น wrapper ระดับสูงที่รองรับ asyncio สำหรับโมดูล bluetooth มีคอรูทีนที่สะอาดสำหรับการสแกน การเชื่อมต่อ การโฆษณา บริการ GATT และช่องทาง L2CAP

การดำเนินการระยะไกลทั้งหมด (connect, disconnect, client read/write, server indicate, l2cap recv/send, pair) รองรับ await และรองรับการหมดเวลา

บทบาทที่รองรับ:

Broadcaster (advertiser) --- สร้างเพย์โหลดโฆษณาและ scan-response สำหรับฟิลด์ทั่วไป แบ่งเพย์โหลดอัตโนมัติระหว่างการโฆษณาและ scan response โฆษณาอย่างไม่สิ้นสุดหรือตามระยะเวลาที่กำหนด
Peripheral --- รอการเชื่อมต่อจาก central รอการแลกเปลี่ยน MTU
Observer (scanner) --- การสแกนแบบ passive และ active รวมเพย์โหลดโฆษณาและ scan-response ของอุปกรณ์เดียวกัน แยกวิเคราะห์ฟิลด์ทั่วไปจากเพย์โหลดโฆษณา
Central --- เชื่อมต่อกับ peripheral เริ่มการแลกเปลี่ยน MTU
GATT Client --- ค้นพบบริการ / ลักษณะเฉพาะ / ตัวบ่งชี้ (เลือกตาม UUID ได้); อ่าน / เขียน / เขียนพร้อม response บน characteristics และ descriptors; สมัครรับการแจ้งเตือนและ indications (ผ่าน CCCD); รอรับการแจ้งเตือนและ indications
GATT Server --- ลงทะเบียนบริการ / ลักษณะเฉพาะ / ตัวบ่งชี้; รอรับการเขียนบน characteristics และ descriptors; ดักจับคำขออ่าน; ส่งการแจ้งเตือนและ indications (และรอการตอบสนอง)
L2CAP --- รับและเชื่อมต่อช่องทาง L2CAP แบบ connection-oriented จัดการ flow control ของช่องทาง
Security --- การจัดการ key/secret ด้วย JSON รองรับ pairing เริ่มต้น ตรวจสอบสถานะการเข้ารหัส / การตรวจสอบตัวตน

ตัวอย่าง¶

สแกนหาอุปกรณ์ 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())

เชื่อมต่อกับ peripheral ที่โฆษณาบริการ Heart Rate ในฐานะ central และสมัครรับ measurement notifications:

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

ทำหน้าที่เป็น peripheral: ลงทะเบียนบริการ GATT โฆษณา และส่ง notifications ไปยังผู้เชื่อมต่อ:

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 ที่อยู่เบื้องหลังและเรียกใช้ตัวจัดการการปิดระบบของ sub-module ที่ลงทะเบียนไว้ หลังจากเรียกใช้ฟังก์ชันนี้ scanners, advertisers, connections และช่องทาง L2CAP ทั้งหมดจะถูกปิดลง

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

ส่งคืน scan async context-manager / async-iterator ที่ yield อินสแตนซ์ ScanResult สำหรับแต่ละอุปกรณ์ที่ค้นพบ (หรือสำหรับข้อมูลโฆษณาใหม่แต่ละชุดจากอุปกรณ์ที่รู้จัก)

duration_ms: ระยะเวลาการสแกน ในหน่วยมิลลิวินาที ส่ง 0 เพื่อสแกนอย่างไม่สิ้นสุดจนกว่า context manager จะออก
interval_us: ช่วงเวลาการสแกน ในหน่วยไมโครวินาที ค่าเริ่มต้นคือ 1,280,000
window_us: หน้าต่างการสแกน ในหน่วยไมโครวินาที (ต้องน้อยกว่าหรือเท่ากับ interval_us) ค่าเริ่มต้นคือ 11,250
active: ถ้าเป็น True จะทำการสแกนแบบ active (ร้องขอข้อมูล scan response) ค่าเริ่มต้นคือ 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¶

Async coroutine ที่เริ่มโฆษณาและรอการเชื่อมต่อ central ที่เข้ามา ส่งคืน DeviceConnection ที่แทน central ที่เชื่อมต่อ หรือ raise asyncio.TimeoutError เมื่อหมดเวลา

interval_us: ช่วงเวลาการโฆษณา ในหน่วยไมโครวินาที
adv_data: เพย์โหลดโฆษณาดิบ ถ้าไม่ได้ตั้งค่า adv_data จะถูกสร้างจากอาร์กิวเมนต์คีย์เวิร์ดที่เหลือ
resp_data: เพย์โหลด scan response ดิบ จะถูกเติมอัตโนมัติจาก adv_data ที่เกินขนาดหากจำเป็น
connectable: ถ้าเป็น True จะเป็นการโฆษณาแบบ connectable
limited_disc: ใช้ flag limited-discoverable แทน general
br_edr: ตั้งค่า flag BR/EDR-supported
name: ชื่อเฉพาะในพื้นที่แบบสมบูรณ์ที่ต้องการฝังไว้ (ไม่บังคับ)
services: Iterable ของ bluetooth.UUID ที่ต้องการโฆษณา
appearance: ค่า appearance 16 บิต (ดู Bluetooth assigned numbers)
manufacturer: Tuple ของ (company_id, data_bytes) สำหรับโฆษณาเป็นข้อมูลเฉพาะผู้ผลิต
timeout_ms: หยุดโฆษณาหลังจากผ่านไปตามจำนวนมิลลิวินาทีนี้โดยไม่มีการเชื่อมต่อ None หมายถึงโฆษณาจนกว่าจะเชื่อมต่อ

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

ลงทะเบียนออบเจ็กต์ Service หนึ่งรายการขึ้นไป (พร้อมกับ characteristics และ descriptors) กับ GATT server ต้องเรียกครั้งเดียวก่อนเริ่ม advertise การเรียกครั้งถัดไปจะแทนที่การลงทะเบียนก่อนหน้า

services: อินสแตนซ์ Service หนึ่งรายการขึ้นไป

ค่าคงที่ระดับโมดูล¶

aioble.ADDR_PUBLIC¶: ประเภทที่อยู่อุปกรณ์ BLE แบบสาธารณะ (0)

aioble.ADDR_RANDOM¶: ประเภทที่อยู่อุปกรณ์ BLE แบบสุ่ม (1)

ข้อยกเว้น¶

exception aioble.GattError¶: Raise เมื่อการดำเนินการ GATT ระยะไกล (read / write / indicate) เสร็จสิ้นด้วยสถานะที่ไม่ใช่ศูนย์ รหัสสถานะสามารถเข้าถึงได้ที่แอตทริบิวต์ _status

exception aioble.DeviceDisconnectedError¶: Raise ภายใน async operation (เช่น read, write, notified) เมื่อการเชื่อมต่อที่อยู่เบื้องหลังหลุดออกขณะรอ

exception aioble.L2CAPDisconnectedError¶: Raise เมื่อพยายามดำเนินการ send/recv/flush ของช่องทาง L2CAP บน (หรือถูกขัดจากการ) ช่องทางที่ยกเลิกการเชื่อมต่อแล้ว

exception aioble.L2CAPConnectionError¶: Raise โดย DeviceConnection.l2cap_connect เมื่อการสร้างช่องทางล้มเหลว รหัสสถานะ Bluetooth คืออาร์กิวเมนต์แรก

คลาส¶

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

แทนอุปกรณ์ BLE ระยะไกลด้วยที่อยู่ อินสแตนซ์ Device สองรายการถือว่าเท่ากันหากทั้ง addr_type และ addr ตรงกัน ใช้เป็น handle สำหรับเริ่มการเชื่อมต่อ

addr_type: ต้องเป็น ADDR_PUBLIC หรือ ADDR_RANDOM
addr: ที่อยู่หกไบต์เป็น bytes หรือสตริงเลขฐานสิบหกที่คั่นด้วยโคลอน (เช่น "aa:bb:cc:dd:ee:ff")

addr_type¶: ประเภทที่อยู่ที่ใช้สร้างอุปกรณ์

addr¶: ที่อยู่อุปกรณ์หกไบต์ดิบ

addr_hex() → str¶: ส่งคืนที่อยู่ในรูปแบบสตริงเลขฐานสิบหกที่คั่นด้วยโคลอน

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 เริ่มต้นการเชื่อมต่อ GAP กับอุปกรณ์นี้และส่งคืน DeviceConnection ที่ได้ ยกเลิกการสแกนที่กำลังดำเนินอยู่

timeout_ms: ระยะเวลารอการเชื่อมต่อให้เสร็จสมบูรณ์
scan_duration_ms: ระยะเวลาการสแกนเริ่มต้นก่อนเชื่อมต่อ (ขึ้นอยู่กับคอนโทรลเลอร์)
min_conn_interval_us / max_conn_interval_us: ขอบเขตช่วงเวลาการเชื่อมต่อ ในหน่วยไมโครวินาที (ไม่บังคับ)

class aioble.DeviceConnection¶

การเชื่อมต่อ GAP ที่ใช้งานอยู่กับ Device ส่งคืนโดย Device.connect() หรือ advertise รองรับการใช้เป็น async with context manager ที่ยกเลิกการเชื่อมต่ออัตโนมัติเมื่อออก

อย่าสร้างโดยตรง

device¶: Device เบื้องหลัง

encrypted¶: True เมื่อลิงก์เข้ารหัสแล้ว (เช่น หลัง pairing)

authenticated¶: True ถ้าลิงก์ได้รับการตรวจสอบตัวตนแล้ว (MITM-protected pair)

bonded¶: True ถ้า pairing สร้าง bonding keys

key_size¶: ขนาด encryption key ที่เจรจาแล้วในหน่วยไบต์ หรือ False ถ้าไม่มีการเข้ารหัส

mtu¶: ATT MTU ที่เจรจาแล้วหลัง exchange_mtu หรือ None จนกว่าจะถูกตั้งค่า

is_connected() → bool¶: ส่งคืนว่าการเชื่อมต่อยังใช้งานอยู่หรือไม่

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

Async ยกเลิกการเชื่อมต่อและรอ disconnection IRQ

timeout_ms: เวลาสูงสุดที่รอการยกเลิกการเชื่อมต่อ

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

Async รอให้การเชื่อมต่อถูกยุติโดยฝ่ายใดฝ่ายหนึ่ง ถ้า disconnect เป็น True จะยกเลิกการเชื่อมต่ออย่างแข็งขันก่อน

timeout_ms: เวลาสูงสุดที่รอ None หมายถึงรอตลอดไป
disconnect: ถ้าเป็น True จะเริ่มต้นการยกเลิกการเชื่อมต่อ

timeout(timeout_ms: int | None) → DeviceTimeout¶

ส่งคืน context manager ที่ยกเลิก body ถ้าหมดเวลา (raise asyncio.TimeoutError) หรืออุปกรณ์ยกเลิกการเชื่อมต่อ (raise DeviceDisconnectedError)

timeout_ms: เวลาหมดอายุในมิลลิวินาที หรือ None สำหรับไม่มีเวลาหมดอายุ

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

Async เริ่มการแลกเปลี่ยน ATT MTU และส่งคืน MTU ที่เจรจาแล้ว

mtu: MTU ที่ต้องการตั้งค่าบนอินเทอร์เฟซ BLE เบื้องหลังก่อนการแลกเปลี่ยน (ไม่บังคับ)
timeout_ms: เวลาหมดอายุสำหรับการแลกเปลี่ยน

service(uuid: bluetooth.UUID, timeout_ms: int = 2000) → Awaitable[ClientService | None]¶: Async ค้นพบบริการระยะไกลเดี่ยวที่ตรงกับ uuid หรือ None ถ้าไม่พบ

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

ส่งคืน async iterator ของออบเจ็กต์ 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]¶

Async เริ่มต้น pairing บนการเชื่อมต่อนี้ อัปเดตแอตทริบิวต์ encrypted / authenticated / bonded / key_size เมื่อเสร็จสมบูรณ์

bond: บันทึก pairing keys อย่างถาวร
le_secure: ใช้ LE Secure Connections
mitm: ต้องการการป้องกัน man-in-the-middle
io: ค่าคงที่ความสามารถ IO (เช่น 3 สำหรับไม่มี input/output)
timeout_ms: เวลาหมดอายุของ pairing

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

Async ฟังบน PSM ที่กำหนดและส่งคืน L2CAPChannel เมื่อ remote เปิดใช้งาน

psm: Protocol/Service Multiplexer ที่ต้องการฟัง
mtu: ขนาดรับสูงสุด ในหน่วยไบต์
timeout_ms: เวลาสูงสุดที่รอให้ remote เชื่อมต่อ

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

Async เปิดช่องทาง L2CAP ไปยัง remote บน PSM ที่กำหนด

psm: Protocol/Service Multiplexer ที่ต้องการเชื่อมต่อ
mtu: ขนาดรับสูงสุด ในหน่วยไบต์
timeout_ms: เวลาหมดอายุการเชื่อมต่อ

class aioble.ScanResult¶

อุปกรณ์เดี่ยวที่ค้นพบระหว่าง scan อินสแตนซ์เดิมจะถูก yield ซ้ำเมื่อข้อมูลโฆษณาใหม่มาถึง

อย่าสร้างโดยตรง

device¶: Device เบื้องหลัง

rssi¶: RSSI ล่าสุดที่รายงาน ในหน่วย dBm

adv_data¶: เพย์โหลดโฆษณาดิบ (bytes หรือ None)

resp_data¶: เพย์โหลด scan response ดิบ (bytes หรือ None) ถ้าเปิดใช้งานการสแกนแบบ active

connectable¶: True ถ้าการโฆษณาล่าสุดเป็นแบบ connectable

name() → str | None¶: ถอดรหัสชื่อในพื้นที่ที่โฆษณาแบบสมบูรณ์ (หรือย่อ) จากเพย์โหลด หรือ None ถ้าไม่มี

services() → Iterator[bluetooth.UUID]¶: Generator ที่ yield bluetooth.UUID แต่ละรายการที่โฆษณาในฟิลด์รายการบริการ 16/32/128 บิต

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

Generator ที่ yield tuple (company_id, data) จากฟิลด์โฆษณาเฉพาะผู้ผลิต

filter: ถ้าระบุ จะ yield เฉพาะรายการที่ 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 characteristic ในพื้นที่ การสร้างจะเพิ่มเข้าไปใน service โดยอัตโนมัติ

service: Service ที่เป็นเจ้าของ
uuid: UUID ของ characteristic
read, write, write_no_response, notify, indicate: ค่า boolean ที่เลือกการดำเนินการ GATT ที่รองรับ
initial: ค่าเริ่มต้น (bytes) (ไม่บังคับ)
capture: ถ้าเป็น True ค่าที่เขียนจะถูกเก็บในคิว (สูงสุด 10 รายการ) เพื่อไม่ให้การเขียนต่อเนื่องกันอย่างรวดเร็วสูญหาย การเรียก written แต่ละครั้งจะส่งคืน tuple (connection, data)

uuid¶: UUID ของ characteristic

flags¶: Bitmask ของ flag คุณสมบัติ GATT ที่สร้างจาก constructor

descriptors¶: รายการออบเจ็กต์ Descriptor ที่ผูกกับ characteristic นี้

read() → bytes¶: อ่านค่าปัจจุบันจากฐานข้อมูล GATT ในพื้นที่

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

อัปเดตค่าในฐานข้อมูล GATT ในพื้นที่

data: ไบต์ค่าใหม่
send_update: ถ้าเป็น True จะแจ้งเตือน/indicate การเชื่อมต่อที่สมัครรับทุกรายการด้วย

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

ส่ง GATT Notify ไปยัง connection

connection: การเชื่อมต่อ client เป้าหมาย
data: เพย์โหลดที่ต้องการส่ง ถ้าเป็น None จะส่งค่าในพื้นที่ปัจจุบัน

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

Async ส่ง GATT Indicate ไปยัง connection และรอการยืนยันจาก client Raise GattError เมื่อสถานะไม่ใช่ศูนย์

connection: การเชื่อมต่อ client เป้าหมาย
data: เพย์โหลดที่ต้องการ indicate หรือ None เพื่อส่งค่าในพื้นที่
timeout_ms: เวลาสูงสุดที่รอการยืนยัน

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

Async รอการเขียนจากระยะไกล ส่งคืน DeviceConnection ที่เขียน หรือ (connection, data) ถ้า characteristic ถูกสร้างด้วย capture=True

timeout_ms: เวลาสูงสุดที่รอ None รอตลอดไป

on_read(connection: DeviceConnection) → int¶: Override hook ที่ถูกเรียกแบบ synchronous เมื่อได้รับคำขออ่านจากระยะไกล ส่งคืน 0 เพื่ออนุญาตการอ่าน หรือรหัสข้อผิดพลาด ATT ที่ไม่ใช่ศูนย์เพื่อปฏิเสธ การ implementation เริ่มต้นส่งคืน 0

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

Characteristic ที่มีบัฟเฟอร์ GATT สำรองที่กำหนดค่าได้ มีประโยชน์สำหรับรับค่าที่ใหญ่กว่าขนาดแอตทริบิวต์เริ่มต้น หรือสำหรับการเขียนต่อเนื่องกัน

max_len: ขนาดบัฟเฟอร์ ในหน่วยไบต์
append: ถ้าเป็น True การเขียนต่อเนื่องจะต่อท้ายในบัฟเฟอร์แทนที่จะเขียนทับ

อาร์กิวเมนต์อื่นส่งต่อไปยัง Characteristic

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

GATT descriptor ในพื้นที่ การสร้างจะเพิ่มเข้าไปใน characteristic โดยอัตโนมัติ รับช่วง read, write และ written จาก Characteristic

characteristic: Characteristic ที่เป็นเจ้าของ
uuid: UUID ของ descriptor
read, write: ค่า boolean ที่เลือกการดำเนินการ GATT ที่รองรับ
initial: ค่าเริ่มต้น (bytes) (ไม่บังคับ)

class aioble.ClientService¶

บริการ GATT ระยะไกลที่ค้นพบบน peer ส่งคืนโดย DeviceConnection.service() หรือวนซ้ำจาก DeviceConnection.services()

อย่าสร้างโดยตรง

connection¶: DeviceConnection ที่เป็นเจ้าของ

uuid¶: UUID ของบริการระยะไกล

characteristic(uuid: bluetooth.UUID, timeout_ms: int = 2000) → Awaitable[ClientCharacteristic | None]¶: Async ค้นพบ characteristic เดี่ยวตาม UUID หรือ None ถ้าไม่พบ

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

ส่งคืน async iterator ของออบเจ็กต์ ClientCharacteristic ใช้กับ async for และรันลูปจนเสร็จสมบูรณ์

uuid: ตัวกรอง UUID (ไม่บังคับ)
timeout_ms: เวลาหมดอายุต่อการค้นพบ

class aioble.ClientCharacteristic¶

GATT characteristic ระยะไกลที่ค้นพบบน peer ส่งคืนโดย ClientService.characteristic() หรือวนซ้ำจาก ClientService.characteristics()

อย่าสร้างโดยตรง

service¶: ClientService ที่เป็นเจ้าของ

uuid¶: UUID ของ characteristic

properties¶: Bitmask ของการดำเนินการ GATT ที่รองรับตามที่ peer รายงาน

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

Async ส่งคำสั่ง GATT Read และส่งคืนค่า Raise GattError เมื่อสถานะไม่ใช่ศูนย์

timeout_ms: เวลาหมดอายุการอ่าน

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

Async ส่งคำสั่ง GATT Write

data: ค่าที่ต้องการเขียน
response: True ต้องการ write-response (และ raise GattError เมื่อล้มเหลว) False สำหรับ write-without-response None (ค่าเริ่มต้น) เลือกอัตโนมัติตามที่ peer โฆษณา
timeout_ms: เวลาหมดอายุการเขียน (เกี่ยวข้องเฉพาะเมื่อ response เป็น True)

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

Async รอ notification ถัดไปบน characteristic นี้และส่งคืนเพย์โหลด ส่งคืนทันทีถ้ามี notification ในคิวอยู่แล้ว

timeout_ms: เวลาสูงสุดที่รอ None รอตลอดไป

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

Async รอ indication ถัดไปบน characteristic นี้และส่งคืนเพย์โหลด

timeout_ms: เวลาสูงสุดที่รอ

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

Async เขียน Client Characteristic Configuration Descriptor (CCCD) เพื่อสมัคร (หรือยกเลิกสมัคร) รับ notifications และ/หรือ indications

notify: เปิดใช้งาน notifications
indicate: เปิดใช้งาน indications

descriptor(uuid: bluetooth.UUID, timeout_ms: int = 2000) → Awaitable[ClientDescriptor | None]¶: Async ค้นพบ descriptor เดี่ยวตาม UUID หรือ None ถ้าไม่พบ

descriptors(timeout_ms: int = 2000) → ClientDiscover¶: ส่งคืน async iterator ของออบเจ็กต์ ClientDescriptor ใช้กับ async for และรันลูปจนเสร็จสมบูรณ์

class aioble.ClientDescriptor¶

GATT descriptor ระยะไกลที่ค้นพบบน peer รับช่วง read และ write จาก ClientCharacteristic

อย่าสร้างโดยตรง

characteristic¶: ClientCharacteristic ที่เป็นเจ้าของ

uuid¶: UUID ของ descriptor

class aioble.L2CAPChannel¶

ช่องทาง L2CAP แบบ connection-oriented ที่ใช้งานอยู่ ส่งคืนโดย DeviceConnection.l2cap_accept() หรือ DeviceConnection.l2cap_connect() รองรับการใช้เป็น async with context manager ที่ยกเลิกการเชื่อมต่ออัตโนมัติเมื่อออก

อย่าสร้างโดยตรง

our_mtu¶: ขนาดสูงสุด ในหน่วยไบต์ ที่ peer อาจส่งถึงเราใน SDU เดี่ยว

peer_mtu¶: ขนาดสูงสุด ในหน่วยไบต์ ที่เราอาจส่งถึง peer ใน SDU เดี่ยว

available() → bool¶: ส่งคืน True แบบ synchronous ถ้าข้อมูลที่รับบัฟเฟอร์ไว้พร้อม (กล่าวคือ recvinto จะไม่บล็อก)

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

Async รับเข้า buf ส่งคืนจำนวนไบต์ที่อ่าน รอข้อมูลใหม่ถ้าช่องทางว่างเปล่า

buf: บัฟเฟอร์ที่จัดสรรล่วงหน้าสำหรับเติมข้อมูล
timeout_ms: เวลาสูงสุดที่รอ None รอตลอดไป

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

Async ส่ง buf บนช่องทาง แบ่ง payload ขนาดใหญ่เป็น chunk ขนาด MTU รอ flow-control credits ตามต้องการ

buf: ออบเจ็กต์คล้าย bytes ที่ต้องการส่ง
timeout_ms: เวลาสูงสุดที่รอต่อ chunk
chunk_size: Override ขนาด chunk ต่อการเรียกที่ต้องการ จำกัดไว้ที่ min(our_mtu * 2, peer_mtu)

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

Async รอจนกว่า send ที่ค้างอยู่จะถูก drain โดยคอนโทรลเลอร์

timeout_ms: เวลาสูงสุดที่รอ

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

Async ยกเลิกการเชื่อมต่อช่องทางและรอ disconnection IRQ

timeout_ms: เวลาสูงสุดที่รอ

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

Async รอจนกว่าช่องทางจะถูกยกเลิกการเชื่อมต่อโดยฝ่ายใดฝ่ายหนึ่ง

timeout_ms: เวลาสูงสุดที่รอ

aioble --- Async BLE¶

ตัวอย่าง¶

ฟังก์ชันระดับโมดูล¶

ค่าคงที่ระดับโมดูล¶

ข้อยกเว้น¶

คลาส¶

`aioble` --- Async BLE¶