13

タスクを実行し、ユーザーに応答を求めるループ python 関数を作成しようとしています。ユーザーが指定された時間内に応答しない場合、シーケンスが繰り返されます。

これは、次の質問に大まかに基づいています: raw_input に時間制限を設定する方法

タスクは で表されsome_function()ます。タイムアウトは秒単位の変数です。次のコードには 2 つの問題があります。

  1. raw_input プロンプトは、ユーザーがプロンプトを出すかどうかに関係なく、指定された 4 秒後にタイムアウトしません。

  2. 'q' の raw_input が入力されると (入力されたものはすべて文字列として自動的に入力されることがわかっているため、'' なしで)、関数はループを終了しません。

`

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"

`

4

6 に答える 6

3

警告: これは要求に応じて *nix および OSX で動作することを意図していますが、Windows では動作しません。

以下のコードの基礎として、この ActiveState レシピの変更を使用しました。タイムアウトで入力を読み取ることができる使いやすいオブジェクトです。ポーリングを使用して文字を 1 つずつ収集し、raw_input()/の動作をエミュレートしますinput()

タイムアウト付き入力

注:どうやら_getch_nix()以下の方法はOPでは機能しませんが、OSX 10.9.5では機能します。_getch_osx()Carbon は 64 ビットを完全にサポートしていないため、32 ビットの Python でのみ動作するように見えますが、代わりに呼び出すことができます。

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
于 2015-08-31T14:43:12.627 に答える
0

some_function入力がタイムアウトしたときに呼び出す代わりに、それをバックグラウンド スレッドに変え、タイムアウトの間隔で処理を続けるとしたらどうでしょうか? メインスレッドが入力待ちで永久にブロックされている間、作業は続行されます。ワーカーが何をしているか (仕事中か睡眠中か) に基づいて、その入力に対して異なる反応を示すことができます。完全に無視することもできます。私の知る限り、入力を受け取らないか、入力を取っても無視するかに顕著な違いはありません。それを活かしたのがこのアイデアです。

注: 私が意図しているのは、特定のケースで適切である場合とそうでない場合がある問題について別の考え方を示すことだけです。でも、とても柔軟だと思います。

コンセプトの証明:

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が入力されます。通常、コマンド ライン アプリケーションはこのように動作するため、修正する価値があるかどうかはわかりません。sだけで構成される入力をキャンセルと見なすこともできます。別のオプションは、を使用せずに直接読み取ることです。qqqqstdinraw_input

コードの構造をより良くするためのいくつかのアイデアは、メインスレッドをさらに愚かにし、何をすべきかを決定するために (キューを使用して) すべての入力をワーカーに渡すようにすることです。

于 2015-09-04T00:26:22.647 に答える
0

別のスレッドから別のメッセージを表示せずに、時間が経過すると有効期限が切れるプロンプトを表示する方法はないと思います。

raw_input への呼び出しの前に次の行を追加できます。

 thread.start_new_thread(interrupt_user,())

interrupt_user関数は次のように定義できます。

sleep(5)
print "\nTime up"

関数内でraw_input_with_timesleep を呼び出さないでください。代わりに、raw_input への呼び出し前からの時間を保存し、呼び出し後の経過時間が 5 秒を超えているかどうかを判断します。また、ユーザーが「q」を入力した場合、それ自体を呼び出してはならないため、ループが停止します。

于 2015-08-24T23:47:25.547 に答える