PAYATHON 2026

GPIO 项目中的 Asyncio 代码为何导致动画无限重复？

排障赵师傅

结论

should_animate.clear() 会让当前动画循环退出，但后续 GPIO 回调仍能启动新一轮播放。所示代码没有“播放完成后自动重启”的逻辑。

如果 done animating 之后又出现新一轮时间输出和 audio，说明 run_audio_and_animate() 被再次调用。先记录 audio_callback() 的进入次数，追查重复触发的来源。仅凭当前代码，还无法确定原因是电气干扰、延迟处理的边沿通知，还是未展示代码中的其他调用。

同步 GPIO 调用可以和 asyncio 配合，调度方式需要调整：程序入口只调用一次 asyncio.run()，GPIO 回调只提交通知，由同一个事件循环管理音频和电机动作。

为什么 GPIO 读数为高，回调仍可能执行？

GPIO.FALLING 检测高电平到低电平的变化，GPIO.input() 读取调用那一刻的电平。边沿发生和读取电平的时间可能不同。

例如，输入短暂变低后恢复为高，等 Python 执行回调或打印时，读数已经是 1。所以，动画期间几次读到高电平，并不能排除之前发生过下降沿。

这个装置可以从以下几处排查：

  • 电机启停或扬声器播放时，输入信号是否受到干扰。
  • Furby 的机械动作是否碰到了舌头开关。
  • 按钮接线、上拉和公共地连接是否可靠。
  • 回调线程被长时间占用后，是否有事件通知被延迟处理。

这些原因都还需要验证。

rpi-lgpio 的 bouncetime 要求信号稳定相应时间后才报告边沿，与 RPi.GPIO 的去抖行为不同。bouncetime=500 不会在动画执行期间屏蔽按钮，也不保证下一次回调一定来自一次新的人工按键。rpi-lgpio 去抖说明

原代码中需要修正的地方

回调中的 asyncio.run() 会一直等待整轮任务结束

def audio_callback(pin):
    asyncio.run(run_audio_and_animate())

这段代码在回调所在的线程中创建事件循环，运行任务并完成清理后才返回。任务并没有交给程序开头创建的那个 loop。

如果 GPIO 库在后台线程中执行回调，这种写法就会长时间占用该线程。它本身不会产生下降沿，但会让事件通知的处理和排查更复杂。Python asyncio 运行器说明

main() 实际上是同步函数

loop.run_until_complete(main())

Python 会先执行 main()，再把返回值传给 run_until_complete()。原代码的 main() 阻塞在 signal.pause()，因此外层事件循环没有按预期运行。即使 main() 正常返回，返回值通常也是 None，不是可等待对象。

应将它改为 async def main()，通过异步等待维持程序运行。

clear() 不会中断 asyncio.sleep(2)

原动画要等到下一次检查条件时才会退出：

while should_animate.is_set():
    pwm.ChangeDutyCycle(50)
    await asyncio.sleep(2)

音频结束后，电机可能要等当前这次休眠结束才停止。这段代码也只是在反复设置 50% 占空比，没有每隔两秒切换电机状态。

done animating 说明执行到了 finally。异常和任务取消也会执行 finally，所以排查时应保留异常日志。

音频失败时也必须执行停止逻辑

原代码是：

await asyncio.to_thread(play, produce_clip(...))
should_animate.clear()

如果音频处理或播放抛出异常，后面的 clear() 就不会执行。停止通知应放进 finally。

produce_clip(...) 作为传参表达式，会先在事件循环线程中执行。如果它耗时较长，应单独交给 asyncio.to_thread()。Python 线程执行说明

调整方式与代码示例

调度可以这样安排：

  1. 主线程运行一个长期存在的事件循环。
  2. GPIO 回调通过 loop.call_soon_threadsafe() 提交通知，随即返回。
  3. 在事件循环中检查忙碌状态，直接忽略播放期间收到的触发。
  4. 每轮播放使用独立的结束事件，在 finally 中通知电机停止。

跨线程提交普通回调应使用 call_soon_threadsafe()，asyncio.Event 的操作则应留在事件循环线程内。Python 并发与多线程说明

以下示例适用于 Python 3.11 及以上。保留已有的 produce_clip()、音频文件加载和相关初始化，用这段代码替换原来的回调与调度部分。

import asyncio
import logging
import random
import time
from datetime import datetime

import RPi.GPIO as GPIO
from pydub.playback import play

MOTOR_PIN = 37
BUTTON_PIN = 16

logging.basicConfig(level=logging.INFO)


