User input with a timeout, in a loop

Question

I'm trying to create a looping python function which performs a task and prompts the user for a response and if the user does not respond in the given time the sequence will repeat.

This is loosely based off this question: How to set time limit on raw_input

The task is represented by some_function(). The timeout is a variable in seconds. I have two problems with the following code:

1. The raw_input prompt does not timeout after the specified time of 4 seconds regardless of whether the user prompts or not.

2. When raw_input of 'q' is entered (without '' because I know anything typed is automatically entered as a string) the function does not exit the loop.

`

import thread
import threading
from time import sleep

def raw_input_with_timeout():
    prompt = "Hello is it me you're looking for?"
    timeout = 4
    astring = None
    some_function()
    timer = threading.Timer(timeout, thread.interrupt_main)
    try:
        timer.start()
        astring = raw_input(prompt)
    except KeyboardInterrupt:
        pass
    timer.cancel()
    if astring.lower() != 'q':
        raw_input_with_timeout()
    else:
        print "goodbye"

`

Answer 1

Warning: This is intended to work in *nix and OSX as requested but definitely will not work in Windows.

I've used this modification of an ActiveState recipe as a basis for the code below. It's an easy-to-use object that can read input with a timeout. It uses polling to collect characters one at a time and emulate the behavior of raw_input() / input().

Input with Timeout

Note: apparently the _getch_nix() method below doesn't work for OP but it does for me on OSX 10.9.5. You might have luck calling _getch_osx() instead although it seems to work in 32-bit python only since Carbon doesn't fully support 64-bit.

import sys
import time


class TimeoutInput(object):
    def __init__(self, poll_period=0.05):
        import sys, tty, termios  # apparently timing of import is important if using an IDE
        self.poll_period = poll_period

    def _getch_nix(self):
        import sys, tty, termios
        from select import select
        fd = sys.stdin.fileno()
        old_settings = termios.tcgetattr(fd)
        try:
            tty.setraw(sys.stdin.fileno())
            [i, o, e] = select([sys.stdin.fileno()], [], [], self.poll_period)
            if i:
                ch = sys.stdin.read(1)
            else:
                ch = ''
        finally:
            termios.tcsetattr(fd, termios.TCSADRAIN, old_settings)
        return ch

    def _getch_osx(self):
        # from same discussion on the original ActiveState recipe:
        # http://code.activestate.com/recipes/134892-getch-like-unbuffered-character-reading-from-stdin/#c2
        import Carbon
        if Carbon.Evt.EventAvail(0x0008)[0] == 0:  # 0x0008 is the keyDownMask
            return ''
        else:
            # The event contains the following info:
            # (what,msg,when,where,mod)=Carbon.Evt.GetNextEvent(0x0008)[1]
            #
            # The message (msg) contains the ASCII char which is
            # extracted with the 0x000000FF charCodeMask; this
            # number is converted to an ASCII character with chr() and
            # returned
            (what,msg,when,where,mod)=Carbon.Evt.GetNextEvent(0x0008)[1]
            return chr(msg & 0x000000FF)

    def input(self, prompt=None, timeout=None,
              extend_timeout_with_input=True, require_enter_to_confirm=True):
        """timeout: float seconds or None (blocking)"""
        prompt = prompt or ''
        sys.stdout.write(prompt)  # this avoids a couple of problems with printing
        sys.stdout.flush()  # make sure prompt appears before we start waiting for input
        input_chars = []
        start_time = time.time()
        received_enter = False
        while (time.time() - start_time) < timeout:
            # keep polling for characters
            c = self._getch_osx()  # self.poll_period determines spin speed
            if c in ('\n', '\r'):
                received_enter = True
                break
            elif c:
                input_chars.append(c)
                sys.stdout.write(c)
                sys.stdout.flush()
                if extend_timeout_with_input:
                    start_time = time.time()
        sys.stdout.write('\n')  # just for consistency with other "prints"
        sys.stdout.flush()
        captured_string = ''.join(input_chars)
        if require_enter_to_confirm:
            return_string = captured_string if received_enter else ''
        else:
            return_string = captured_string
        return return_string

Test it

# this should work like raw_input() except it will time out
ti = TimeoutInput(poll_period=0.05)
s = ti.input(prompt='wait for timeout:', timeout=5.0,
             extend_timeout_with_input=False, require_enter_to_confirm=False)
print(s)

Repeated Input

This implements your original intention as I understand it. I don't see any value to making recursive calls - I think what you want is just to get input repeatedly? Please correct me if that is wrong.

ti = TimeoutInput()
prompt = "Hello is it me you're looking for?"
timeout = 4.0
while True:
    # some_function()
    s = ti.input(prompt, timeout)
    if s.lower() == 'q':
        print "goodbye"
        break

Answer 2

You can set an alarm before input and then bind the alarm to a custom handler. after the given period alarms goes off, handler raises an exception, and your custom input function may handle the rest.
a quick example:

