`aioble` --- BLE غير المتزامن#

aioble هو غلاف عالي المستوى ومتوافق مع asyncio حول وحدة bluetooth. يوفر دوال coroutine نظيفة للمسح والاتصال والإعلان وخدمات GATT وقنوات L2CAP.

جميع العمليات البعيدة (الاتصال، قطع الاتصال، القراءة/الكتابة من العميل، الإشارة من الخادم، استقبال/إرسال l2cap، الإقران) قابلة للانتظار (awaitable) وتدعم المهل الزمنية.

الأدوار المدعومة:

المُذيع (المُعلِن) --- توليد حمولات الإعلان والاستجابة للمسح للحقول الشائعة، تقسيم الحمولة تلقائياً بين الإعلان واستجابة المسح، الإعلان إلى أجل غير مسمى أو لمدة محددة.
الطرفية (Peripheral) --- انتظار الاتصال من جهاز مركزي، انتظار تبادل MTU.
المراقب (الماسح) --- المسح السلبي والنشط، دمج حمولات الإعلان والاستجابة للمسح لنفس الجهاز، تحليل الحقول الشائعة من حمولات الإعلان.
المركزي (Central) --- الاتصال بجهاز طرفي، بدء تبادل 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#

يُرجع مدير سياق غير متزامن / مُكرِّراً غير متزامن من نوع scan يُنتج نسخ ScanResult لكل جهاز فريد يُكتشف (أو لكل قطعة جديدة من بيانات الإعلان من جهاز معروف).

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#

دالة coroutine غير متزامنة تبدأ الإعلان وتنتظر اتصالاً وارداً من جهاز مركزي. تُرجع DeviceConnection يمثل الجهاز المركزي المتصل، أو تطلق asyncio.TimeoutError عند انتهاء المهلة.

interval_us: فترة الإعلان بالميكرو ثانية.
adv_data: حمولة الإعلان الخام. إذا لم تُحدَّد، تُبنى adv_data من الوسائط المفتاحية المتبقية.
resp_data: حمولة استجابة المسح الخام. تُملأ تلقائياً بما يفيض من adv_data عند الحاجة.
connectable: إذا كانت True، فهذا إعلان قابل للاتصال.
limited_disc: استخدام علامة القابلية المحدودة للاكتشاف بدلاً من العامة.
br_edr: تعيين علامة دعم BR/EDR.
name: اسم محلي كامل اختياري لتضمينه.
services: كائن قابل للتكرار من bluetooth.UUID للإعلان عنه.
appearance: قيمة المظهر بـ 16 بت (انظر الأرقام المخصصة لـ Bluetooth).
manufacturer: صف (tuple) من (company_id, data_bytes) للإعلان عنه كبيانات خاصة بالمُصنِّع.
timeout_ms: إيقاف الإعلان بعد هذا العدد من الميلي ثانية دون اتصال. القيمة None تعني الإعلان حتى يحدث الاتصال.

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

تسجيل كائن Service واحد أو أكثر (وخصائصها وواصفاتها) لدى خادم GATT. يجب استدعاؤه مرة واحدة قبل بدء advertise. تستبدل الاستدعاءات اللاحقة التسجيل السابق.

services: نسخة Service واحدة أو أكثر.

ثوابت على مستوى الوحدة#

aioble.ADDR_PUBLIC#: نوع عنوان جهاز BLE العام (0).

aioble.ADDR_RANDOM#: نوع عنوان جهاز BLE العشوائي (1).

الاستثناءات#

exception aioble.GattError#: يُطلق عندما تكتمل عملية GATT بعيدة (قراءة / كتابة / إشارة) بحالة غير صفرية. يتوفر رمز الحالة في السمة _status.

exception aioble.DeviceDisconnectedError#: يُطلق داخل عملية غير متزامنة (مثل القراءة، الكتابة، الإشعار) عندما يسقط الاتصال الأساسي أثناء الانتظار.

exception aioble.L2CAPDisconnectedError#: يُطلق عند محاولة عملية إرسال/استقبال/تفريغ على قناة L2CAP مقطوعة (أو عند مقاطعتها بسببها).

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، أو سلسلة سداسية عشرية مفصولة بنقطتين (مثل "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]#

غير متزامنة. تبدأ اتصال 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 يقطع الاتصال تلقائياً عند الخروج.

لا تنشئه مباشرة.

device#: كائن Device الأساسي.

encrypted#: True بمجرد تشفير الرابط (مثلاً بعد الإقران).

authenticated#: True إذا كان الرابط مُصادَقاً عليه (إقران محمي ضد MITM).

bonded#: True إذا أنتج الإقران مفاتيح ربط (bonding).

key_size#: حجم مفتاح التشفير المتفاوض عليه بالبايتات، أو False إذا لم يكن مشفراً.

mtu#: قيمة ATT MTU المتفاوض عليها بعد exchange_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]#

غير متزامنة. تنتظر إنهاء الاتصال من قبل أي من الطرفين. إذا كانت 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: قيمة MTU المفضلة الاختيارية لتعيينها على واجهة BLE الأساسية قبل التبادل.
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: إبقاء مفاتيح الإقران (persist).
le_secure: استخدام LE Secure Connections.
mitm: اشتراط الحماية من هجمات الوسيط (man-in-the-middle).
io: ثابت قدرة الإدخال/الإخراج (مثل 3 لعدم وجود إدخال/إخراج).
timeout_ms: مهلة الإقران.

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

غير متزامنة. تستمع على الـ PSM المحدد وتُرجع L2CAPChannel بمجرد فتح الطرف البعيد لها.

psm: مُجمِّع البروتوكول/الخدمة (PSM) للاستماع عليه.
mtu: الحد الأقصى لحجم الاستقبال بالبايتات.
timeout_ms: الحد الأقصى لزمن انتظار اتصال الطرف البعيد.

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

غير متزامنة. تفتح قناة L2CAP إلى الطرف البعيد على الـ PSM المحدد.

psm: مُجمِّع البروتوكول/الخدمة (PSM) للاتصال به.
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]#: مولّد (generator) يُنتج كل bluetooth.UUID مُعلَن عنه في حقول قائمة الخدمات بـ 16/32/128 بت.

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

