如何在设定的时间后终止函数？

Question

在我正在编写的程序中，我需要在设定的时间内运行一个函数，然后在该时间后继续运行我正在等待输入的函数中的程序主代码（要扫描的NFC卡），我希望程序在设定的时间后停止等待

我尝试过使用pytimedinput，但这只适用于正常输入，也适用于尝试子流程（将这部分代码放在不同的文件中），但是这在代码将运行的raspberry pi上不起作用

任何帮助都将不胜感激

Answer 1

这是一种方法：

import multiprocessing
import time

def f():
    while 1:
        print(".")
        time.sleep(1)

if __name__ == "__main__":
    p = multiprocessing.Process(target=f)
    p.start()  # exits immediately leaving the process running
    p.join(timeout=3)
    if p.is_alive():
        p.kill()

    print("Bye")

用if __name__ == "__main__":保护主代码很重要，请参见Programming guidelines for multiprocessing

这里有一个关于如何从衍生流程获取数据的示例：

import multiprocessing
import time

def f(q):
    while 1:
        q.put(".")
        time.sleep(1)

if __name__ == "__main__":
    q = multiprocessing.Queue()
    p = multiprocessing.Process(target=f, args=(q,))
    p.start()  # exits immediately leaving the process running
    p.join(timeout=3)
    if p.is_alive():
        p.kill()
    print("Queue content: ")
    while not q.empty():
        print(q.get())

    print("Bye")

Answer 2

我将使用类似于javascript中使用的设置超时方法的东西

我相信你会在之前的帖子中找到答案： Postpone code for later execution in python (like setTimeout in javascript)

Answer 3

这里有一个示例解决方案供您参考。您可以在这里使用decorator来修饰任何要设置超时的函数

from datetime import datetime, timezone
import multiprocessing
import time


def timeout(seconds):
    def decorator(func):
        def wrapper(*args, **kwargs):
            response = multiprocessing.Manager().dict()

            proc = multiprocessing.Process(target=func, args=(*args, response), kwargs=kwargs)
            proc.start()

            start = time.time()
            while (time.time() - start) < seconds and proc.is_alive():
                pass
            if proc.is_alive():
                print(f"Process is not finished after {seconds} seconds. Terminating...")
                proc.terminate()

            proc.join()
            return dict(response)

        return wrapper

    return decorator


@timeout(10)
def scan_nfc_card(card_name, response):
    # If you need to return anything from the function with a timeout, use the response dictionary.
    response['counter'] = 0
    while True:
        response['counter'] += 1
        print(f"Waiting for NFC card {card_name} to be scanned...")
        time.sleep(1)


print(f"[{datetime.now(timezone.utc)}] Start first process")
response = scan_nfc_card("card-ABC")
print(f"{response=}")
print(f"[{datetime.now(timezone.utc)}] Continue other processes")

输出：

$ python3 script.py 
[2021-09-02 10:23:50.475114+00:00] Start first process
Waiting for NFC card card-ABC to be scanned...
Waiting for NFC card card-ABC to be scanned...
Waiting for NFC card card-ABC to be scanned...
Waiting for NFC card card-ABC to be scanned...
Waiting for NFC card card-ABC to be scanned...
Waiting for NFC card card-ABC to be scanned...
Waiting for NFC card card-ABC to be scanned...
Waiting for NFC card card-ABC to be scanned...
Waiting for NFC card card-ABC to be scanned...
Waiting for NFC card card-ABC to be scanned...
Process is not finished after 10 seconds. Terminating...
response={'counter': 10}
[2021-09-02 10:24:00.502205+00:00] Continue other processes

• 如您所见，函数达到了10秒的超时时间

比如说NFC扫描马上就成功了

@timeout(10)
def scan_nfc_card(card_name, response):
    response['counter'] = 0
    while True:
        response['counter'] += 1
        print("Waiting for NFC card to be scanned...")
        break  # Break right away

输出：

$ python3 script.py 
[2021-09-02 10:23:43.343436+00:00] Start first process
Waiting for NFC card card-ABC to be scanned...
response={'counter': 1}
[2021-09-02 10:23:43.385748+00:00] Continue other processes

• 现在，函数立即结束