You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

Raspberry Pi Pico W使用Aioble断连时出现OSError: [Errno 22] EINVAL问题

解决Aioble断连重连时的OSError: [Errno 22] EINVAL问题

问题分析

当中心设备意外断电时,外设端的BLE连接状态未被正确清理,后续重连操作尝试使用无效的连接句柄或状态,从而触发OSError: [Errno 22] EINVAL错误。

修复方案

1. 扩大异常捕获范围

在所有涉及BLE连接操作的代码块(数据收发、状态检查等)中捕获OSError,触发异常时主动重置连接状态。

修改后的peripheral_task示例代码:

import aioble
import bluetooth
import asyncio

# 定义BLE服务与特征
UART_SERVICE_UUID = bluetooth.UUID("6E400001-B5A3-F393-E0A9-E50E24DCCA9E")
UART_RX_CHAR_UUID = bluetooth.UUID("6E400002-B5A3-F393-E0A9-E50E24DCCA9E")
UART_TX_CHAR_UUID = bluetooth.UUID("6E400003-B5A3-F393-E0A9-E50E24DCCA9E")

uart_service = aioble.Service(UART_SERVICE_UUID)
uart_rx = aioble.Characteristic(uart_service, UART_RX_CHAR_UUID, write=True, write_no_response=True)
uart_tx = aioble.Characteristic(uart_service, UART_TX_CHAR_UUID, notify=True)
aioble.register_services(uart_service)

connected_connection = None

async def peripheral_task():
    global connected_connection
    while True:
        try:
            if connected_connection is None:
                # 启动广播等待连接
                connected_connection = await aioble.advertise(
                    100_000,
                    name="Pico-W-Peripheral",
                    services=[UART_SERVICE_UUID],
                )
                print("已连接中心设备")
            
            # 主动检查连接有效性
            if not connected_connection.is_connected():
                connected_connection = None
                print("连接已断开,重启广播")
                continue
            
            # 数据收发逻辑
            data = await uart_rx.written()
            print(f"收到数据: {data}")
            await uart_tx.notify(connected_connection, b"ACK")
            
        except OSError as e:
            # 处理连接相关的OSError,包括EINVAL
            if e.args[0] == 22:
                print("连接状态无效,重置连接")
            else:
                print(f"发生OSError: {e}")
            # 强制重置连接状态
            connected_connection = None
            # 短暂延迟避免频繁广播
            await asyncio.sleep(1)

async def main():
    await peripheral_task()

asyncio.run(main())

2. 主动检测连接状态

每次操作前通过connected_connection.is_connected()检查连接有效性,提前处理断开情况,避免无效操作触发错误。

3. 强制清理连接资源

捕获异常或检测到断开时,必须将connected_connection置为None,确保下一次循环重新广播并建立新连接。

关键说明

  • Aioble在连接意外中断后,原连接对象会处于无效状态,直接调用会触发EINVAL错误,因此必须显式重置。
  • 添加asyncio.sleep(1)可避免设备断连后立即高频广播,减少资源消耗。

内容的提问来源于stack exchange,提问作者Space Cowboy

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.07.01 22:27:15