异步 GPIO 实时控制：克服阻塞轮询的障碍

2024-03-18 06:49:19

利用异步编程实现 GPIO 实时控制

简介

在构建基于树莓派的嵌入式系统时，同时操作多个 GPIO 设备至关重要。传统上，GPIO 操作使用阻塞式的轮询，这可能会阻碍实时响应和多任务处理。本文将探讨如何使用 AsyncIO 和 GPIO Zero 库的巧妙结合来实现异步 GPIO 事件处理，从而解决这个问题。

GPIO Zero 和 AsyncIO 的障碍

GPIO Zero 是一个轻量级 Python 库，用于轻松控制 GPIO 引脚。AsyncIO 是一个异步事件循环，允许非阻塞 I/O 操作。然而，将 GPIO Zero 与 AsyncIO 集成时，我们遇到了一个问题：传统的 GPIO Zero 轮询阻碍了异步执行。

异步轮询的解决方案

为了解决这个问题，我们引入了 gpiozero-async 库。它提供了 async_input_device() 函数，允许我们在 AsyncIO 中非阻塞地轮询 GPIO 引脚。通过使用这个函数，我们可以创建异步 GPIO 设备，并使用 async for 循环来监视引脚状态的变化。

重构代码示例

考虑以下代码示例：

import asyncio
from gpiozero import RotaryEncoder, Button

rotary_encoder = RotaryEncoder(a=10, b=9, wrap=False, max_steps=0)
button = Button(7)

async def rotary_detection():
    last_rotary_value = 0
    while True:
        current_rotary_value = rotary_encoder.steps
        if last_rotary_value != current_rotary_value:
            rotDiff = last_rotary_value - current_rotary_value
            if rotDiff > 0:
                print("turned CW")
            else:
                print("turned CCW")
            last_rotary_value = current_rotary_value
        await asyncio.sleep(0.01)

async def button_detection():
    while True:
        if button.is_pressed:
            print("Button pressed")
        await asyncio.sleep(0.01)

async def main():
    await asyncio.gather(
        rotary_detection(),
        button_detection()
    )

if __name__ == "__main__":
    asyncio.run(main())

这个示例使用阻塞式的轮询来监视旋转编码器和按钮的状态，这可能会阻碍异步执行。要解决这个问题，我们可以使用 async_input_device() 函数来重构代码：

import asyncio
from gpiozero.pins.async import async_input_device

async_rotary_encoder = async_input_device(RotaryEncoder, a=10, b=9, wrap=False, max_steps=0)
async_button = async_input_device(Button, 7)

async def rotary_detection():
    async for pin, action in async_rotary_encoder:
        current_rotary_value = async_rotary_encoder.steps
        if last_rotary_value != current_rotary_value:
            rotDiff = last_rotary_value - current_rotary_value
            if rotDiff > 0:
                print("turned CW")
            else:
                print("turned CCW")
            last_rotary_value = current_rotary_value

async def button_detection():
    async for pin, action in async_button:
        if async_button.is_pressed:
            print("Button pressed")

async def main():
    await asyncio.gather(
        rotary_detection(),
        button_detection()
    )

if __name__ == "__main__":
    asyncio.run(main())