Ceci est une approche multiplateforme en Python 3.8+ (bien qu'elle puisse être adaptée à Python 3.6+) qui n'utilise que threading
(donc pas de multiprocessing
ou d'appels à des utilitaires shell). Elle est destinée à l'exécution de scripts à partir de la ligne de commande et n'est pas très adaptée à une utilisation dynamique.
Vous pouvez envelopper la fonction intégrée input
comme suit. Dans ce cas, je redéfinis le nom intégré input
en tant que wrapper, car cette implémentation exige que tous les appels à input
soient routés à travers celui-ci. (Avertissement: c'est pourquoi ce n'est probablement pas une très bonne idée, juste une différente, pour s'amuser.)
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):
"""Imite :function:`builtins.input`, avec un délai facultatif
:param prompt: chaîne à passer à builtins.input
:param timeout: durée d'attente de l'entrée en secondes; None signifie indéfiniment
:return: l'entrée reçue si le délai n'est pas dépassé, sinon 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) # facultatif: terminer la ligne de l'invite si aucune entrée n'est reçue
last_call_timed_out = True
return None
input_thread.start()
return input_func
input = _make_input_func()
del _make_input_func
(J'ai défini la configuration dans la fonction unique _make_input_func
pour masquer les variables "statiques" de input
dans sa fermeture, afin d'éviter de polluer l'espace de noms global.)
L'idée ici est de créer un thread séparé qui gère tous les appels à builtins.input
, et de faire en sorte que le wrapper input
gère le délai. Étant donné qu'un appel à builtins.input
bloque toujours jusqu'à ce qu'il y ait une entrée, lorsque le délai expire, le thread spécial attend toujours une entrée, mais le wrapper input
retourne (avec None
). Au prochain appel, si le dernier appel a expiré, il n'a pas besoin de rappeler builtins.input
(car le thread d'entrée attend déjà une entrée), il imprime simplement l'invite, puis attend que ledit thread retourne une entrée, comme toujours.
Après avoir défini ce qui précède, essayez d'exécuter le script suivant:
import time
if __name__ == '__main__':
timeout = 2
start_t = time.monotonic()
if (inp := input(f"Entrez quelque chose (vous avez {timeout} secondes): ", timeout)) is not None:
print("Entrée reçue:", repr(inp))
else:
end_t = time.monotonic()
print(f"Délai dépassé après {end_t - start_t} secondes")
inp = input("Entrez quelque chose d'autre (j'attendrai cette fois-ci): ")
print("Entrée reçue:", repr(inp))
input(f"Dernière chance de dire quelque chose (vous avez {timeout} secondes): ", timeout)