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.

Tuesday, March 22, 2011

User Pattern: Button Status

Introduction into User Patterns

Many of you know about Design Patterns of software development, invented to make complicated solutions look less complicated. The problem with Design Patterns is that they describe solutions to coding problems - not for user problem. People abusing Design Patterns tend to align the functionality of application to patterns, sacrificing features users wanted. The application that looks well-designed in the eyes of developers, in fact makes even simple features hard to implement or just takes too much time. As time passes, users become dissatisfied and eventually lose all interest in the product.

People tried to approach the problem from the other end with User Interface Design guidelines. This often made developers unhappy, because their Design Patterns don't explain how to work with web and Ajax application where "business logic" becomes shared between client and server. Developers wanted to see how different frameworks approach this stuff and to make the development process more intuitive. For this reason there should be a way to see how some typical features/functionality are implemented in different frameworks and this should be more detailed than a wiki or a blog in 15 minutes.

So, let me introduce User Patterns. User Pattern is a very abstract story, missing details about design decisions, but detailed enough to provide implementation and see how _simple_, _extensible_ and _maintainable_ this implementation is.

Button Status User Pattern

User story: I want a page with a button and a status field displayed. When I click the button, the status should be updated.

Description: There are five levels in the quest for the implementation of this pattern.
Level 1. The status is blank - it is fetched/calculated only when the button is pressed.
Level 2. The status is initially available - and it is refreshed/calculated when the button is pressed.
Level 3. When refresh/recalculation fails, the error value is displayed in status field.
Level 4. When refresh/recalculation fails, the error value is displayed in status field along with user friendly message in separate field
Level 5. The status value is stored in application database - when the button is pressed, it is recalculated, saved and only when displayed to the user
Level 6. When recalculation/save or fetch operation fails, user should be presented with friendly message explaining the error and further action (i.e. report problem link)

I'm looking forward to create samples for AppEngine webapp, classic Django and Django-nonrel. Feel free to share your snippets.