Search code examples
pythonpython-3.xmultithreadingpython-multithreading

Taking in multiple inputs for a fixed time


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!


Solution

  • 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".