【问题标题】:Taking in multiple inputs for a fixed time [duplicate]在固定时间内接受多个输入[重复]
【发布时间】:2019-04-09 14:03:52
【问题描述】:

我正在使用 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()
  • @Feitan Portor 这就是我开始在下面写答案的方式。问题在于,您的守护程序提示线程一直存在,直到整个过程结束。通常你会假设程序应该在超时发生之后继续做一些事情,然后你不希望提示线程继续接受进一步的输入。您的实现的一些具体可避免的问题是它在计时器内使用忙等待(添加一些 time.sleep 以防止敲击 cpu )和 exit() 将在终端中工作,但在 IDE 控制台中不起作用(更改为 sys.exit())跨度>

标签: python python-3.x multithreading python-multithreading


【解决方案1】:

此解决方案平台无关并且立即中断输入以通知现有超时。它不必等到用户点击 ENTER 来发现超时。除了及时通知用户之外,这还确保在超时介入后不会进一步处理任何输入。

特点

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

此答案使用线程管理器实例,该实例在 单独的提示线程和MainThread。管理器线程检查超时并将输入从提示线程转发到父线程。如果 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”期间此处弹出的超时消息。

【讨论】:

  • 谢谢!这似乎是我正在寻找的解决方案:) 肯定会尝试这个
【解决方案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() 函数会立即读取该缓冲区。

【讨论】:

    【解决方案3】:

    这是一个简短的方法,不使用信号,注意:在用户输入内容然后检查条件之前,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()
    

    【讨论】:

    • 这行不通。循环条件将被评估一次,然后等待循环体终止,然后再次评估它,允许循环体运行任意时间
    • 感谢您的指出。用它更新了我的答案。
    猜你喜欢
    • 2021-10-29
    • 1970-01-01
    • 2015-11-24
    • 2016-01-01
    • 1970-01-01
    • 1970-01-01
    • 1970-01-01
    • 2015-11-10
    • 2019-10-24
    相关资源
    最近更新 更多