Taking in multiple inputs for a fixed time [duplicate]
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 aseparate 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 anthreading.Event
instance to assure the prompt-thread shuts down beforecontinuing. See further details in the doc-texts of the specific methods:
from threading import Thread, Eventfrom queue import Queue, Emptyimport timeSENTINEL = 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:
>$ Helloyou typed: Hello>$ WorTime 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,sysdef 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.
Here's a short way of doing that, Without using Signals, NOTE: While loop will be blocked until the user has inputted something and then check for condition.
from datetime import datetime, timedeltat = 5 # You can type for 5 secondsdef timeup(): final_time = datetime.now() + timedelta(seconds=t) print("You can enter now for" + str(t) + " seconds") while datetime.now() < final_time: input() print("STOP TYPING")timeup()