【问题标题】:User input with a timeout, in a loop循环中超时的用户输入
【发布时间】:2015-08-24 23:43:41
【问题描述】:

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

这大致基于这个问题:How to set time limit on 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"

`

【问题讨论】:

  • 你能把解决方案限制在一个操作系统上吗?或者你需要它用于 windows 和 linux 还是...?
  • @kobejohn,最好是 Linux,包括 Mac OS 等衍生产品。
  • 请确认您是否真的需要递归,以便我可以稍微更改您问题的措辞。我在下面的答案中详细说明了这个问题。
  • @user3374113 抱歉,对您没有任何帮助。仅供参考,如果您发现一些可行的方法,您可以发布它并将您自己的答案标记为解决方案。

标签: python python-2.7 raw-input keyboardinterrupt


【解决方案1】:

警告:这旨在按要求在 *nix 和 OSX 中工作,但绝对不能在 Windows 中工作。

我使用 ActiveState 配方的this modification 作为下面代码的基础。这是一个易于使用的对象,可以读取超时输入。它使用轮询一次收集一个字符并模拟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,我尝试实施您的解决方案,但fileno() 出现错误。所以我认为你得到了我想要的......所以 some_function 被启动并需要多长时间才能完成。在 some_function 完成时,屏幕上会出现一个提示,您是否愿意继续,如果在指定时间后说 5 秒用户没有做任何事情 some_function 将再次执行其职责,并且该过程继续进行,直到我在分配内提示时键入 q时间。希望这是有道理的,对不起,如果上面不清楚。我很高兴给你赏金,但我想要一个有效的解决方案
  • @user3374113 感谢您提供信息。 1) 请提供您收到的错误的详细信息。我使用 *nix 进行测试的能力有限,但我会尝试。 2)你到底在使用什么操作系统? 3)根据您的描述,您想要的是循环而不是递归。上面的解决方案完全符合您在循环中描述的内容。希望我们可以让它发挥作用。
  • @user3374113 .... 也许你是working in IDLE?如果是这样,请在 IDLE 之外运行脚本。虽然 IDLE 有很多优点,但它往往会在后台搞砸一些意想不到的事情。
  • 请忽略我之前提到的错误。我得到的错误是:error: (25, 'Inappropriate ioctl for device') 。我正在使用enthought树冠。错误指向:c = self._getch(),特别是_getch(self) 中的old_settings = termios.tcgetattr(fd)。顺便说一句,我正在使用 OSX Mountain lion。
  • @user3374113 我没有 osx 系统可以对此进行测试,但我在同一个 ActiveState 讨论中找到了一些代码。你可以试试上面修改过的代码吗?我刚刚添加了一个特定于 osx 的函数,但现在它被硬编码以使用 osx 函数。如果它有效,我会重写它以透明地工作。
【解决方案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

信用到期:Keyboard input with timeout in Python

【讨论】:

    【解决方案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”,那么它不应该调用自身,因此循环将停止。

    【讨论】:

    • 我刚试过这个,但是当用户在 5 秒后没有输入任何内容时它仍然没有超时。不过,你已经解决了我的“q”问题。
    【解决方案4】:

    另一种方法是将 IO 阻塞放置在新线程中(与您建议的方案相反,您将它放在主线程中)。对此的警告是,在 python 中没有一种干净的方式来杀死一个线程,所以这对于重复调用并不好(N 个线程会一直挂到 main 结束,我认为 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))
    

    【讨论】:

      【解决方案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)
      

      【讨论】:

        【解决方案6】:

        如果不是在输入超时时调用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 代码。只有在尝试了一些从线程中获取输入的替代实现之后,我才意识到将用户输入留给主线程有多大的优势。

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

        使代码结构更好的一些想法是使主线程更加笨拙,并让它将所有输入传递给工作线程(使用队列)来决定要做什么。

        【讨论】:

          猜你喜欢
          • 2017-06-11
          • 1970-01-01
          • 1970-01-01
          • 2013-11-16
          • 2020-02-15
          • 1970-01-01
          • 1970-01-01
          • 1970-01-01
          • 2015-06-14
          相关资源
          最近更新 更多