使用uvicorn连续运行gpiozero监听器

Question

我正在尝试写一个python应用程序，将运行在树莓派，这将有套接字连接(与uvicorn)和物理输入侦听器。我打算同时侦听套接字连接和gpio事件，而不会互相阻塞。这就是我到目前为止所知道的：

api.py

import uvicorn
import asyncio
from interaction.volume import VolumeControl
from system.platform_info import PlatformInfo
from connection.api_socket import app


class Api:
    def __init__(self):
        pass

    def initialize_volume_listener(self):
        volume_controller = VolumeControl()
        volume_controller.start_listener()

    def start(self):
        PlatformInfo().print_info()
        self.initialize_volume_listener()
        uvicorn.run(app, host='127.0.0.1', port=5000, loop="asyncio")

volume_control.py

import asyncio
from gpiozero import Button
from connection.api_socket import volume_up


class VolumeControl:
    def __init__(self):
        self.volume_up_button = Button(4)

    def volume_up(self):
        loop = asyncio.new_event_loop()
        asyncio.set_event_loop(loop)
        future = asyncio.ensure_future(volume_up(None, None))
        loop.run_until_complete(future)
        loop.close()

    def start_listener(self):
        self.volume_up_button.when_pressed = self.volume_up

api_socket.py

import socketio
from system.platform_info import PlatformInfo

sio = socketio.AsyncServer(async_mode='asgi', cors_allowed_origins='*')
app = socketio.ASGIApp(sio)


@sio.on('connect')
async def test_connect(sid, environ):
    system_info = PlatformInfo().get_info()
    current_volume = 35
    initial_data = {"system_info": system_info,
                    "settings": {"volume": current_volume}
                    }
    await sio.emit('initial_data', initial_data, room=sid)


@sio.on('disconnect request')
async def disconnect_request(sid):
    await sio.disconnect(sid)


@sio.on('disconnect')
async def test_disconnect(sid):
    print('Client disconnected')
    await sio.emit('disconnect', {'data': 'Connected', 'count': 0}, room=sid)


@sio.on('volume_up')
async def volume_up(sid, volume=None):
    increased_volume = 25
    await sio.emit('volume_up', {'volume': increased_volume})


@sio.on('volume_down')
async def volume_down(sid, volume=None):
    decreased_volume = 25
    await sio.emit('volume_down', {'volume': decreased_volume})

我尝试过使用asyncio，但我对python的异步特性还是个新手。问题是，我无法连续运行按钮侦听器，因此当套接字函数正在运行时，我将能够同时侦听按钮交互，而不会阻塞其他按钮。按钮侦听器根本不起作用。相反，只要uvicorn应用程序启动，按钮侦听器就会一直运行。

任何帮助都将不胜感激。谢谢。

Answer 1

@Miguel感谢您的回答。按照您的建议，我已经在while循环中启动了gpio，并在循环中使用了asyncio.run()命令来调用相关的socketio函数。它按预期工作。附注:我已经用参数daemon=True启动了GPIO线程。这使我可以在退出主线程(即uvicorn服务器)时立即退出gpio循环。最终代码如下：

api_socket.py

@sio.on('video_load')
async def load_video(sid, video_number=3):
    data = open(os.path.join(os.getcwd(), f'sample_videos/dummy_video_{str(video_number)}.mp4'), 'rb').read()
    print('Streaming video...')
    await sio.emit('video_load', {'source': data}, room=sid)

nfc_listener.py

class NFCListener:

    reading = True

    def __init__(self):
        GPIO.setmode(GPIO.BOARD)
        self.rdr = RFID()
        util = self.rdr.util()
        util.debug = True
        self.listener_thread = threading.Thread(target=self.start_nfc, daemon=True)

    def start_nfc(self):
        selected_video = None
        while self.reading:
            self.rdr.wait_for_tag()
            (error, data) = self.rdr.request()
            if not error:
                print("\nCard identified!")
                (error, uid) = self.rdr.anticoll()
                if not error:
                    # Print UID
                    card_uid = str(uid[0])+" "+str(uid[1])+" " + \
                        str(uid[2])+" "+str(uid[3])+" "+str(uid[4])
                    print(card_uid)
                    if card_uid == "166 107 86 20 143":
                        if selected_video != 2:
                            selected_video = 2
                            asyncio.run(load_video(None, selected_video))
                    else:
                        if selected_video != 3:
                            selected_video = 3
                            asyncio.run(load_video(None, selected_video))

    def start_reading(self):
        self.listener_thread.start()

Answer 2

gpiozero创建了一个执行回调的新线程(这方面的文档不是很好)。如果回调应该在主asyncio循环中执行，那么您需要将控制传递回主线程。

call_soon_threadsafe方法为您实现了这一点。本质上，它在等待发生时将回调添加到主异步循环调用的任务列表中。

但是，异步循环对于每个线程都是本地的:请参阅get_running_loop

因此，当在主asyncio线程中创建gpiozero对象时，您需要在调用回调时使该循环对象对该对象可用。

下面是我如何对调用asyncio MQTT方法的PIR执行此操作：

class PIR:
    def __init__(self, mqtt, pin):
        self.pir = MotionSensor(pin=pin)
        self.pir.when_motion = self.motion
        # store the mqtt client we'll need to call
        self.mqtt = mqtt
        # This PIR object is created in the main thread
        # so store that loop object
        self.loop = asyncio.get_running_loop()

    def motion(self):
        # motion is called in the gpiozero monitoring thread
        # it has to use our stored copy of the loop and then
        # tell that loop to call the callback:
        self.loop.call_soon_threadsafe(self.mqtt.publish,
                                       f'sensor/gpiod/pir/kitchen', True)

你可能想要这个：

import asyncio
from gpiozero import Button
from connection.api_socket import volume_up

class VolumeControl:
    def __init__(self):
        self.volume_up_button = Button(4)
        self.loop = asyncio.get_running_loop()

    def volume_up_cb(self):
        self.loop.call_soon_threadsafe(volume_up, None, None)

    def start_listener(self):
        self.volume_up_button.when_pressed = self.volume_up_cb

更干净-和线程安全！:)