これは、Python 3.8以降(Python 3.6以降に適応できます)のクロスプラットフォームアプローチであり、使用するだけthreading
multiprocessing
です(つまり、シェルユーティリティを使用しないか呼び出します)。これは、コマンドラインからスクリプトを実行することを目的としており、動的な使用にはあまり適していません。
組み込みinput
関数は次のようにラップできます。この場合、組み込みの名前input
をラッパーとして再定義しています。この実装では、すべての呼び出しをinput
これを介してルーティングする必要があるためです。(免責事項:それが、楽しみのために、おそらくあまり良い考えではなく、ただ別の考えである理由です。)
import atexit
import builtins
import queue
import threading
def _make_input_func():
prompt_queue = queue.Queue(maxsize=1)
input_queue = queue.Queue(maxsize=1)
def get_input():
while (prompt := prompt_queue.get()) != GeneratorExit:
inp = builtins.input(prompt)
input_queue.put(inp)
prompt_queue.task_done()
input_thread = threading.Thread(target=get_input, daemon=True)
last_call_timed_out = False
def input_func(prompt=None, timeout=None):
"""Mimics :function:`builtins.input`, with an optional timeout
:param prompt: string to pass to builtins.input
:param timeout: how long to wait for input in seconds; None means indefinitely
:return: the received input if not timed out, otherwise None
"""
nonlocal last_call_timed_out
if not last_call_timed_out:
prompt_queue.put(prompt, block=False)
else:
print(prompt, end='', flush=True)
try:
result = input_queue.get(timeout=timeout)
last_call_timed_out = False
return result
except queue.Empty:
print(flush=True) # optional: end prompt line if no input received
last_call_timed_out = True
return None
input_thread.start()
return input_func
input = _make_input_func()
del _make_input_func
(グローバル名前空間の汚染を回避するために、クロージャー内のの「静的」変数_make_input_func
を非表示にするために、1回限りのセットアップを定義しました。)input
ここでの考え方は、へのすべての呼び出しを処理する別のスレッドをbuiltins.input
作成し、input
ラッパーにタイムアウトを管理させることです。の呼び出しbuiltins.input
は入力があるまで常にブロックされるため、タイムアウトが終了しても、特別なスレッドは入力を待機していますが、input
ラッパーは(with None
)を返します。次の呼び出しで、最後の呼び出しがタイムアウトした場合、builtins.input
(入力スレッドはすでに入力を待機しているため)再度呼び出す必要はなく、プロンプトを出力し、そのスレッドが入力を返すのを待ちます。いつものように。
上記を定義したら、次のスクリプトを実行してみてください。
import time
if __name__ == '__main__':
timeout = 2
start_t = time.monotonic()
if (inp := input(f"Enter something (you have {timeout} seconds): ", timeout)) is not None:
print("Received some input:", repr(inp))
else:
end_t = time.monotonic()
print(f"Timed out after {end_t - start_t} seconds")
inp = input("Enter something else (I'll wait this time): ")
print("Received some input:", repr(inp))
input(f"Last chance to say something (you have {timeout} seconds): ", timeout)