I will focus on the Windows solution, as the Unix solution has been presented several times in this thread (using the select
API).
Some solutions suggest using the msvcrt.kbhit()
function to check if there is a user input. This function only works if the program is running in the interactive console and it does not handle multiprocessing case, which uses pipe-based communication. My solution relies on Windows API directly using the pywin32
library:
import msvcrt
import time
import win32api, win32file, win32pipe
def input_timeout(timeout: float) -> str:
"""
Waits for specified period in seconds for user input. If user input is not received in time
function raises exception.
"""
start = time.time()
handle = win32api.GetStdHandle(win32api.STD_INPUT_HANDLE)
typ = win32file.GetFileType(handle)
buffer = ""
while (time.time() - start) < timeout:
# check the type of stdin handle
if typ == win32file.FILE_TYPE_PIPE:
# limited to 1024 bytes per line !!!
data, _, _ = win32pipe.PeekNamedPipe(handle, 1024)
# usage of CR or LF depends on what data you send to subprocess
if "\n" in data:
return input()
else:
time.sleep(0.1)
else:
while msvcrt.kbhit():
ch = msvcrt.getwche()
if ch == "\x08":
buffer = buffer[:-1]
# windows interactive console uses CR character (consider handling \n too)
elif ch == "\r":
print() # add missing LF
return buffer
# possibly other control characters need to be handled here
else:
buffer += ch
time.sleep(0.1)
print("Buffer:", repr(buffer))
raise TimeoutError()
# prompt
print("Enter value (you only have 2.5 seconds): ", end="", flush=True)
val = input_timeout(2.5)
print("Value entered:", val)
I personally recommend using a dedicated thread to handle stdin input with a callback. A lot of people try to use a daemon thread to handle stdin and they rely on the system to kill the thread when the application exits, which is obviously a bad practice - you always need to clean up your mess. Here is my solution:
import threading
import time
from typing import Callable, Optional
import win32api, win32event, win32file, win32pipe
def hook_input(callback: Callable[[str], None], keep_running: threading.Event) -> threading.Thread:
"""
Creates background thread that waits for input on stdin. When a new line is read from stdin
the callback is called with line read from stdin. You can use keep_running flag to terminate
background thread gracefully.
"""
def thread_fun(cb: Callable[[str], None], flag: threading.Event):
handle = win32api.GetStdHandle(win32api.STD_INPUT_HANDLE)
typ = win32file.GetFileType(handle)
while keep_running.is_set():
# WaitForSingleObject is not supported when stdin is pipe (e.g. when running as subprocess)
if typ == win32file.FILE_TYPE_PIPE:
# alternatively you can specify nonzero parameter to retrieve data in pipe buffer
# and check for LF character before calling input() function
_, available, _ = win32pipe.PeekNamedPipe(handle, 0)
if available > 0:
cb(input())
else:
time.sleep(0.1)
else:
# alternatively you cn use msvcrt.kbhit() with msvcrt.getwche() function, but you
# neet to handle control characters yourself (such as 0x08)
if win32event.WaitForSingleObject(handle, 100) == win32event.WAIT_OBJECT_0:
cb(input())
else:
time.sleep(0.1)
thread = threading.Thread(target=thread_fun, args=(callback, keep_running), daemon=False)
thread.start()
return thread
# hook input
flag = threading.Event()
flag.set()
thr = hook_input(lambda val: print("Received:", val), flag)
# do your stuff
# finish
flag.clear()
thr.join()
You can also handle user input timeout in the callback function. The second implementation is safe to use with asyncio
library, in case you replace time.sleep()
function with asyncio.sleep()
. Note that in this case WaitForSingleObject()
will block event loop from running, therefore I suggest you use combination of short WaitForSingleObject()
with longer asyncio.sleep()