如何使用 asyncio 制作一个 python 装饰器函数以在 n 秒内重新安排一个函数？

Question

这是我当前的代码：

import asyncio
import serial


def repeat(seconds):
    def wrap(func):
        def decorated(*args):
            loop = asyncio.get_event_loop()
            loop.call_at(loop.time() + seconds, decorated, *args)
            print('scheduled')
            func(*args)
            return func
        return decorated
    return wrap


@repeat(10)
def send_command(ser, text):
    try:
        if text == 'DATA':
            print("\nSending Data Request")
            ser.write(b'\n022022')
        elif text == 'STAY':
            print("\nInmediate Stay ARM")
            ser.write(b'\n0640010002044D')
        elif text == 'AWAY':
            print("\nInmediate Away ARM")
            ser.write(b'\n0640010003044E')
        elif text == 'DISARM':
            print("\nDISARM")
            ser.write(b'\n0940010001040700055B')
        elif text == 'CHIME':
            print("\nChime toggle")
            ser.write(b'\n0640010007014F')
    except serial.SerialException as e:
        raise e


if __name__ == '__main__':

    ser = serial.Serial('/dev/ttyUSB0', 9600, parity=serial.PARITY_ODD, write_timeout=0, timeout=0)

    loop = asyncio.get_event_loop()
    loop.call_at(loop.time() + 5, send_command, ser, 'DATA')
    loop.run_forever()

在这个例子中，我尝试将 send_command 函数安排到运行在 5 秒内，之后每 10 秒。它似乎有效，但我对 loop.call_at(loop.time() + seconds, decorated, *args) 调用有点困惑。如有任何意见，我们将不胜感激。

Answer 1

你的代码很好（除了你应该 return 装饰器中的 func 结果），我稍微改进了它。

使用合成示例更容易查看和测试：

import asyncio
import functools


def repeat(seconds):
    def wrap(func):
        @functools.wraps(func)  # see 
        def decorated(*args, **kwargs):
            # We should call func that decorated again to
            # force decorator's `call_at` code to execute again:
            loop = asyncio.get_event_loop()
            loop.call_at(
                loop.time() + seconds, 
                functools.partial(decorated, *args, **kwargs)  # see 
            )
            # We should return result of func's excecution:
            return func(*args, **kwargs)
        return decorated
    return wrap


@repeat(2)
def send_command():
    print('send_command')


async def main():
    send_command()  # call once, rescheduling started

    for i in range(10):
        print(i)
        await asyncio.sleep(1)


if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    loop.run_until_complete(main())

输出：

send_command
0
1
send_command
2
3
send_command
4
5
send_command
6
7
send_command
8
9
send_command
[Finished in 10.3s]

如何使用 asyncio 制作一个 python 装饰器函数以在 n 秒内重新安排一个函数？

How can I make a python decorator function to reschedule a function in n seconds using asyncio?

python

decorator

python-asyncio