如何在Python asyncio循环中实现RPi.GPIO事件回调适配LoRa SX127x

接触Python asyncio已有数日，正在树莓派上开发LoRa SX127x模块相关代码。需要将GPIO事件回调集成到asyncio的主运行循环中——已知GPIO事件在后台独立线程运行，并不在asyncio循环内，需让on_rx_done能在asyncio代码中正常工作。

已查阅Python文档中的asyncio.Event、asyncio.loop、asyncio.future，但无法确定哪种方案有效，优先希望使用Python 3.9官方文档中的高层asyncio API。

参考同步版本的LoRa接收代码，希望将其改造为兼容asyncio的版本。预期实现逻辑为：异步的_dio0()回调触发异步的on_rx_done()，将payload存入queue_1，再由make_json_string()函数处理payload。

已尝试方案 #

• 使用ContextVar在asyncio循环间传递queue_1
• 尝试设置asyncio.Event但失败
• 曾触发错误“<_ "UnixSelectorEventLoop" running=True closed=False debug=False>”，目前无响应

当前临时方案 #

采用异步轮询GPIO输入（2023年3月23日更新），但希望找到基于GPIO事件回调的合理解决方案。

代码示例 #

简化版当前尝试代码 #

import asyncio
from SX127x.LoRa import *
from SX127x.LoRaArgumentParser import LoRaArgumentParser
from SX127x.board_config import BOARD
from contextvars import ContextVar

class LoRaRcvCont(LoRa):

   server_status = False

   def __init__(self, verbose=False):
        super(LoRaRcvCont, self).__init__(verbose)

        # here bellow add event function is called
        # GPIO.add_event_detect(dio_number, GPIO.RISING, callback=callback)
        # equivalant GPIO.add_event_detect(gpio_4, GPIO.RISING, callback=_dio0)

        self.set_mode(MODE.SLEEP)
        self.set_dio_mapping([0] * 6)

   async def _dio0(self, channel):#override callback function as async function
        # DIO0 00: RxDone
        # DIO0 01: TxDone
        # DIO0 10: CadDone
        print('_dio')
        if self.dio_mapping[0] == 0:
            await self.on_rx_done()
        elif self.dio_mapping[0] == 1:
            await self.on_tx_done()
        elif self.dio_mapping[0] == 2:
            await self.on_cad_done()
        else:
            raise RuntimeError("unknown dio0mapping!")

    async def on_rx_done(self):#override function as async function
        print("\nLoRa RxDone")
        await asyncio.sleep(.01)
        if self.dio_mapping[0] == 0:
            print("LoRa RxDone")
            self.clear_irq_flags(RxDone=1)
            payload = self.read_payload(nocheck=True)

            # add task in queue_1 if payload not empty
            if len(payload) > 0:
                temp_queue_1 = queue_1_var.get()
                try:
                    print('queue_1 put task, start')
                    await temp_queue_1.put(payload)
                except:  # pylint: disable=bare-except
                    print('ERROR in put queue_1.')
                finally:
                    print('queue_1 put task,  done')
            # return payload

    async def make_json_string(self, queue_1, queue_2):
        while True:
            # print("make_json_string")
            # get a unit of work
            # print("queue_1 get = ")
            payload = await queue_1.get()

            # Notify the queue that the "work item" has been processed.
            queue_1.task_done()

    async def start(self, queue_1):
        self.reset_ptr_rx()
        self.set_mode(MODE.RXCONT)
        count = 0
        #payload = [205, 171, 1, 82, 77, 49, 48, 48, 48, 48]

        # asyncio.get_event_loop()
        while True:

            await asyncio.sleep(3)
            # await queue_1.put(payload)

            count = count + 1
            print("loop" + str(count))

async def main():

    BOARD.setup()

    # create the shared queue
    queue_1 = asyncio.Queue()
    queue_1_var.set(queue_1)


    # Create an Event object.
    event = asyncio.Event()

    lora = lora_config_setup()

    try:
        print("asyncio loop starts ...")
        await asyncio.gather(
            asyncio.create_task(lora.start(queue_1)),
            #asyncio.Event.set(lora._dio0(None)),
            asyncio.create_task(lora.make_json_string(queue_1),
        )
        # )

        print("main done")
    except KeyboardInterrupt:
        sys.stdout.flush()
        print("")
        sys.stderr.write("KeyboardInterrupt\n")
    finally:
        sys.stdout.flush()
        print("")
        lora.set_mode(MODE.SLEEP)

    debug_print(lora)
    BOARD.teardown()

########################
# code starts from here
#########################
queue_1_var = ContextVar('queue_1')
asyncio.run(main())
print("code finished")

异步轮询版本代码（2023年3月23日更新） #

...
class LoRaRcvCont(LoRa):

    server_status = False

    def __init__(self, verbose=False):
        super(LoRaRcvCont, self).__init__(verbose)
        self.set_mode(MODE.SLEEP)
        self.set_dio_mapping([0] * 6)
    async def async_dio0(self, channel):
        # DIO0 00: RxDone
        # DIO0 01: TxDone
        # DIO0 10: CadDone

        print('in _dio')
        while True:
            await asyncio.sleep(.1) 
            if GPIO.input(channel): #irq pin high
                
                #self.get_dio_mapping() #get dio mapping
                print('got interrupt _dio')
                print(f"rx_done: {self.dio_mapping[0]}")
                if self.dio_mapping[0] == 0:
                    await self.on_rx_done()
                elif self.dio_mapping[0] == 1:
                    await self.on_tx_done()
                elif self.dio_mapping[0] == 2:
                    await self.on_cad_done()
                else:
                    raise RuntimeError("unknown dio0mapping!")
            else:
                pass
...

#main asyncio coroutine here all code is executed
async def main():

    BOARD.setup()

    # create the shared queue
    queue_1 = asyncio.Queue()
    queue_1_var.set(queue_1)


    # Create an Event object.
    event = asyncio.Event()

    lora = lora_config_setup()
    # http_server_check() # was using request module

    try:
        print("asyncio loop starts ...")
        await asyncio.gather(
            asyncio.create_task(lora.start(queue_1)),
            # asyncio.create_task(lora.on_rx_done()),
            # asyncio.create_task(lora.on_tx_done),
            asyncio.create_task(lora.async_dio0(BOARD.DIO0)),
            asyncio.create_task(lora.make_json_string(queue_1)),
        )

        print("main done")
    except KeyboardInterrupt:
        sys.stdout.flush()
        print("")
        sys.stderr.write("KeyboardInterrupt\n")
    finally:
        sys.stdout.flush()
        print("")
        lora.set_mode(MODE.SLEEP)

    debug_print(lora)
    BOARD.teardown()

queue_1_var = ContextVar('queue_1')

asyncio.run(main())
print("code finished")

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