Taking in multiple inputs for a fixed time
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.
Features
- Platform independent (Unix / Windows).
- StdLib only, no external dependencies.
- Threads only, no Subprocesses.
- Immediate interrupt at timeout.
- Clean shutdown of prompter at timeout.
- Unlimited inputs possible during time span.
- Easy expandable PromptManager class.
- Program may resume after timeout, multiple runs of prompter instances possible without program restart.
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()
).
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()
Example Output:
>$ 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".
You can use the poll() method (tested on Linux):
import select,sys
def timed_input(sec):
po= select.poll() # creating a poll object
# register the standard input for polling with the file number
po.register(sys.stdin.fileno(), select.POLLIN)
while True:
# start the poll
events= po.poll(sec*1000) # timeout: milliseconds
if not events:
print("\n Sorry, it's too late...")
return ""
for fno,ev in events: # check the events and the corresponding fno
if fno == sys.stdin.fileno(): # in our case this is the only one
return(input())
s=timed_input(10)
print("From keyboard:",s)
The stdin buffers the pressed keys, and the input() function read that buffer at once.