Benutzereingabe mit einem Timeout, in einer Schleife

Question

Ich versuche, eine Schleife Python-Funktion, die eine Aufgabe führt und fordert den Benutzer für eine Antwort und wenn der Benutzer nicht in der vorgegebenen Zeit die Sequenz wird wiederholt.

Dieses basiert weg lose auf diese Frage: How to set time limit on raw_input

Die Aufgabe von some_function() vertreten ist. Das Timeout ist eine Variable in Sekunden. Ich habe zwei Probleme mit dem folgenden Code:

1. raw_input Die Aufforderung wird Timeout nicht nach der festgelegten Zeit von 4 Sekunden, unabhängig davon, ob der Benutzer aufgefordert werden oder nicht.

2. Wenn raw_input von 'q' eingegeben wird (ohne '', da ich weiß, dass alles eingegebene Text automatisch als Zeichenfolge eingegeben wird), verlässt die Funktion die Schleife nicht.

`

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" 

`

Answer 1

Ich glaube nicht, dass es eine Möglichkeit ist eine Aufforderung zu zeigen, dass die Zeit vergeht verfallen nach, ohne eine andere Nachricht von einem anderen Thread angezeigt werden.

Sie können die folgende Zeile vor dem Aufruf in den raw_input:

thread.start_new_thread(interrupt_user,()) 

Sie die interrupt_user Funktion definieren können, wie folgt:

sleep(5) 
print "\nTime up" 

In der raw_input_with_time Funktion, schlafen nicht nennen. Speichern Sie stattdessen die Zeit vor dem Aufruf von raw_input und bestimmen Sie, ob die verstrichene Zeit nach dem Aufruf mehr als 5 Sekunden beträgt. Wenn der Benutzer 'q' eingegeben hat, sollte er sich nicht selbst aufrufen, damit die Schleife stoppt.

Answer 2

Warnung: Dies soll in * nix und OSX wie gewünscht funktionieren, aber wird definitiv nicht in Windows funktionieren.

Ich habe this modification eines ActiveState Rezept als Grundlage für den folgenden Code verwendet. Es ist ein einfach zu bedienendes Objekt, das Eingaben mit einem Timeout lesen kann. Es verwendet Abfragen, um Zeichen nacheinander zu sammeln und das Verhalten von raw_input()/input() zu emulieren.

Eingang mit Timeout

Hinweis: offenbar die _getch_nix() Methode unten funktioniert nicht für OP aber es hat für mich auf OSX 10.9.5. Sie haben vielleicht Glück, _getch_osx() stattdessen aufzurufen, obwohl es scheint, in 32-Bit-Python nur zu arbeiten, da Carbon 64-Bit nicht vollständig unterstützt.

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 

Testen Sie

# 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) 

wiederholte Eingabe

Dies setzt Ihre ursprüngliche Absicht, wie ich es verstehe. Ich sehe keinen Wert für rekursive Aufrufe - ich denke, was Sie wollen, ist nur wiederholt Eingabe zu erhalten?Bitte korrigieren Sie mich, wenn das falsch ist.

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 

Answer 3

Ein anderer Weg, es zu tun, ist die IO-Blockierung im neuen Thread zu platzieren (im Gegensatz zu Ihrer vorgeschlagenen Regelung im Gegensatz, wo Sie es in Ihrem Haupt-Thread). Der Vorbehalt dafür ist, dass es keinen sauberen Weg gibt, einen Thread in Python zu löschen, so dass dies nicht gut mit wiederholten Aufrufen funktioniert (N Threads bleiben bis zum Ende hängen, und ich denke, dass raw_input nicht gut spielt ...) .

Also, seien Sie gewarnt, arbeitet dieses eine Mal, bei weitem nicht perfekt Lösung

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

Answer 4

Dies ist nur prof of concept. Benutzer nach Eingabedaten fragen.

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) 

Answer 5

Sie können vor der Eingabe einen Alarm festlegen und den Alarm dann an einen benutzerdefinierten Handler binden. nach dem angegebenen Zeitraum Alarme ausgelöst werden, Handler löst eine Ausnahme aus, und Ihre benutzerdefinierte input Funktion kann den Rest behandeln.
ein kurzes Beispiel:

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 

Kredit wo sie fällig ist: Keyboard input with timeout in Python

Answer 6

Was passiert, wenn statt some_function aufzuzurufen, wenn die Eingabezeiten aus, Sie dass in einen Hintergrundthread drehen, die mit einem gehen halten Intervall der Zeitüberschreitung? Die Arbeit wird fortgesetzt, während der Haupt-Thread permanent blockiert wird, während auf die Eingabe gewartet wird. Sie können sich entscheiden, anders auf diese Eingabe zu reagieren, basierend auf dem, was der Mitarbeiter tut (arbeiten oder schlafen) - Sie könnten ihn einfach komplett ignorieren. AFAIK, gibt es keinen erkennbaren Unterschied zwischen der Eingabe oder der Eingabe, aber dem Ignorieren. Diese Idee nutzt das.

Hinweis: Alles, was ich beabsichtige, ist es, eine andere Art des Nachdenkens über das Problem zu zeigen, das in Ihrem speziellen Fall angemessen sein kann oder nicht. Ich denke jedoch, dass es sehr flexibel ist.

Proof-of-Concept:

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") 

Es beruht nicht auf irgendetwas Plattform-spezifische; Es ist nur einfacher Python-Code. Erst nachdem ich einige alternative Implementierungen aus einem Thread heraus getestet habe, habe ich festgestellt, wie groß der Vorteil ist, der Benutzereingaben für den Hauptthread übrig lässt.

Ich habe nicht zu viel Wert darauf gelegt, es sicher und sauber zu machen, aber sicher kann es unter Beibehaltung der Gesamtstruktur gemacht werden. Der größte Fehler, den ich darin sehen kann, ist, dass frühere Eingaben niemals weggehen. Es verursacht Verwirrung, wenn der Worker ausgibt und die frühere Eingabe verdeckt. Wenn Sie q drücken, aber nicht Enter in der Zeit, drücken q und Enter nächste Zeit in der Eingabe von qq, selbst wenn diese q s nicht nebeneinander auf dem Bildschirm sind. Im Allgemeinen funktionieren Kommandozeilen-Anwendungen so, ich bin mir nicht sicher, ob es sich lohnt, sie zu reparieren. Sie können in Betracht ziehen, auch Eingaben zu akzeptieren, die nur aus s bestehen. Eine andere Möglichkeit wäre, direkt von stdin zu lesen, ohne raw_input zu verwenden.

Eine Idee, um die Struktur des Codes schöner zu machen wäre, den Haupt-Thread sogar dümmer zu machen und es alle Eingabe an den Arbeiter übergeben (mit einer Warteschlange) zu entscheiden, was zu tun ist.