import signal
class InputTimedOut(Exception):
    pass

def inputTimeOutHandler(signum, frame):
    "called when read times out"
    print 'interrupted!'
    raise InputTimedOut

signal.signal(signal.SIGALRM, inputTimeOutHandler)

def input_with_timeout(timeout=0):
    foo = ""
    try:
            print 'You have {0} seconds to type in your stuff...'.format(timeout)
            signal.alarm(timeout)
            foo = raw_input()
            signal.alarm(0)    #disable alarm
    except InputTimedOut:
            pass
    return foo

s = input_with_timeout(timeout=3)
print 'You typed', s

Credit where it is due: Keyboard input with timeout in Python

Answer 3

I do not think that there is a way to show a prompt that will expire after time passes without displaying a different message from another thread.

You can add the following line before the call to raw_input:

 thread.start_new_thread(interrupt_user,())

You can define the interrupt_user function as follows:

sleep(5)
print "\nTime up"

In the raw_input_with_time function, do not call sleep. Instead, save the time from before the call to raw_input, and determine if the elapsed time after the call is more than 5 seconds. Also, if the user entered 'q' then it should not call itself so the loop will stop.

Answer 4

Another way of doing it is to place the IO blocking in the new thread (as opposed to your proposed scheme where you have it in your main thread). The caveat for this is that there is not a clean way of killing a thread in python, so this does not play nice with repeating calls (N threads will hang around until main ends, and I think raw_input does not play nice...).

So, be warned, this works once, far from perfect solution

import threading
import Queue

def threaded_raw_input(ret_queue):
    print("In thread")
    prompt = "Hello is it me you're looking for?"
    astring = raw_input(prompt)
    ret_queue.put(astring)

if __name__ == '__main__':
    print("Main")
    ret_queue = Queue.Queue()
    th = threading.Thread(target=threaded_raw_input, args=(ret_queue,))
    th.daemon = True    
    th.start()
    try:
        astring = ret_queue.get(timeout=4)
    except Queue.Empty:
        print("\nToo late")
    else:
        print("Your input {}".format(astring))

Answer 5

This is just prof of concept. Asking user for input data.

import time, os
import curses

def main(win):
    win.nodelay(True)
    x=0
    output=""
    while 1:
        win.clear()
        win.addstr(str("Prompt:"))
        win.addstr(str(output))
        x+=1
        try:
           key = win.getkey()
           if key == os.linesep:
              return output
           output += str(key)
           x = 0             
        except: 
           pass
        if x>=50:  # 5s
           return output
        time.sleep(0.1) 

curses.wrapper(main)

Answer 6

What if instead of calling some_function when the input times out, you turn that into a background thread that keeps going with an interval of the timeout? The work will keep going while the main thread is permanently blocked on waiting for input. You may decide to react differently to that input based on what the worker is doing (working or sleeping) - you might just completely ignore it. AFAIK, there is no noticeable difference between not taking input or taking input but ignoring it. This idea leverages that.

Note: All I intend to do is to demonstrate another way of thinking about the problem that may or may not be appropriate in your particular case. I do think it is very flexible though.

Proof of concept:

from __future__ import print_function
from threading import Event, Thread
from time import sleep

def some_function():
    print("Running some function")
    sleep(1)

def raw_input_with_timeout():
    cancel_event = Event()
    wip_event = Event() # Only needed to know if working or waiting

    def worker():
        timeout = 4
        try:
            while not cancel_event.is_set():
                wip_event.set()
                some_function()
                print("Repeating unless 'q' is entered within %d secs!" % timeout)
                wip_event.clear()
                cancel_event.wait(timeout)
        finally:
            wip_event.clear()

    worker_thread = Thread(target=worker)
    worker_thread.start()
    try:
        while not cancel_event.is_set():
            try:
                if raw_input() == 'q' and not wip_event.is_set():
                    cancel_event.set()
            except KeyboardInterrupt:
                pass
    finally:
        cancel_event.set()
        worker_thread.join()
    print("Goodbye")

It doesn't rely on anything platform specific; it's just simple Python code. Only after trying some alternative implementations taking input from within a thread, I realized how much of an advantage leaving user input to the main thread is.

I didn't pay too much attention to making it safe and clean, but for sure it can be done while keeping the overall structure. The biggest flaw I can see with it is that earlier input is never going away. It causes confusion when the worker outputs, obscuring the earlier input. If you press q but not Enter in time, pressing q and Enter next time results in the input of qq even when those qs are not next to each other on the screen. Generally this is how command line applications work, so I'm not sure if it's worth fixing. You might consider accepting input consisting only of qs as cancelation as well. Another option would be to read from stdin directly, not using raw_input.

Some idea to make the structure of the code nicer would be to make the main thread even dumber and have it pass all input to the worker (using a queue) to decide what to do with.