Пользовательский ввод с тайм-аутом, в цикле

Я пытаюсь создать циклическую функцию python, которая выполняет задачу и запрашивает у пользователя ответ, и если пользователь не отвечает в заданное время, последовательность будет повторяться.

Это частично основано на следующем вопросе: Как установить ограничение по времени для raw_input

Задача представлена ​​some_function(). Тайм-аут - это переменная в секундах. У меня две проблемы со следующим кодом:

  1. Приглашение raw_input не истекает по истечении заданного времени в 4 секунды, независимо от того, запрашивает пользователь или нет.

  2. Когда вводится raw_input из 'q' (без '', потому что я знаю, что все набранное автоматически вводится как строка), функция не выходит из цикла.

`

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"

`


person user3374113    schedule 24.08.2015    source источник
comment
stackoverflow.com/questions/2408560/   -  person Renae Lider    schedule 25.08.2015
comment
Можно ли ограничить решение одной ОС? Или он нужен для windows и linux или ...?   -  person KobeJohn    schedule 28.08.2015
comment
@kobejohn, предпочтительно Linux, включая производные, такие как Mac OS.   -  person user3374113    schedule 28.08.2015
comment
Пожалуйста, подтвердите, действительно ли вам нужно, чтобы это было рекурсивным, чтобы я мог немного изменить формулировку вашего вопроса. Я подробно изложил этот вопрос в ответе ниже.   -  person KobeJohn    schedule 01.09.2015
comment
@ user3374113 извините, у вас ничего не сработало. К вашему сведению, если вы найдете что-то, что действительно работает, вы можете опубликовать это и отметить свой ответ как решение.   -  person KobeJohn    schedule 05.09.2015


Ответы (6)


Предупреждение: это предназначено для работы в * nix и OSX по запросу, но определенно не будет работать в Windows.

Я использовал эту модификацию рецепта ActiveState в качестве основы для кода ниже. Это простой в использовании объект, который может читать ввод с таймаутом. Он использует опрос для сбора символов по одному и имитирует поведение raw_input() / input().

Ввод с тайм-аутом

Примечание: очевидно, что метод _getch_nix() ниже не работает для OP, но он работает для меня в OSX 10.9.5. Возможно, вам повезет с вызовом _getch_osx(), хотя, похоже, он работает только в 32-разрядном питоне, поскольку 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
person KobeJohn    schedule 31.08.2015
comment
Привет, @kobejohn, я пытался реализовать ваше решение, но для fileno() возникла ошибка. Итак, я думаю, вы получили то, что я хочу ... Итак, some_function запускается и занимает столько времени, сколько требуется для завершения. По завершении some_function на экране появится запрос, хотите ли вы продолжить, если по истечении указанного времени, скажем, 5 секунд пользователь ничего не сделал, some_function снова выполнит свои обязанности, и процесс продолжается, пока я не наберу q, когда будет предложено в выделенном время. Надеюсь, это имело смысл, извините, если это не было ясно выше. Я рад дать вам награду, но мне нужен рабочий сол - person user3374113; 01.09.2015
comment
@ user3374113 Спасибо за информацию. 1) Пожалуйста, подробно опишите полученную ошибку. У меня ограниченные возможности для тестирования с * nix, но я попробую. 2) Какую именно ОС вы используете? 3) Исходя из вашего описания, вам нужен цикл, а не рекурсия. Приведенное выше решение делает именно то, что вы описали с помощью цикла. Надеюсь, мы сможем заставить это работать. - person KobeJohn; 01.09.2015
comment
@ user3374113 .... может быть, вы работаете в IDLE? Если да, запустите сценарий вне IDLE. Хотя IDLE имеет много хороших сторон, он имеет тенденцию портить неожиданные вещи в фоновом режиме. - person KobeJohn; 01.09.2015
comment
пожалуйста, не обращайте внимания на предыдущую ошибку, о которой я имел в виду. Я получаю следующее сообщение об ошибке: error: (25, 'Inappropriate ioctl for device') . Я использую элитный навес. Ошибка указывает на: c = self._getch() и, в частности, на old_settings = termios.tcgetattr(fd) в _getch(self). Кстати, я использую OSX Mountain lion. - person user3374113; 02.09.2015
comment
@ user3374113 У меня нет систем OSX, на которых можно было бы это протестировать, но я нашел некоторый код в том же обсуждении ActiveState. Можете ли вы попробовать модифицированный код выше? Я только что добавил функцию, специфичную для osx, но теперь она жестко запрограммирована для использования функции osx. Если заработает, перепишу, чтобы работало прозрачно. - person KobeJohn; 02.09.2015
comment
Спасибо за изменение кода. Я попробовал снова, а затем в другой раз, и он по-прежнему выдает ту же ошибку error: (25, 'Inappropriate ioctl for device'). Я могу сдаться ... Я дам тебе награду через несколько часов, если я сумею что-нибудь сделать. - person user3374113; 02.09.2015
comment
Я нашел рабочее решение, которое мне просто нужно применить в моем случае: stackoverflow.com/a/28424105/3374113 - person user3374113; 02.09.2015
comment
Хорошо, кое-что. 1) Не чувствуйте себя обязанным присуждать награду, если она вам не подходит. 2) Каким-то образом моя правка вчера не сохранилась, поэтому код на самом деле не изменился. Я сожалею о том, что. Я изменил его только сейчас, чтобы включить специфичный для OSX код. 3) Я попробовал это на системе OSX 10.9.5, и код * nix работал без проблем. Однако специфичный для OSX код не работает с 64-битным Python. Если вы используете 32-битный питон, это может сработать. Спасибо, что разместили другой код, который у вас есть. Я тоже посмотрю на это. - person KobeJohn; 02.09.2015
comment
Ясно, я дал ему взломать и получил другую ошибку: AttributeError: 'module' object has no attribute 'Evt' в _getch_osx(). Баунти заканчивается примерно через 2 дня, так что время еще есть, и я дам его вам из-за ваших усилий, спасибо :) - person user3374113; 02.09.2015
comment
@ user3374113 Я почти уверен, что это означает, что вы используете 64-битный Python на OSX. Думаю, по умолчанию включено. Это прискорбно, потому что, по моему опыту, это вызывает больше проблем, чем решает. Что касается кода * nix, который у вас не работает, я предполагаю, что это из-за того, что вы работаете из заданной среды ? Не могли бы вы запустить с обычного терминала python timed_input.py и посмотреть, работает ли он? - person KobeJohn; 02.09.2015
comment
О, ладно, мне придется попробовать это завтра, для меня уже ранние часы - завтра я попробую. - person user3374113; 02.09.2015
comment
@KobeJohn, есть идеи, почему пользователь получил эту ошибку in _getch_nix [i, o, e] = select([sys.stdin.fileno()], [], [], self.poll_period) error: (4, 'Interrupted system call') - person nmz787; 10.07.2018
comment
@ nmz787 Извините, я не знаю почему, и у меня сейчас нет времени разбираться. Может быть, вам удастся задать новый вопрос о stackoverflow? Если вы это сделаете, убедитесь, что вы изолировали ошибку, определите среду и т. Д., Насколько это возможно, прежде чем публиковать ее. - person KobeJohn; 10.07.2018

