固定时间内接收多个输入

6

我正在使用Python 3编写一个程序,需要在一定时间内询问用户多个输入。以下是我的尝试:

from threading import Timer
##
def timeup():
    global your_time
    your_time = False
    return your_time
##
timeout = 5
your_Time = True
t = Timer(timeout, timeup)
t.start()
##
while your_time == True:
    input()
t.cancel()
print('Stop typing!')

问题在于,即使时间到了,代码仍在等待输入。我希望循环在时间完全用尽时停止。我该怎么做?谢谢!


由于答案已被锁定,我将在这里回答... 这将要求输入内容,并在5秒后自动关闭,说明计时器已结束。代码:import time import threading timer = 5 timeout = time.time() + 1 * timer def do_input(): while True: message = input("Message: ") def do_timer(): while time.time() < timeout: continue print("\n计时器已结束!") exit() for i in range(1): thread = threading.Thread(target=do_input) thread.daemon = True thread.start() thread1 = threading.Thread(target=do_timer).start() - Feitan Portor
@Feitan Portor 这基本上就是我在下面写答案的方式。问题在于,你的守护进程提示线程会一直存在,直到整个进程结束。通常情况下,你会认为程序应该在超时发生后继续执行某些操作,然后你不希望提示线程继续接收输入。你的实现中有一些特定的可避免问题,其中包括在计时器中使用繁忙等待(添加一些 time.sleep 以防止占用 CPU),而 exit() 在终端中可以工作,但在 IDE 控制台中无法工作(改为 sys.exit())。 - Darkonaut
3个回答

6

这个解决方案是跨平台的,立即中断打字操作以通知已存在的超时。它不必等到用户按下ENTER键才发现超时发生了。除了及时通知用户外,这还确保了超时后没有输入会被进一步处理。

特性

  • 跨平台(Unix / Windows)。
  • 仅使用StdLib,无需外部依赖。
  • 仅使用线程,无需使用子进程。
  • 在超时时立即中断。
  • 在超时时清洁关闭提示器。
  • 在时间范围内可进行无限输入。
  • 易于扩展的PromptManager类。
  • 程序可以在超时后恢复,可以多次运行提示器实例而无需重新启动程序。

这个答案使用了一个线程管理器实例,它在一个单独的提示线程和主线程之间进行调解。管理线程检查超时并把来自提示线程的输入转发给父线程。这种设计使得在需要使MainThread非阻塞的情况下容易进行修改(在_poll中更改以取代阻塞的queue.get())。

在超时时,管理线程请求按下ENTER键继续,并使用 threading.Event 实例来确保在继续之前提示线程关闭。有关特定方法的详细信息,请参见文档文本:

from threading import Thread, Event
from queue import Queue, Empty
import time


SENTINEL = object()


class PromptManager(Thread):

    def __init__(self, timeout):
        super().__init__()
        self.timeout = timeout
        self._in_queue = Queue()
        self._out_queue = Queue()
        self.prompter = Thread(target=self._prompter, daemon=True)
        self.start_time = None
        self._prompter_exit = Event()  # synchronization for shutdown
        self._echoed = Event()  # synchronization for terminal output

    def run(self):
        """Run in worker-thread. Start prompt-thread, fetch passed
        inputs from in_queue and check for timeout. Forward inputs for
        `_poll` in parent. If timeout occurs, enqueue SENTINEL to
        break the for-loop in `_poll()`.
        """
        self.start_time = time.time()
        self.prompter.start()

        while self.time_left > 0:
            try:
                txt = self._in_queue.get(timeout=self.time_left)
            except Empty:
                self._out_queue.put(SENTINEL)
            else:
                self._out_queue.put(txt)
        print("\nTime is out! Press ENTER to continue.")
        self._prompter_exit.wait()

    @property
    def time_left(self):
        return self.timeout - (time.time() - self.start_time)

    def start(self):
        """Start manager-thread."""
        super().start()
        self._poll()

    def _prompter(self):
        """Prompting target function for execution in prompter-thread."""
        while self.time_left > 0:
            self._in_queue.put(input('>$ '))
            self._echoed.wait()  # prevent intermixed display
            self._echoed.clear()

        self._prompter_exit.set()

    def _poll(self):
        """Get forwarded inputs from the manager-thread executing `run()`
        and process them in the parent-thread.
        """
        for msg in iter(self._out_queue.get, SENTINEL):
            print(f'you typed: {msg}')
            self._echoed.set()
        # finalize
        self._echoed.set()
        self._prompter_exit.wait()
        self.join()


if __name__ == '__main__':

    pm = PromptManager(timeout=5)
    pm.start()

示例输出:

>$ Hello
you typed: Hello
>$ Wor
Time is out! Press ENTER to continue.

Process finished with exit code 0

请注意,这里的超时提示消息是在尝试输入“World”期间弹出的。

谢谢!这似乎是我正在寻找的解决方案 :) 我一定会尝试这个。 - Ma. Julie Anne Gala

2
您可以使用poll()方法(在Linux上测试通过):
import select,sys

def timed_input(sec):

    po= select.poll()   # creating a poll object
    # register the standard input for polling with the file number 
    po.register(sys.stdin.fileno(), select.POLLIN)  

    while True:
        # start the poll
        events= po.poll(sec*1000)   # timeout: milliseconds
        if not events:
            print("\n Sorry, it's too late...")
            return ""

        for fno,ev in events:     #  check the events and the corresponding fno  
            if fno == sys.stdin.fileno():  # in our case this is the only one
                return(input())


s=timed_input(10)
print("From keyboard:",s)  

标准输入(stdin)会缓存按下的键,而input()函数会一次性读取该缓存。


1
这里有一种简单的方法来实现,不需要使用信号。请注意:while循环将会被阻塞直到用户输入了内容并且检查条件。
from datetime import datetime, timedelta
t = 5  # You can type for 5 seconds
def timeup():
    final_time = datetime.now() + timedelta(seconds=t)
    print("You can enter now for" + str(t) + " seconds")
    while datetime.now() < final_time:
        input()

    print("STOP TYPING")

timeup()

这样做不行。循环条件只会被评估一次,然后等待循环体终止再次评估它,允许循环体运行任意时间。 - sam-pyt
谢谢您指出。已将它更新到我的回答中。 - Vineeth Sai

网页内容由stack overflow 提供, 点击上面的
可以查看英文原文,
原文链接