User input with a timeout, in a loop
Asked Answered
S

6

13

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"

`

Stoffel answered 24/8, 2015 at 23:43 Comment(5)
#2409060Discordancy
Can you limit the solution to one OS? Or do you need it for windows and linux or ...?Nevillenevin
@kobejohn, preferably Linux including derivatives like Mac OS.Stoffel
Please confirm whether or not you actually need this to be recursive so I can change the wording of your question a little. I put more detail on this issue in the answer below.Nevillenevin
@Stoffel sorry nothing worked for you. FYI If you find something that does work, you can post it and mark your own answer as the solution.Nevillenevin
N
3

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
Nevillenevin answered 31/8, 2015 at 14:43 Comment(13)
Hi @kobejohn, I tried to implement your solution but an error came up for fileno(). So I think you get what I want... So some_function is initiated and takes how ever long it takes to complete. At the completion of some_function a prompt will come up on screen will you like to continue, if after a specified time say 5 seconds the user did not do anything some_function will perform its duties again and the process continues until I type q when prompted within allocated time. Hope that made sense, sorry if It wasn't clear above. I am happy to give you the bounty but I would like a working solStoffel
@Stoffel Thanks for the information. 1) Please provide the details of the error you get. I have limited ability to test with *nix but I will try. 2) What OS exactly are you using? 3) Based on your description what you want is a loop rather than recursion. The solution above does exactly what you described with a loop. Hopefully we can get it working.Nevillenevin
@Stoffel .... maybe you are working in IDLE? If so, please run the script outside of IDLE. While IDLE is has many good points, it tends to mess up unexpected things in the background.Nevillenevin
please ignore the previous error i was referring to. The error I am getting is: error: (25, 'Inappropriate ioctl for device') . I am using enthought canopy. The error is pointing to: c = self._getch() and specifically old_settings = termios.tcgetattr(fd) in _getch(self). I'm using OSX Mountain lion btw.Stoffel
@Stoffel I have no osx systems to test this on but I found some code in the same ActiveState discussion. Can you try the modified code above? I just added a function specific to osx but it is hard coded now to use the osx function. If it works, I'll rewrite it to work transparently.Nevillenevin
Thank you for amending your code. I tried it again and then another time and it still gives the same error error: (25, 'Inappropriate ioctl for device'). I might give up... I'll give you the bounty in a few hours time if I can manage myself to get something going.Stoffel
I found a working solution I just need to apply it to my case: https://mcmap.net/q/908831/-asking-a-user-for-input-only-for-a-limited-amount-of-time-in-python-duplicateStoffel
Ok a few things. 1) Don't feel obligated to award the bounty if it doesn't work for you. 2) Somehow my edit didn't save yesterday so the code didn't actually change. I'm sorry about that. I have changed it just now to include the osx-specific code. 3) I got a 10.9.5 OSX system to try this on and the *nix code worked with no problems. However the osx-specific code doesn't work with 64-bit python. If you use 32-bit python, it may work. Thanks for posting the other code you got working. I'll look at it also.Nevillenevin
I see, well gave this one a crack and got a different error: AttributeError: 'module' object has no attribute 'Evt' in _getch_osx(). The bounty ends in about 2 days time so there is still time and I will give it you because of the effort you put in, thank you :)Stoffel
@Stoffel I'm pretty sure that means you are using 64 bit python on OSX. I think it's included by default. That's unfortunate because it causes more trouble than it solves in my experience. As for the *nix code not working for you, I'm guessing it is because you are running from within the enthought environment? Could you run from a plain terminal python timed_input.py and see if it works?Nevillenevin
Oh ok, I will have to try this tomorrow it is already early hours of the morning for me - I will have a go at it tomorrow.Stoffel
@Nevillenevin any ideas why a user got this error in _getch_nix [i, o, e] = select([sys.stdin.fileno()], [], [], self.poll_period) error: (4, 'Interrupted system call')Stillhunt
@Stillhunt I'm sorry I do not know why and I do not have time to investigate now. Maybe you will have luck asking a new question on stackoverflow? If you do that, make sure to isolate the error, determine the environment, etc. as much as possible before posting it.Nevillenevin
J
1

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

Jacobi answered 3/9, 2015 at 23:11 Comment(0)
W
0

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.

Wongawonga answered 24/8, 2015 at 23:47 Comment(1)
I just tried this but it is still doesn't timeout when the user does not input anything after 5 seconds. You have solved my 'q' problem though.Stoffel
C
0

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))
Corrosion answered 3/9, 2015 at 5:59 Comment(0)
D
0

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)
Dachshund answered 3/9, 2015 at 22:1 Comment(0)
C
0

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.

Cresol answered 4/9, 2015 at 0:26 Comment(0)

© 2022 - 2024 — McMap. All rights reserved.