MicroPython asyncio：如何等待两个事件中的任意一个被设置？

在MicroPython的asyncio框架中，你可以通过asyncio.wait()函数实现等待多个事件中的任意一个被触发，具体实现方式如下：

直接使用asyncio.wait()实现 #

import asyncio

ev1 = asyncio.Event()
ev2 = asyncio.Event()

async def func():
    # 等待ev1或ev2任意一个被设置
    done, pending = await asyncio.wait(
        [ev1.wait(), ev2.wait()],
        return_when=asyncio.FIRST_COMPLETED
    )
    
    # 检查并处理触发的事件
    if ev1.is_set():
        print("事件ev1被触发")
        # 若需要重复使用事件，记得重置状态
        ev1.clear()
    if ev2.is_set():
        print("事件ev2被触发")
        ev2.clear()

关键细节说明 #

• asyncio.wait()接收一个可等待对象列表（这里传入两个事件的wait()方法），通过return_when=asyncio.FIRST_COMPLETED参数指定：只要有一个可等待对象完成，就立即结束等待并返回。
• 返回值中的done集合包含已完成的任务，pending是未完成的任务。你可以通过事件的is_set()方法直接判断哪个事件被触发。

封装辅助协程简化复用 #

如果需要在多个协程中复用这个逻辑，可以封装一个通用的辅助协程：

async def wait_any(*events):
    # 生成所有事件的wait任务
    tasks = [event.wait() for event in events]
    # 等待第一个完成的任务
    done, _ = await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED)
    return next(iter(done))

# 调用示例
async def func():
    await wait_any(ev1, ev2)
    # 执行后续业务逻辑

注意事项 #

• 确保你的ESP32 MicroPython固件版本支持asyncio.wait()，主流发行版本均已支持，若使用老旧版本建议升级固件。
• 如果需要添加超时限制，只需给asyncio.wait()增加timeout参数，比如timeout=3表示最多等待3秒。

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