مولّد (generator) يُنتج صفوف (tuples) (company_id, data) من حقول الإعلان الخاصة بالمُصنِّع.

filter: إذا أُعطي، تُنتَج فقط الإدخالات التي يتطابق معرّف شركتها.

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: قيم منطقية (Booleans) تختار عمليات GATT المدعومة.
initial: قيمة أولية اختيارية (bytes).
capture: إذا كانت True، تُوضع القيم المكتوبة في طابور (حتى عمق 10) بحيث لا تُفقد عمليات الكتابة المتتالية السريعة. ثم يُرجع كل استدعاء لـ written صفاً (tuple) من نوع (connection, data).

uuid#: معرّف UUID للخاصية.

flags#: قناع بِتّي (bitmask) لعلامات خاصية 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#

إرسال GATT Notify إلى connection.

connection: اتصال العميل المستهدف.
data: الحمولة المراد إرسالها. إذا كانت None، تُرسل القيمة المحلية الحالية.

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

غير متزامنة. ترسل GATT Indicate إلى connection وتنتظر تأكيد العميل. تطلق GattError عند حالة غير صفرية.

connection: اتصال العميل المستهدف.
data: الحمولة المراد الإشارة إليها، أو None لإرسال القيمة المحلية.
timeout_ms: الحد الأقصى لزمن انتظار التأكيد.

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

غير متزامنة. تنتظر عملية كتابة بعيدة. تُرجع DeviceConnection الكاتب، أو (connection, data) إذا أُنشئت الخاصية بـ capture=True.

timeout_ms: الحد الأقصى لزمن الانتظار. القيمة None تنتظر إلى الأبد.

on_read(connection: DeviceConnection) → int#: خطّاف تجاوز (override) يُستدعى تزامنياً عند استقبال قراءة بعيدة. أرجع 0 للسماح بالقراءة أو رمز خطأ ATT غير صفري لرفضها. التنفيذ الافتراضي يُرجع 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 محلي. يؤدي إنشاء واحد تلقائياً إلى إلحاقه بـ characteristic. يرث read و write و written من Characteristic.

characteristic: كائن Characteristic المالك.
uuid: معرّف UUID للواصف.
read, write: قيم منطقية (Booleans) تختار عمليات 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#: قناع بِتّي (bitmask) لعمليات GATT المدعومة كما أبلغ عنها النظير.

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

غير متزامنة. تُصدر GATT Read وتُرجع القيمة. تطلق GattError عند حالة غير صفرية.

timeout_ms: مهلة القراءة.

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

غير متزامنة. تُصدر GATT Write.

data: القيمة المراد كتابتها.
response: True لاشتراط استجابة-كتابة (وإطلاق GattError عند الفشل). False للكتابة-بدون-استجابة. None (الافتراضية) تختار تلقائياً بناءً على ما يُعلنه النظير.
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]#

غير متزامنة. تكتب واصف إعداد خاصية العميل (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 بعيد اكتُشف على نظير. يرث read و write من ClientCharacteristic.

لا تنشئه مباشرة.

characteristic#: كائن ClientCharacteristic المالك.

uuid#: معرّف UUID للواصف.

class aioble.L2CAPChannel#

قناة L2CAP نشطة موجهة للاتصال. تُرجَع بواسطة DeviceConnection.l2cap_accept() أو DeviceConnection.l2cap_connect(). تدعم الاستخدام كمدير سياق async with يقطع الاتصال تلقائياً عند الخروج.

لا تنشئه مباشرة.

our_mtu#: الحد الأقصى للحجم بالبايتات الذي يمكن للنظير إرساله إلينا في وحدة SDU واحدة.

peer_mtu#: الحد الأقصى للحجم بالبايتات الذي يمكننا إرساله إلى النظير في وحدة SDU واحدة.

available() → bool#: تُرجع تزامنياً True إذا كانت بيانات الاستقبال المخزنة جاهزة (أي أن recvinto لن تتوقف للانتظار).

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

غير متزامنة. تستقبل في buf، وتُرجع عدد البايتات المقروءة. تنتظر بيانات جديدة إذا كانت القناة فارغة.

buf: مخزن مؤقت مخصص مسبقاً لملئه.
timeout_ms: الحد الأقصى لزمن الانتظار. القيمة None تنتظر إلى الأبد.

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

غير متزامنة. ترسل buf على القناة، مع تجزئة الحمولات الأكبر إلى أجزاء بحجم MTU. تنتظر أرصدة التحكم في التدفق حسب الحاجة.

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: الحد الأقصى لزمن الانتظار.

aioble --- BLE غير المتزامن#

أمثلة#

دوال على مستوى الوحدة#

ثوابت على مستوى الوحدة#

الاستثناءات#

الأصناف#

`aioble` --- BLE غير المتزامن#