在固定时间内接收多个输入

Question

我使用的是Python 3，我想编写一个程序，在一定时间内要求多个用户输入。这是我的尝试：

from threading import Timer
##
def timeup():
    global your_time
    your_time = False
    return your_time
##
timeout = 5
your_Time = True
t = Timer(timeout, timeup)
t.start()
##
while your_time == True:
    input()
t.cancel()
print('Stop typing!')

问题是，即使时间到了，代码仍然等待输入。我希望循环在时间用完时停止。我该怎么做呢？谢谢！

Answer 1

此解决方案与 平台无关， 并且会 立即
中断键入以告知有关现有超时的信息。不必等到用户按下ENTER键就可以发现发生了超时。除了及时通知用户之外，这还可以确保在进一步处理超时后不再输入任何内容。

特征

• 平台无关（Unix / Windows）。
• 仅StdLib，无外部依赖项。
• 仅线程，无子进程。
• 超时立即中断。
• 超时时清除提示器的关闭。
• 在时间跨度内可以无限输入。
• 易于扩展的PromptManager类。
• 程序可能会在超时后恢复，可能会在不重新启动程序的情况下多次运行提示实例。

此答案使用线程管理器实例，该实例在单独的提示线程和MainThread之间进行中介。管理器线程检查超时，并将输入从提示线程转发到父线程。通过这种设计，可以轻松修改MainThread，使其无需阻塞（进行更改_poll以替换阻塞queue.get()）。

在超时时，管理器线程要求ENTER继续，并使用一个
threading.Event实例来确保提示线程在继续之前关闭。在特定方法的文档文本中查看更多详细信息：

from threading import Thread, Event
from queue import Queue, Empty
import time


SENTINEL = object()


class PromptManager(Thread):

    def __init__(self, timeout):
        super().__init__()
        self.timeout = timeout
        self._in_queue = Queue()
        self._out_queue = Queue()
        self.prompter = Thread(target=self._prompter, daemon=True)
        self.start_time = None
        self._prompter_exit = Event()  # synchronization for shutdown
        self._echoed = Event()  # synchronization for terminal output

    def run(self):
        """Run in worker-thread. Start prompt-thread, fetch passed
        inputs from in_queue and check for timeout. Forward inputs for
        `_poll` in parent. If timeout occurs, enqueue SENTINEL to
        break the for-loop in `_poll()`.
        """
        self.start_time = time.time()
        self.prompter.start()

        while self.time_left > 0:
            try:
                txt = self._in_queue.get(timeout=self.time_left)
            except Empty:
                self._out_queue.put(SENTINEL)
            else:
                self._out_queue.put(txt)
        print("\nTime is out! Press ENTER to continue.")
        self._prompter_exit.wait()

    @property
    def time_left(self):
        return self.timeout - (time.time() - self.start_time)

    def start(self):
        """Start manager-thread."""
        super().start()
        self._poll()

    def _prompter(self):
        """Prompting target function for execution in prompter-thread."""
        while self.time_left > 0:
            self._in_queue.put(input('>$ '))
            self._echoed.wait()  # prevent intermixed display
            self._echoed.clear()

        self._prompter_exit.set()

    def _poll(self):
        """Get forwarded inputs from the manager-thread executing `run()`
        and process them in the parent-thread.
        """
        for msg in iter(self._out_queue.get, SENTINEL):
            print(f'you typed: {msg}')
            self._echoed.set()
        # finalize
        self._echoed.set()
        self._prompter_exit.wait()
        self.join()


if __name__ == '__main__':

    pm = PromptManager(timeout=5)
    pm.start()

示例输出：

>$ Hello
you typed: Hello
>$ Wor
Time is out! Press ENTER to continue.

Process finished with exit code 0

请注意，在尝试键入“世界”时，此处会弹出超时消息。