基于adafruit_ble和Bleak实现无延迟BLE UART数据传输的可靠性优化及扩展方案咨询

Hey there! Let's tackle your BLE communication headaches step by step—fixing reliability, ditching those forced sleeps, squashing reconnection errors, and prepping for larger data payloads.

1. Ditch the asyncio.sleep(): Implement a Proper Acknowledgement Mechanism #

The 5-second sleep is a band-aid because your sender is guessing when the receiver will respond, instead of listening actively. Let's replace that with Bleak's notification feature to react instantly when the receiver sends data:

Modified Sender Code (Bleak on Windows) #

import asyncio
from bleak import BleakClient, BleakScanner

async def run(write_value):
    mac_addr = "XX:XX:XX:XX:XX"
    tx_charac = "X"  # Replace with your actual TX characteristic UUID
    rx_charac = "X"  # Replace with your actual RX characteristic UUID
    
    # First, confirm the device is discoverable to avoid connection failures
    device = await BleakScanner.find_device_by_address(mac_addr, timeout=10)
    if not device:
        print("Device not found!")
        return None

    # Use a future to wait for the response without hardcoded delays
    response_received = asyncio.Future()

    def notification_handler(sender, data):
        # Triggered immediately when the receiver sends data
        response_received.set_result(data)

    async with BleakClient(device, timeout=30) as client:
        # Start listening for responses BEFORE sending data
        await client.start_notify(rx_charac, notification_handler)
        # Send data with response=True to ensure delivery confirmation from BLE stack
        await client.write_gatt_char(tx_charac, bytearray(write_value), response=True)
        # Wait until we get the receiver's ack (no sleep needed!)
        answer = await response_received
        await client.stop_notify(rx_charac)
        print(f"Received response: {answer}")
    
    return answer

# Test with your 15-element list
x = list(range(15))  # Replace with your actual data
answer = asyncio.run(run(x))

Modified Receiver Code (CircuitPython on ItsyBitsy nRF52840) #

Your original receiver echoes partial data as it arrives—let's wait for the full payload before sending an acknowledgement:

from adafruit_ble import BLERadio
from adafruit_ble.advertising.standard import ProvideServicesAdvertisement
from adafruit_ble.services.nordic import UARTService

ble = BLERadio()
uart = UARTService()
advertisement = ProvideServicesAdvertisement(uart)

EXPECTED_LENGTH = 15  # Adjust this later when sending larger data

while True:
    print("Starting BLE advertisement...")
    ble.start_advertising(advertisement)
    print("Waiting for central device connection...")
    while not ble.connected:
        pass
    print("Connected!")
    
    # Buffer to collect incoming data until full payload is received
    data_buffer = bytearray()
    while ble.connected:
        # Read available bytes (non-blocking)
        incoming = uart.read(EXPECTED_LENGTH)
        if incoming:
            data_buffer.extend(incoming)
            # Check if we've received the full expected payload
            if len(data_buffer) >= EXPECTED_LENGTH:
                full_payload = data_buffer[:EXPECTED_LENGTH]
                # Clear leftover bytes for next transmission
                data_buffer = data_buffer[EXPECTED_LENGTH:]
                
                # Process your data here
                sequence = [x for x in full_payload]
                print(f"Received full sequence: {sequence}")
                
                # Send a confirmation back to the sender
                uart.write(full_payload)
    
    print("Disconnected. Restarting advertisement...")

2. Fix the Reconnection AttributeError #

That AttributeError: 'NoneType' object has no attribute 'bluetooth_address' pops up when Bleak tries to connect to a device it can't find. Here's how to fix it:

• Always scan first: As shown in the sender code, use BleakScanner.find_device_by_address() to confirm the device is online before connecting.
• Add retry logic: If the first connection fails, retry 2-3 times instead of bailing immediately.
• Avoid duplicate advertising: On the receiver, only call ble.start_advertising() once per disconnected state—your original code calls it every loop iteration, which can cause advertising conflicts.

3. Prepare for Larger Data Payloads (e.g., 100 Elements) #

BLE UART has a default MTU (Maximum Transmission Unit) of ~20 bytes, so you can't send 100 bytes in one go. Let's implement a simple packetization protocol:

Protocol Idea #

1. Packet structure: Each packet starts with 2 bytes for total payload length, 1 byte for packet number, followed by up to 17 bytes of data (3 bytes of overhead).
2. Sender: Split your long list into chunks, add headers, send packets one by one.
3. Receiver: Collect all packets, reassemble into the full payload, then send a final acknowledgement.

Example Sender Packetization #

def split_into_packets(data_list, mtu=20):
    # MTU minus 3 bytes for header (total length + packet number)
    chunk_size = mtu - 3
    total_length = len(data_list)
    packets = []
    for i in range(0, total_length, chunk_size):
        chunk = data_list[i:i+chunk_size]
        # Header: 2 bytes for total length, 1 byte for packet index
        header = total_length.to_bytes(2, byteorder='little') + bytes([i//chunk_size])
        packet = header + bytearray(chunk)
        packets.append(packet)
    return packets

# Usage in the run() function
x = list(range(100))  # Your 100-element list
packets = split_into_packets(x)
for packet in packets:
    await client.write_gatt_char(tx_charac, packet, response=True)

Example Receiver Reassembly #

from adafruit_ble import BLERadio
from adafruit_ble.advertising.standard import ProvideServicesAdvertisement
from adafruit_ble.services.nordic import UARTService

ble = BLERadio()
uart = UARTService()
advertisement = ProvideServicesAdvertisement(uart)

while True:
    ble.start_advertising(advertisement)
    while not ble.connected:
        pass
    print("Connected!")
    
    full_data = None
    total_length = 0
    received_chunks = {}
    
    while ble.connected:
        incoming = uart.read(20)  # Read up to MTU size
        if incoming and len(incoming) >= 3:
            # Parse header
            total_length = int.from_bytes(incoming[:2], byteorder='little')
            packet_num = incoming[2]
            chunk = incoming[3:]
            # Store the chunk
            received_chunks[packet_num] = chunk
            # Check if all chunks are received
            if len(received_chunks) * (20-3) >= total_length:
                # Reassemble full data
                full_data = bytearray()
                for i in range(len(received_chunks)):
                    full_data.extend(received_chunks[i])
                # Trim to exact total length
                full_data = full_data[:total_length]
                print(f"Received full data (length {total_length}): {[x for x in full_data]}")
                # Send final acknowledgement
                uart.write(b"SUCCESS")
                # Reset for next transmission
                full_data = None
                total_length = 0
                received_chunks = {}
    
    print("Disconnected.")

Final Tips for Extendability #

• Use proper UUIDs: Replace the placeholder "X" with full 128-bit characteristic UUIDs for better cross-device compatibility.
• Add error handling: Include try/except blocks in both sender and receiver to handle unexpected disconnections or corrupted data.
• Negotiate larger MTU: You can request a higher MTU between Bleak and the nRF52840 to reduce the number of packets needed for large payloads.

备注：内容来源于stack exchange，提问作者the_raging_bear