用户输入超时，在循环中

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.

我正在尝试创建一个循环python函数，它执行一个任务并提示用户进行响应，如果用户在给定的时间内没有响应，那么序列将重复。

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

这是基于这样一个问题:如何设置raw_input的时间限制

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

任务由some_function()表示。超时是一个以秒为单位的变量。以下代码有两个问题:

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

   无论用户是否提示，在指定的时间4秒之后，raw_input提示符都不会超时。

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.

   当输入“q”的raw_input时(没有“因为我知道任何类型的东西都会自动作为字符串输入)，函数不会退出循环。

`

”

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"

`

”

6 个解决方案

#1

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

警告:这是为了按照要求在*nix和OSX中工作，但绝对不会在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().

我使用了对ActiveState菜谱的修改作为下面代码的基础。它是一个易于使用的对象，可以读取具有超时的输入。它使用轮询每次收集一个字符，并模拟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.

注意:显然下面的_getch_nix()方法对OP不起作用，但是对OSX 10.9.5上的我起作用。您可能会很幸运地调用_getch_osx()，尽管它似乎只适用于32位的python，因为Carbon并不完全支持64位。

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

#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

应记的地方:Python中带有超时的键盘输入

#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:

您可以在调用raw_input之前添加以下行:

 thread.start_new_thread(interrupt_user,())

You can define the interrupt_user function as follows:

可以定义interrupt_user函数如下:

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.

在raw_input_with_time函数中，不要调用sleep。相反，从调用raw_input之前保存时间，并确定调用之后的运行时间是否超过5秒。此外，如果用户输入“q”，那么它不应该调用自己，因此循环将停止。

#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...).

另一种方法是在新线程中放置IO阻塞(与在主线程中设置IO阻塞的方案相反)。需要注意的是，在python中没有一种清除线程的干净方法，因此对于重复调用来说，这种方法并不合适(N个线程将一直挂起直到主线程结束，而且我认为raw_input并不合适…)。

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))

#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)

#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.

如果在输入超时时调用some_function，而不是调用some_function，那么您将该函数转换为一个后台线程，该线程将继续执行超时的间隔?当主线程在等待输入时被永久阻塞时，工作将继续进行。你可能会根据员工在做什么(工作或睡觉)而决定对输入做出不同的反应——你可能会完全忽略它。AFAIK，在不接受输入和接受输入之间没有明显的区别，但是忽略它。这个想法利用。

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.

它不依赖于任何特定平台;它只是简单的Python代码。在尝试了一些从线程内部获取输入的替代实现之后，我才意识到将用户输入留给主线程的好处有多大。

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.

我并没有过多地关注它的安全性和干净性，但可以肯定的是，它可以在保持整体结构的同时进行。我能看到的最大的缺陷是，早期的投入永远不会消失。当worker输出时，它会造成混乱，使先前的输入变得模糊。如果你按q但没有及时输入，按q，下次再输入将会导致qq的输入，即使这些q在屏幕上不是相邻的。一般来说，这就是命令行应用程序的工作方式，因此我不确定它是否值得修复。您也可以考虑接受只包含qs的输入作为取消。另一个选项是直接从stdin中读取，而不是使用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.

使代码结构更美观的一些想法是使主线程更笨，并让它将所有输入传递给工作人员(使用队列)，以决定如何处理。