root/projects/WinDictator/trunk/windictator/linux/keyer.py

# WinDictator:
# Dictate in Windows, have the text typed in Linux via X faked keystroke events
#
# Copyright (C) 2005-2006 by Edwin A. Suominen, http://www.eepatents.com
#
# This program is free software; you can redistribute it and/or modify it under
# the terms of the GNU General Public License as published by the Free Software
# Foundation; either version 2 of the License, or (at your option) any later
# version.
# 
# This program is distributed in the hope that it will be useful, but WITHOUT
# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
# FOR A PARTICULAR PURPOSE.  See the file COPYING for more details.
# 
# You should have received a copy of the GNU General Public License along with
# this program; if not, write to the Free Software Foundation, Inc., 51
# Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA


"""
Keyer for the virtual typist
"""

import os, warnings
from twisted.internet import defer, protocol, reactor

from protocol import ProcessProtocol

BINARY = "/usr/local/bin/typist"


class Keyer:
    """
    I send fake keystrokes to the currently active window using the I{typist}
    process.
    """
    def __init__(self):
        self.dStop = defer.Deferred()
    
    def __del__(self):
        """
        Closes the pipe to the typist process
        """
        self.shutdown()

    def startup(self):
        """
        Opens a pipe to the typist process, which executes a keystroke with
        each new line sent and returns the keycode used on stdout.
        """
        def ready(transport):
            self.dStack = []
            self.process = transport

        def error(self, failure):
            print "Error running typist process:", failure.printTraceback()
            return failure
            
        d = defer.Deferred()
        d.addCallbacks(ready, error)
        p = ProcessProtocol(d, self.dStop, self.whatTyped)
        reactor.spawnProcess(p, BINARY, ('typist',), env=os.environ)
        return d

    def shutdown(self):
        if hasattr(self, 'process'):
            self.process.loseConnection()
            del self.process
            return self.dStop
        else:
            return defer.succeed(None)

    def pressKey(self, keySym):
        """
        Instructs the typist process to depress the specified I{keySym}.
        """
        return self.key('d', keySym)

    def releaseKey(self, keySym):
        """
        Instructs the typist process to release the specified I{keySym}.
        """
        return self.key('u', keySym)

    def key(self, du, keySym):
        """
        Sends a key press (I{d}) or release (I{u}) command to the typist
        process via STDIN. Returns a C{Deferred} that fires with the keycode
        for the pressed or released key.
        """
        def send(null):
            self.process.write("%s %s\n" % (du, keySym))
            d = defer.Deferred()
            if self.dStack:
                warnings.warn(
                    "New keystroke being sent before previous one confirmed",
                    stacklevel=2)
            self.dStack.append(d)
            return d

        if hasattr(self, 'process'):
            return send(None)
        else:
            d = self.startup()
            d.addCallback(send)
            return d

    def whatTyped(self, line):
        """
        Callback for STDOUT feedback from the I{typist} process, with a line
        containing the keycode of the keysym that it has accepted.

        Fires the C{Deferred} for the last-sent keystroke with the integer
        keycode if returned from the I{typist} process, or with C{None} if the
        keystroke wasn't accepted.
        """
        d = self.dStack.pop()
        try:
            result = int(line.strip())
        except:
            result = None
        d.callback(result)