在固定时间内接收多个输入 [英] Taking in multiple inputs for a fixed time
问题描述
我正在使用Python 3,我想编写一个程序,该程序在一定时间内要求多个用户输入.这是我的尝试:
I'm using Python 3 and I wanted to code a program that asks for multiple user inputs for a certain amount of time. Here is my attempt at that:
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!')
问题是,即使时间到了,代码仍然等待输入.我希望循环在时间用完时完全停止.我该怎么做呢?谢谢!
The problem is, the code still waits for an input even if the time is up. I would like the loop to stop exactly when the time runs out. How do I do this? Thank you!
推荐答案
此解决方案是与平台无关的,并且立即会中断键入以通知现有的超时.不必等到用户按下ENTER键就可以发现发生了超时.除了及时通知用户之外,这还可以确保在进一步处理进入的超时后不再输入任何内容.
This solution is platform-independent and immediately interrupts typing to inform about an existing timeout. It doesn't have to wait until the user hits ENTER to find out a timeout occured. Besides informing the user just-in-time this ensures no input after the timeout stepped in is further processed.
功能
- 与平台无关(Unix/Windows).
- 仅StdLib,没有外部依赖项.
- 仅线程,没有子进程.
- 超时立即中断.
- 在超时时清除提示的关闭.
- 在整个时间段内可以无限输入.
- 易于扩展的PromptManager类.
- 程序可能会在超时后恢复,可能会在不重新启动程序的情况下多次运行提示实例.
此答案使用线程管理器实例,该实例在一个
单独的提示线程和MainThread.管理器线程检查超时,并将输入从提示线程转发到父线程.通过这种设计,可以轻松修改MainThread,使其无需阻塞(更改_poll
来替换阻塞的queue.get()
).
This answer uses a threaded manager instance, which mediates between a
separate prompting thread and the MainThread. The manager-thread checks for timeout and forwards inputs from the prompt-thread to the parent-thread. This design enables easy modification in case MainThread would need to be non-blocking (changes in _poll
to replace blocking queue.get()
).
在超时时,管理器线程要求ENTER继续并使用
threading.Event
实例,以确保提示线程在关闭之前关闭
继续.在特定方法的文档文本中查看更多详细信息:
On timeout the manager thread asks for ENTER to continue and uses an
threading.Event
instance to assure the prompt-thread shuts down before
continuing. See further details in the doc-texts of the specific methods:
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
请注意,在尝试输入世界"时,此处会弹出超时消息.
Note the timeout-message here popped up during the attempt of typing "World".
这篇关于在固定时间内接收多个输入的文章就介绍到这了,希望我们推荐的答案对大家有所帮助,也希望大家多多支持IT屋!