循环中带有超时的用户输入

Question

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

这部分内容基于以下问题：如何在 raw_input 上设置时间限制

任务由 some_function（） 表示。超时是以秒为单位的变量。我对以下代码有两个问题：

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

2. 当输入 '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"

- user3374113

---

https://dev59.com/03E95IYBdhLWcg3wN7Kv - Renae Lider

你能将解决方案限制在一个操作系统上吗？还是需要适用于Windows和Linux等多个操作系统？ - KobeJohn

@kobejohn，最好使用Linux，包括Mac OS等衍生版本。 - user3374113

请确认您是否实际需要这个递归，以便我可以稍微修改您的问题措辞。我在下面的答案中对此问题进行了更详细的说明。 - KobeJohn

@user3374113 很抱歉没有任何方法对您有用。顺便提一下，如果您找到了有效的解决方案，可以发布并将自己的答案标记为解决方案。 - KobeJohn

Answer 1

警告：此代码仅适用于*nix和OSX系统，不支持Windows。

我使用了这个修改版的ActiveState recipe作为下面代码的基础。它是一个易于使用的对象，可以在超时时限内读取输入。它使用轮询一次收集一个字符，并模拟raw_input() / input()的行为。

带有超时的输入

注意：显然下面的_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

测试它

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

重复输入

根据我的理解，这实现了您的原始意图。我认为没有必要进行递归调用 - 我想你只是想要重复获取输入？如果我理解错了，请纠正我。

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

- KobeJohn

---

嗨 @kobejohn，我尝试实现你的解决方案，但是fileno()出现了一个错误。所以我想你明白我想要什么了...所以某个函数被启动并将花费多长时间来完成。在一些功能完成后，屏幕上会出现提示是否要继续，如果在指定的时间（比如5秒）内用户没有做任何事情，那么一些功能将再次执行其职责，并且该过程会继续，直到我在分配的时间内键入q为止。希望这有意义，如果上面不清楚，很抱歉。我很愿意给你赏金，但我希望它能正常工作。 - user3374113

请忽略之前的错误，我指的是：error: (25, 'Inappropriate ioctl for device')。我正在使用 Enthought Canopy。该错误指向：c = self._getch()，具体来说是 _getch(self) 中的 old_settings = termios.tcgetattr(fd)。顺便提一下，我正在使用 OSX Mountain Lion。 - user3374113

感谢您修改代码。我尝试了两次，但仍然出现相同的错误 error: (25, 'Inappropriate ioctl for device')。我可能要放弃了...如果我能自己解决问题，几个小时后我会给您奖励。 - user3374113

我找到了一个可行的解决方案，现在只需要将它应用到我的情况中：http://stackoverflow.com/a/28424105/3374113 - user3374113

我明白了，好的，我尝试了一下，但是出现了另一个错误：AttributeError: 'module' object has no attribute 'Evt' 在 _getch_osx() 中。赏金还有大约2天的时间，所以还有时间，我会把它给你，因为你付出了努力，谢谢 :) - user3374113

Answer 2

你可以在输入前设置一个闹钟，然后将闹钟绑定到自定义处理程序上。 在给定的时间段后，闹钟响起，处理程序会引发异常，然后你的自定义input函数可能会处理剩余部分。
快速示例：

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: Python中带有超时的键盘输入

- Kamyar Ghasemlou

Answer 3

我认为没有办法显示一个提示，它会在时间过去后过期，而不显示来自另一个线程的不同消息。

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

 thread.start_new_thread(interrupt_user,())

您可以按照以下方式定义interrupt_user函数：

sleep(5)
print "\nTime up"

在raw_input_with_time函数中，不要调用sleep。相反，保存从调用raw_input之前的时间，并确定调用后经过的时间是否超过5秒。此外，如果用户输入了'q'，则不应该再次调用自身，以便循环停止。

- Arthur Laks

---

我刚试了一下，但是当用户在5秒钟内没有输入任何内容时，它仍然没有超时。不过你已经解决了我的“q”问题。 - user3374113

Answer 4

另一种方法是将IO阻塞放置在新线程中（而不是您提出的方案，在主线程中进行）。这种方法的注意事项是，在Python中没有一种干净的方式来终止线程，因此这与重复调用不兼容（N个线程将一直挂起，直到主线程结束，而且我认为raw_input也不太友好...）。

所以，请注意，这只能工作一次，远非完美的解决方案。

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

- agomcas

Answer 5

这只是一个概念证明。要求用户输入数据。

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)

- Abc Xyz

Answer 6

如果在输入超时时不是调用some_function，而是将其转换为一个后台线程，并保持超时的间隔，会怎样呢？当主线程永久阻塞等待输入时，工作将继续进行。您可以根据工作人员正在做什么（工作或睡眠）来决定对该输入做出不同的反应 - 您可能只是完全忽略它。据我所知，不接受输入或接受输入但忽略它之间没有明显的区别。这个想法利用了这一点。

注意：我打算做的只是展示另一种思考问题的方式，可能适用于您特定的情况，也可能不适用。不过，我认为它非常灵活。

概念证明：

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

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

我没有太关注使其安全和清洁，但肯定可以在保持整体结构的同时完成。我能看到它最大的缺陷是早期的输入永远不会消失。当工作人员输出时，它会导致混淆，遮盖了早期的输入。如果你及时按下q但没有按下Enter，下次按下q和Enter时，即使这些q在屏幕上不相邻，也会输入qq。通常这就是命令行应用程序的工作方式，所以我不确定是否值得修复。您可能还考虑接受仅由q组成的输入作为取消。另一个选择是直接从stdin读取，而不使用raw_input。

使代码结构更好的一些想法是使主线程变得更加愚蠢，并将所有输入传递给工作线程（使用队列）来决定如何处理。

- Thijs van Dien