Python Bleak扫描蓝牙广播时如何在满足条件后提前退出事件循环

核心思路是借助asyncio.Event作为终止信号，在检测回调中满足条件时触发信号，扫描协程监听到信号后立刻终止扫描流程，不需要等待预设超时。

修改后的完整可运行代码如下：

import asyncio
import struct

from bleak import BleakScanner

timeout_seconds = 10
address_to_look_for = 'masked'
service_id_to_look_for = 'masked'

async def run():
    # 定义终止信号事件
    stop_event = asyncio.Event()
    # 检测回调作为闭包可直接访问stop_event和匹配规则
    def detection_callback(device, advertisement_data):
        # 统一转小写避免大小写不匹配问题
        if device.address.lower() == address_to_look_for.lower():
            byte_data = advertisement_data.service_data.get(service_id_to_look_for)
            if not byte_data:
                return
            # unpack_from返回元组，取第一个元素做判断
            num_to_test = struct.unpack_from('<I', byte_data, 0)[0]
            if num_to_test == 1:
                print('匹配到符合条件的设备，即将终止')
                # 触发终止信号
                stop_event.set()

    scanner = BleakScanner()
    scanner.register_detection_callback(detection_callback)
    await scanner.start()
    try:
        # 优先等待终止信号触发，超时则自动结束
        await asyncio.wait_for(stop_event.wait(), timeout=timeout_seconds)
    except asyncio.TimeoutError:
        print(f"{timeout_seconds}秒扫描超时，未找到符合条件的设备")
    finally:
        # 无论超时还是匹配成功都停止扫描
        await scanner.stop()

if __name__=='__main__':
    # Python 3.7+推荐用asyncio.run简化事件循环操作
    asyncio.run(run())

关键修改说明 #

• 新增asyncio.Event实例作为跨回调和协程的终止信号，回调满足条件时调用set()方法触发信号
• 替换原来的固定时长asyncio.sleep为asyncio.wait_for，实现「信号触发立刻返回/超时自动结束」的逻辑
• 修复原有代码中struct.unpack_from返回元组，直接判断等于1永远不成立的问题
• 新增MAC地址大小写兼容处理、payload非空判断，提升鲁棒性

内容的提问来源于stack exchange，提问作者koffertfisk