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

这是基于这个问题的松散基础：How to set time limit on raw_input

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

无论用户是否提示，raw_input提示在指定的4秒时间后都不会超时 .
当输入'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 回答

警告：这可以按照要求在* nix和OSX中使用，但绝对不能在Windows中使用 .

我使用了this modification的ActiveState配方作为下面代码的基础 . 它是一个易于使用的对象，可以通过超时读取输入 . 它使用轮询一次收集一个字符并模拟 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)

重复输入

这实现了我理解它的初衷 . 我认为进行递归调用没有任何 Value - 我认为你想要的只是重复输入？如果错误，请纠正我 .

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

回复于 2024-04-28T02:03:45+08:00

您可以在输入之前设置警报，然后将警报绑定到自定义处理程序 . 在给定的时间段警报消失后，处理程序引发异常，并且您的自定义 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

信用到期：Keyboard input with timeout in Python

回复于 2024-04-28T02:03:45+08:00

0
我不认为有一种方法可以显示一个提示，它将在时间过后过期而不显示另一个线程的不同消息 .

您可以在调用raw_input之前添加以下行：
```
thread.start_new_thread(interrupt_user,())
```
您可以按如下方式定义 interrupt_user 函数：
```
sleep(5)
print "\nTime up"
```
在 raw_input_with_time 函数中，不要调用sleep . 相反，将调用之前的时间保存到raw_input，并确定调用后经过的时间是否超过5秒 . 此外，如果用户输入'q'，则它不应该调用自身，因此循环将停止 .
回复于 2024-04-28T02:03:45+08:00

另一种方法是将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))

回复于 2024-04-28T02:03:45+08:00

这只是概念的教授 . 询问用户输入数据 .

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)

回复于 2024-04-28T02:03:45+08:00

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

注意：我打算做的就是演示另一种思考问题的方法，这种方式在你的特定情况下可能适合也可能不适合 . 我认为它非常灵活 .

概念证明：
```
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代码 . 只有在尝试从线程中获取输入的一些替代实现之后，我才意识到将用户输入留给主线程的优势有多大 .

我没有太注意使其安全和清洁，但可以肯定它可以在保持整体结构的同时完成 . 我能看到的最大缺陷是早期的输入永远不会消失 . 当 Worker 输出时，它会导致混淆，掩盖了先前的输入 . 如果您及时按 q 但不按 Enter ，则下次按 q 和 Enter 会导致输入 qq ，即使这些 q 在屏幕上彼此不相邻也是如此 . 通常这是如何命令行应用程序工作，所以我'm not sure if it'值得修复 . 您可以考虑接受仅包含 q 的输入作为取消 . 另一种选择是直接从 stdin 读取，而不是使用 raw_input .

使代码结构更好的一些想法是使主线程甚至是dumber并让它将所有输入传递给worker（使用队列）来决定如何处理 .
回复于 2024-04-28T02:03:45+08:00

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

6 回答

带超时输入

测试一下

重复输入

相关问题