Принимая несколько входов в течение фиксированного времени в Python

Question

Я использую 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!')

Проблема в том, что код все еще ожидает ввода, даже если время истекло. Я бы хотел, чтобы цикл прекратился именно тогда, когда закончится время. Как мне это сделать? Спасибо!

Answer 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".

Answer 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() считывает этот буфер сразу.

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