Tuesday, March 29, 2011

Asynchronous input from Windows console

too long didn't read - https://bitbucket.org/techtonik/async-console-input - public domain/MIT example for Windows implemented in pure Python (ctypes)

What is asynchronous input?

Given: The program need to terminate immediately when a subprocess exits or user hits 'q'. It should not eat 100% CPU time.

Blocking synchronous input:
import sys, subprocess
from msvcrt import getch, kbhit

p = subprocess.Popen([r"notepad"], shell=True)

while True:
char = getch()
if char == 'q':
sys.exit('terminated by user')
if p.poll() != None:
sys.exit('terminated by child exit')

This one doesn't work (doesn't exits immediately when child process terminates), because getch() blocks execution key is pressed, and even if child process exits, it won't be detected until some key is pressed.

Non-blocking synchronous input (polling):
import sys, subprocess
from msvcrt import getch, kbhit

p = subprocess.Popen([r"notepad"], shell=True)

while True:
while kbhit():
if getch() == 'q':
sys.exit('terminated by user')
if p.poll() != None:
sys.exit('terminated by child exit')

This works ok, but constant polling uses 100% of CPU resources. It is possible to insert time.sleep() instructions to reduce the carbon footprint, but these crutches will greatly slow down the console when you add z-type to console to spend your time until the background process is finished.

Asynchronous console input on Windows with Python

Asynchronous input allows your program to receive notification from operating system when an input is available. This means that you define events your program needs to react (not necessary console events), send this list to operating system and put your program into wait mode. When system sees this event, your wait function returns and it's possible to inspect/filter event to a greater detail.

Windows provides WaitForMultipleObjects wait function, and in Linux I believe it is select call. If you've tried to understand how Twisted works, but couldn't - will it make more clear if I say that twisted reactor is a wait function? What you do when you code with twisted is just configuring events to react, making chains of them so that one event reacts to another event. This allow to build very interesting, complex and de-coupled asynchronous applications in a couple of days.

O.k. Getting back from Twisted to the asynchronous console input on Windows. Below is the full source code that uses WaitForMultipleObjects. I am afraid that's minimal complete example possible to build with ctypes using Windows API. Tested with Python 2.5

Non-blocking asynchronous input from console:
Example of non-blocking asynchronous console input using
Windows API calls in Python. This can become handy for
async console tools such as IRC client.

Public domain or MIT license
by anatoly techtonik

1. WaitForMultipleObjects is used to listen for the
signals from process and stdin handles
2. When handle is signalled it remains in this state
until reset
3. msvcrt.* keyboard functions don't clear signalled
state from stdin handle, that's why console API
functions are used to clear the input buffer
instead of kbhit()/getch() loop

import ctypes
import ctypes.wintypes
import subprocess

# open notepad in separate process and monitor its execution
# at the same time asynchronously processing events from
# standard input without wasting 100% CPU on looping

# OpenProcess desired access flag
# "the right to use the object for synchronization. This
# enables a thread to wait until the object is in the
# signaled state"
# Constant to get stdin handle with GetStdHandle() call
# Constant for infinite timeout in WaitForMultipleObjects()

# --- processing input structures -------------------------
# INPUT_RECORD structure
# events:
EVENTIDS = dict(
FOCUS_EVENT = 0x0010,
KEY_EVENT = 0x0001, # only key event is handled
MENU_EVENT = 0x0008,
MOUSE_EVENT = 0x0002,
EVENTS = dict(zip(EVENTIDS.values(), EVENTIDS.keys()))
# records:
class _uChar(ctypes.Union):
_fields_ = [('UnicodeChar', ctypes.wintypes.WCHAR),
('AsciiChar', ctypes.wintypes.c_char)]
class KEY_EVENT_RECORD(ctypes.Structure):
_fields_ = [
('keyDown', ctypes.wintypes.BOOL),
('repeatCount', ctypes.wintypes.WORD),
('virtualKeyCode', ctypes.wintypes.WORD),
('virtualScanCode', ctypes.wintypes.WORD),
('char', _uChar),
('controlKeyState', ctypes.wintypes.DWORD)]
class _Event(ctypes.Union):
_fields_ = [('keyEvent', KEY_EVENT_RECORD)]
class INPUT_RECORD(ctypes.Structure):
_fields_ = [
('eventType', ctypes.wintypes.WORD),
('event', _Event)]
# --- /processing input structures ------------------------

np = subprocess.Popen([r"notepad"],
# OpenProcess returns handle that can be used in wait functions
# params: desiredAccess, inheritHandle, processId
nph = ctypes.windll.kernel32.OpenProcess(SYNCHRONIZE, False, np.pid)
print("Started Notepad with pid=%s, handle=%s" % (np.pid, nph))

ch = ctypes.windll.kernel32.GetStdHandle(STD_INPUT_HANDLE)

handles = [ch, nph]

eventnum = ctypes.wintypes.DWORD()
eventread = ctypes.wintypes.DWORD()
inbuf = (INPUT_RECORD * 1)()

print "[q]uit, [s]top console processing, launch bro[w]ser"
stopflag = False
while not stopflag and nph in handles:
print "Waiting for handles %s.." % handles

HandleArrayType = ctypes.wintypes.HANDLE * len(handles)
handle_array = HandleArrayType(*handles)
# params: count, handles, waitAll, milliseconds
ret = ctypes.windll.kernel32.WaitForMultipleObjects(
len(handle_array), handle_array, False, INFINITE)

if handles[ret] == ch:
# msvcrt won't work, because it doesn't reset
# signalled state of stdin handle
import msvcrt
while msvcrt.kbhit():
print "key!"
print msvcrt.getch()
# --- processing input ---------------------------
ch, ctypes.byref(eventnum))
for i in range(eventnum.value):
# params: handler, buffer, length, eventsnum
ch, ctypes.byref(inbuf), 2, ctypes.byref(eventread))
if EVENTS[inbuf[0].eventType] != 'KEY_EVENT':
print EVENTS[inbuf[0].eventType]
keyEvent = inbuf[0].event.keyEvent
if not keyEvent.keyDown:
char = keyEvent.char.UnicodeChar.lower()
#print char, keyEvent
if char == 'q':
print('[q] key pressed. Exiting..')
stopflag = True
elif char == 's':
elif char == 'w':
import webbrowser
#print char
# --- /processing input --------------------------

elif handles[ret] == nph:
print("Notepad is closed. Exiting..")
print("Warning: Unknown return value '%s'" % ret)

print "Done."

Where can it be useful?

Writing first Windows console IRC client in Python and make it cross-platform? Network log viewers with keyboard shortcuts? Real-time roguelikes? Matrix sniffer screensaver? I don't know - its your time. Hope I saved you some though.

Enjoy! If you want to enhance this stuff - feel free to join ever empty https://groups.google.com/forum/#!forum/rainforce for public discussion.