async def animate(pwm, finished):
    try:
        # 保留原来的 0.5 秒启动延迟。
        # 如果音频提前结束，就不启动电机。
        try:
            await asyncio.wait_for(finished.wait(), timeout=0.5)
            return
        except asyncio.TimeoutError:
            pass

        pwm.ChangeDutyCycle(50)
        await finished.wait()
    finally:
        pwm.ChangeDutyCycle(0)
        print("done animating", flush=True)


async def run_audio_and_animate(pwm):
    timestamp = datetime.now()
    hour = timestamp.hour % 12
    minute = timestamp.minute
    flair = random.randint(0, 60)

    print("start", hour, minute, flair, flush=True)

    # produce_clip 沿用现有实现，返回可供 play 播放的音频。
    clip = await asyncio.to_thread(
        produce_clip, hour, minute, flair
    )

    finished = asyncio.Event()
    animation = asyncio.create_task(animate(pwm, finished))

    try:
        await asyncio.to_thread(play, clip)
    finally:
        finished.set()
        await animation


async def main():
    loop = asyncio.get_running_loop()
    trigger = asyncio.Event()
    busy = False
    closing = False

    GPIO.setmode(GPIO.BOARD)
    GPIO.setup(MOTOR_PIN, GPIO.OUT, initial=GPIO.LOW)
    GPIO.setup(
        BUTTON_PIN,
        GPIO.IN,
        pull_up_down=GPIO.PUD_UP,
    )

    pwm = GPIO.PWM(MOTOR_PIN, 100)
    pwm.start(0)

    def request_start(pin, reported_at):
        # 此函数由事件循环执行。
        if closing:
            return

        level = GPIO.input(pin)
        print(
            f"callback at {reported_at:.6f}: "
            f"pin={pin}, level_now={level}, busy={busy}",
            flush=True,
        )

        if busy or trigger.is_set():
            return

        # 额外过滤：通知处理时按钮已经释放，就忽略它。
        if level != GPIO.LOW:
            return

        trigger.set()

    def audio_callback(pin):
        # 此函数可能由 GPIO 库的后台线程调用。
        loop.call_soon_threadsafe(
            request_start, pin, time.monotonic()
        )

    try:
        GPIO.add_event_detect(
            BUTTON_PIN,
            GPIO.FALLING,
            callback=audio_callback,
            bouncetime=500,
        )

        print("RUNNING", flush=True)

        while True:
            await trigger.wait()
            trigger.clear()
            busy = True

            try:
                await run_audio_and_animate(pwm)
            except Exception:
                logging.exception("音频或动画执行失败")
            finally:
                busy = False
    finally:
        closing = True
        GPIO.remove_event_detect(BUTTON_PIN)
        pwm.ChangeDutyCycle(0)
        pwm.stop()
        GPIO.cleanup()


if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        pass

finished 表示“播放已经结束”，因此停止时调用 set()。动画直接等待这个事件，无需每两秒检查一次标志，也就避免了这段等待造成的停止延迟。

音频在工作线程中播放，动画作为异步任务运行。主循环等本轮完成后，再次等待 trigger。忙碌期间处理到的触发会被丢弃，不会排队留到播放结束后执行。

如何判断重复触发是否已经解决？

启动后只按一次按钮，观察日志：

  • 再次出现 callback at ...，说明程序收到了新的回调通知，需要继续排查输入信号和事件报告时机。
  • 播放期间出现 busy=True，表示这些通知会被忽略。
  • 没有新回调却出现第二条 start，则应检查其他调用位置、是否存在多个运行实例，以及未展示的代码。

可以先禁止电机输出，再分别启用电机和音频，比较回调次数。如果只有电机运行时才出现额外通知，就能将排查范围缩小到机械动作或相关电气干扰。

使用时的注意事项

示例保留了 bouncetime=500，便于与原程序对照。使用 rpi-lgpio 时，短于稳定时间的按压可能无法触发，需要根据实际按钮波形调整。

level != GPIO.LOW 会过滤通知到达时按钮已经释放的按压，因此可能漏掉短按。它适合要求“处理通知时按钮仍需按住”的交互，不能代替对输入信号问题的处理。干扰产生的低电平如果符合条件，软件仍可能将它当作按键。

正常播放结束后，代码会通知电机停止。取消对 asyncio.to_thread(play, clip) 的等待，无法强制终止已经运行的 play()。如果需要随时中断音频，应使用提供停止接口的播放方式。

GPIO.BOARD 使用物理引脚编号，这里的 16 指排针上的第 16 脚。接线和排查时应始终使用同一套编号。

备注：内容仅供参考。