使用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

参考了Grismar的解决方案，并将其与之前提到的‍getch函数结合起来，我在Windows 11、Ubuntu 22.04和Debian 12 GNOME平台上进行了测试，结果代码运行得很好。

import threading
import time


def getch(input_str: list) -> None:
    while True:
        """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 periodical(input_str: list):
    while True:
        print("\033c")
        print(f"\nPrint periodic message, user_inputs: {input_str}")
        time.sleep(3)


while True:
    input_str = []
    t = threading.Thread(target=periodical, args=(input_str,), daemon=True).start()
    getch(input_str)

Answer 2

既然你愿意使用 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 这样的环境中运行时，这个方法可能不会正常工作，因为 PyCharm 的控制台实现和标准的控制台（比如 Windows 控制台或 Windows Terminal）不一样。不过在后者的环境下，它似乎能满足你的需求，对吧？

我的例子是定期打印一条消息，但当然，如果你有办法传达这个信息，它也可以直接结束：

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)