在Python中以异步方式持续监控用户输入

Question

我正在寻找一个Python脚本，它在每个时间间隔持续等待用户输入一段时间。如果用户未能在此时间范围内提供输入，脚本应自动执行预定义的操作并无限期地继续此过程。

为了实现这一目标，我制作了一个代码片段，其中

getch

函数捕获用户输入的字符。程序在超时时间内等待用户的输入。如果用户未能在此间隔内提供输入，则会发出通知以表明时间已过。

import threading


def getch(input_str: list) -> None:
    """Gets a single character"""
    try:
        import msvcrt

        input_str.append(msvcrt.getch().decode("utf-8"))
    except ImportError:
        import sys
        import termios
        import tty

        fd = sys.stdin.fileno()
        oldsettings = termios.tcgetattr(fd)
        try:
            tty.setraw(fd)
            ch = sys.stdin.read(1)
        finally:
            termios.tcsetattr(fd, termios.TCSADRAIN, oldsettings)
        input_str.append(ch)


def input_with_timeout(timeout):
    print("You have {} seconds to enter something:".format(timeout))
    input_str = []
    input_thread = threading.Thread(target=getch, args=(input_str,))
    input_thread.start()
    input_thread.join(timeout)

    if input_thread.is_alive():
        print("Time's up! Performing default operation.")
    else:
        print("You entered:", input_str)


while True:
    input_with_timeout(5)

当用户在规定的时间内提供输入时，系统可以正常运行。然而，如果时间过去，就会出现两个问题：首先，文本显示出现故障，导致文本错位；其次，用户必须多次输入响应，因为单次尝试似乎没有注册输入准确地。

因为pynput和keyboard等库在某些环境下有限制，所以我无法使用它们。

Answer 1

既然你很乐意使用

msvcrt

，那么这种最小实现有什么问题吗？

import threading
import time
import msvcrt


def periodical():
    while True:
        print("\nPrint periodic message")
        time.sleep(3)


def listen_for_keypress():
    while True:
        if msvcrt.kbhit():  
            key = msvcrt.getch()
            print(f"\nKey pressed: {key.decode('utf-8')}", end='')


if __name__ == "__main__":
    threading.Thread(target=periodical, daemon=True).start()
    listen_for_keypress()

请注意，直接从 PyCharm 等环境运行时，这将无法正常工作，因为它的控制台实现不像 Windows 控制台主机或 Windows 终端等标准控制台那样工作。但在其中任何一个中，它似乎都能满足您的要求？

我的示例定期打印一条消息，但当然，如果您有办法传达该事实，它也可以终止：

import threading
import time
import msvcrt


def application_timeout(timeout, stop_event):
    time.sleep(timeout)
    stop_event.set()  # signal stop


def listen_for_keypress(stop_event):
    while not stop_event.is_set():
        if msvcrt.kbhit():
            key = msvcrt.getch()
            print(f"\nKey pressed: {key.decode('utf-8')}", end='')


if __name__ == "__main__":
    stop = threading.Event()
    threading.Thread(target=application_timeout, args=(3, stop), daemon=True).start()

    listen_for_keypress(stop)