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

Raspberry Pico W通过BLE连接OBD2适配器后发送AT命令及响应获取问题

问题描述

我正在使用MicroPython和aioble库,将Raspberry Pico W作为BLE中心设备连接OBD2适配器,目前已成功完成设备检测与连接,但发送ATZ、ATD等AT命令时无法正确获取OBD返回的响应。另外我注意到OBD服务的读写使用不同UUID,是否需要分别创建读写特征来处理?

我的代码

import aioble
import bluetooth
import machine
import uasyncio as asyncio


_OBD_UUID_SERVICE = bluetooth.UUID(0xfff0)
_OBD_UUID_CHARACTERISTIC = bluetooth.UUID(0xfff1)

led = machine.Pin("LED", machine.Pin.OUT)
connected = False
alive = False


async def find_remote():
    async with aioble.scan(duration_ms=5000, interval_us=30000, window_us=30000, active=True) as scanner:
        async for result in scanner:
            if (result.name() == "OBDBLE"):
                print("Found OBDBLE")
                for item in result.services():
                    print(item)
                if _OBD_UUID_SERVICE in result.services():
                    print("Found OBDBLE service")
                    return result.device
    return None


async def blink_task():
    print("Blink task started")
    toggle = True

    while True and alive:
        blink = 250
        led.value(toggle)
        toggle = not toggle

        if connected:
            blink = 1000
        else:
            blink = 250
        await asyncio.sleep_ms(blink)

    print("Blink task stopped")


async def peripheral_task():

    global connected, alive
    connected = False
    device = await find_remote()

    if not device:
        print("OBD device not found")
        return

    try:
        print("Connecting to", device)
        connection = await device.connect()

    except asyncio.TimeoutError:
        print("Timeout during connection")
        return

    async with connection:
        print("Connected")
        alive = True
        connected = True

   
        try:
            obd_service = await connection.service(_OBD_UUID_SERVICE)
            print(obd_service)
            obd_characteristic = await obd_service.characteristic(_OBD_UUID_CHARACTERISTIC)
            print(obd_characteristic)

        except asyncio.TimeoutError:
            print("Timeout discovering services/characteristics")
            return

    
        try:
            if obd_service == None:
                print("Obd Disconnected")
                alive = False
                connected = False

            if obd_characteristic == None:
                print("No Control")
                alive = False
                connected = False

        except asyncio.TimeoutError:
            print("Timeout during discovery/service/characteristic")
            alive = False
        
        print("Sending ATZ")
        obd_characteristic.write(b"ATZ",response=True)
        obd_characteristic.notified()
        data = await obd_characteristic.read()
        print(data)


    await connection.disconnected()
    print("Disconnected")


async def main():
    tasks = []
    tasks = [
        asyncio.create_task(blink_task()),
        asyncio.create_task(peripheral_task())
    ]

    await asyncio.gather(*tasks)

while True:
    asyncio.run(main())

运行输出

Blink task started
Found OBDBLE
Found OBDBLE
UUID(0xfff0)
Found OBDBLE service
Connecting to Device(ADDR_PUBLIC, aa:bb:cc:12:22:33)
Connected
Service: 19 24 UUID(0xfff0)
Characteristic: 22 21 18 UUID(0xfff1)
Sending ATZ
b''
Disconnected

OBD设备UUID信息

OBD设备UUID详情


解决方案

1. 读写分离的UUID处理

从UUID截图可以看到,OBD服务下有两个独立特征:

  • 0xfff1:仅用于发送命令的可写特征
  • 0xfff2:用于接收响应的通知特征

这类OBD BLE适配器采用"写命令+通知响应"的单向通信模式,而非同一特征双向读写,因此必须分别获取并处理这两个特征,才能完成命令发送与响应接收的闭环。

2. 修复命令发送与响应接收逻辑

当前代码存在3个关键问题:

  • 未获取接收响应的0xfff2特征,无法接收设备返回的数据
  • AT命令缺少换行符结尾,OBD设备无法识别有效命令
  • 错误使用read()获取响应,BLE通知需通过异步订阅接收

修改后的核心代码片段

# 新增定义通知特征UUID
_OBD_UUID_WRITE_CHAR = bluetooth.UUID(0xfff1)
_OBD_UUID_NOTIFY_CHAR = bluetooth.UUID(0xfff2)

# 连接后同时获取两个特征
obd_service = await connection.service(_OBD_UUID_SERVICE)
obd_write_char = await obd_service.characteristic(_OBD_UUID_WRITE_CHAR)
obd_notify_char = await obd_service.characteristic(_OBD_UUID_NOTIFY_CHAR)

# 创建异步任务监听通知
async def listen_response(char):
    while True:
        data = await char.notified()
        print("收到响应:", data.decode('ascii').strip())

asyncio.create_task(listen_response(obd_notify_char))

# 发送带换行符的AT命令(OBD设备通常需要\r\n或\n结尾)
print("Sending ATZ")
await obd_write_char.write(b"ATZ\r\n", response=True)

# 等待设备处理并返回响应
await asyncio.sleep(1)

完整优化要点

  1. 分别定义读写/通知特征的UUID
  2. 连接后同时初始化两个特征
  3. 启动独立异步任务监听通知特征的消息
  4. 发送命令时添加换行符,确保设备能解析
  5. 移除无效的notified()和read()调用,改用通知回调接收响应

3. 其他建议

  • 可添加超时机制,避免无限等待响应
  • 多次发送命令时,可增加命令间的间隔时间,防止设备处理过载
  • 打印特征属性时,可确认obd_notify_char是否支持通知属性(截图中已标注"Notify")

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

相关产品推荐
方舟 Agent Plan

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

最近更新时间:2026.06.30 10:36:03