Вы можете установить будильник перед вводом, а затем привязать его к индивидуальному обработчику. после того, как заданный период времени сработает, обработчик вызовет исключение, и ваша пользовательская функция 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

Кредит, где это необходимо: Ввод с клавиатуры с тайм-аутом в Python

person Kamyar Ghasemlou    schedule 03.09.2015

Я не думаю, что есть способ показать приглашение, которое истечет по прошествии времени, без отображения другого сообщения из другого потока.

Вы можете добавить следующую строку перед вызовом raw_input:

 thread.start_new_thread(interrupt_user,())

Вы можете определить функцию interrupt_user следующим образом:

sleep(5)
print "\nTime up"

В функции raw_input_with_time не вызывайте сон. Вместо этого сохраните время до вызова raw_input и определите, превышает ли время, прошедшее после вызова, 5 секунд. Кроме того, если пользователь ввел «q», он не должен вызывать себя, поэтому цикл остановится.

person Arthur Laks    schedule 24.08.2015
comment
Я просто попробовал это, но все еще не истекает таймаут, когда пользователь ничего не вводит через 5 секунд. Но вы решили мою проблему с q. - person user3374113; 25.08.2015

Другой способ сделать это - разместить блокировку ввода-вывода в новом потоке (в отличие от предложенной вами схемы, в которой она есть в основном потоке). Предостережение для этого заключается в том, что нет чистого способа убить поток в 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))
person agomcas    schedule 03.09.2015

Это просто проф концепция. Запрашивает у пользователя входные данные.

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)
person Abc Xyz    schedule 03.09.2015

Что, если вместо вызова 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 вовремя, нажатие q и Enter в следующий раз приведет к вводу qq, даже если эти q не находятся рядом друг с другом на экране. Обычно так работают приложения командной строки, поэтому я не уверен, стоит ли это исправлять. Вы также можете принять ввод, состоящий только из q, как отмену. Другой вариант - читать напрямую из stdin, не используя raw_input.

Одна из идей сделать структуру кода лучше - сделать основной поток еще более тупым и передать все входные данные рабочему (используя очередь), чтобы решить, что с ним делать.

person Thijs van Dien    schedule 04.09.2015