root/projects/WinDictator/trunk/windictator/linux/observer.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


"""
Keystroke observer
"""

import os, re
from twisted.internet import defer, reactor
import protocol, keycodes

BINARY = ("/usr/local/bin/cnee",
          "--record", "--keyboard", "--events-to-record", "-1")
HISTORY_LENGTH = 128
MAX_WAIT = 1.0


class KeyHistory(object):
    """
    I maintain a seekable history of key events.
    """
    def __init__(self, N=HISTORY_LENGTH):
        self.N = N
        self.position = 0
        self.keysDown = {}
        self.realHistory = {}
        self.fakeHistory = {}
        self.keyDecoder = keycodes.KeyDecoder()

    def backspace(self, N):
        """
        Deletes the most recent I{N} events that are in my combined history
        buffers.
        """
        real, fake = self.realHistory, self.fakeHistory
        N_limited = min([N, self.N, len(real)+len(fake)])
        timeDict = dict.fromkeys(real.keys(), True)
        timeDict.update(dict.fromkeys(fake.keys(), False))
        timeList = timeDict.keys()
        timeList.sort()
        for key in timeList[-N_limited:]:
            if timeDict[key]:
                whichHistory = real
            else:
                whichHistory = fake
            del whichHistory[key]

    def reportKeyEvent(self, keycode, isKeyPress, isRealTyping, timeReported):
        """
        Reports observation of a new keying event with I{keycode}, indicating:

            - if it I{isKeyPress} (C{True}) or, if C{False}, a key release;

            - if I{isRealTyping} (C{True}) or, if C{False}, if it is a
              combination of faked key events from dictation;

            - the I{timeReported} of the event.
            
        """
        if isKeyPress:
            self.currentEventIsReal = isRealTyping
            self.keysDown[keycode] = None
        elif keycode in self.keysDown:
            self.keysDown[keycode] = timeReported
        if len(self.keysDown) and None not in self.keysDown.values():
            # All pressed keys have now been released
            timeCompleted = max(self.keysDown.values())
            keyCombo = self.keysDown.keys()
            self.put(keyCombo, self.currentEventIsReal, timeCompleted)
            self.keysDown.clear()

    def put(self, keyCombo, isRealTyping, timeCompleted):
        """
        Puts a new key code combination into my history buffer, indicating:

            - if it I{isRealTyping} (C{True}) or, if C{False}, if it is a
              combination of faked key events from dictation;

            - the I{timeCompleted} of the key events in the combination.
            
        """
        # Make a local reference to the right history buffer for this call
        if isRealTyping:
            history = self.realHistory
        else:
            history = self.fakeHistory
        # Make sure we don't overwrite an existing buffer item due to integer
        # time value being the same for very quick succession of key events
        while timeCompleted in history:
            timeCompleted += 1
        # Write the new key event to the buffer
        history[timeCompleted] = keyCombo
        # Keep the total real + fake buffer size at N, doing the rotation on
        # the current buffer if necessary.
        if len(self.realHistory) + len(self.fakeHistory) > self.N:
            oldestKey = min(history.keys())
            del history[oldestKey]
        # All done

    def getKeys(self, N, realOnly=False, fakeOnly=False):
        """
        Get a list of the I{N} key combinations before the current position in
        my history buffer. You can specify I{realOnly} or I{fakeOnly} by
        setting one of those options C{True}.

        The history buffer is of limited size, and values of I{N} in excess of
        that size will only return the full buffer contents. Also, only as many
        items as are in the buffer up to the current position will be returned;
        there is no wrap-around.
        """
        history = {}
        if not realOnly:
            history.update(self.fakeHistory)
        if not fakeOnly:
            history.update(self.realHistory)
        N_limited = min([N, self.N, len(history)])
        timeList = history.keys()
        timeList.sort()
        return [history[x] for x in timeList[-N_limited:]]

    def getText(self, N, realOnly=False, fakeOnly=False):
        """
        Returns a string of the text typed via the past I{N} key combinations
        from my history buffer. You can specify I{realOnly} or I{fakeOnly} by
        setting one of those options C{True}.

        The history buffer is of limited size, and increased values of I{N}
        will not return more text after a certain point. Also, only as many
        items as are in the buffer up to the current position will be returned;
        there is no wrap-around.
        """
        charList = [self.keyDecoder.keyComboToText(x)
                    for x in
                    self.getKeys(N, realOnly=realOnly, fakeOnly=fakeOnly)]
        return ''.join(charList)


class Observer:
    """
    I observe what keystrokes have been generated (real and fake alike) using
    the I{xnee} process.
    """
    def __init__(self, history=None):
        self.dStop = defer.Deferred()
        self.re = re.compile(r'^0,[23],(0,){3}[1-9][0-9]*,0,[0-9]+')
        self.history = history
        self.confirmationRequestFIFO = []

    def __del__(self):
        self.shutdown()

    def startup(self, *null):
        """
        Opens a pipe to the I{xnee} process, which reports each observed
        keystroke with a new line sent via stdout.
        """
        def ready(transport):
            self.process = transport

        def error(self, failure):
            print "Error running xnee process:", failure.printTraceback()
            return failure

        self.confirmationRequestFIFO = []
        d = defer.Deferred()
        d.addCallbacks(ready, error)
        p = protocol.ProcessProtocol(d, self.dStop, self.observed)
        reactor.spawnProcess(
            p, BINARY[0], ("xnee",) + BINARY[1:], env=os.environ)
        return d

    def shutdown(self, *null):
        """
        Shuts down the I{xnee} process in an orderly fashion.
        """
        if hasattr(self, 'process'):
            try:
                self.process.signalProcess(2)
                self.process.loseConnection()
            except:
                pass
            del self.process
            return self.dStop
        else:
            return defer.succeed(None)

    def observed(self, line):
        """
        Callback for reporting new keystrokes observed.
        """
        # Got a report line...
        if not self.re.match(line):
            return
        # ...and it's of a new keying event
        keycode, isKeyPress, timeReported = self.lineParser(line)
        if self.confirmationRequestFIFO:
            # There's a confirmation request outstanding, so it may be a faked
            # event.
            firstRequest = self.confirmationRequestFIFO[0]
            if keycode == firstRequest[0]:
                # Yes, it matches the oldest request.
                firstRequest[1].callback(True)
                self.confirmationRequestFIFO.pop(0)
                isRealTyping = False
            else:
                isRealTyping = True
        else:
            # No confirmation requests are outstanding, so it must be a real
            # event.
            isRealTyping = True
        # Report the keying event, real or fake, to whatever history object I
        # am using
        if self.history is not None:
            self.history.reportKeyEvent(
                keycode, isKeyPress, isRealTyping, timeReported)

    def lineParser(self, line):
        """
        Parses a line received from the xnee process, returning a tuple of
        parsed data if the line reported on a keying event::

            0: *
            1: Keypress = '2', Release = '3'
            2: *
            3: *
            4: *
            5: Keycode
            6: *
            7: Time
            -----------------------------------------------------
            * Seems to be '0' for every keystroke-reporting line
            
        """
        fields = [int(x) for x in line.strip('()').split(',')]
        if fields[0] == 0 and fields[1] in (2,3):
            keycode = fields[5]
            isKeyPress = (fields[1] == 2)
            timeReported = fields[7]
            return keycode, isKeyPress, timeReported

    def confirm(self, keycode):
        """
        Calling this method constitutes a request to confirm when the supplied
        I{keycode} is observed as entered. It immediately returns a C{Deferred}
        in response, which fires with C{True} if the keystroke is confirmed
        after a reasonable amount of time or with C{False} if not.

        Supplying something other than an integer as the I{keycode} will result
        in a C{Deferred} that has fired with C{False}, since no keystroke can
        be confirmed.
        """
        def cancelTimeout(result, callID):
            callID.cancel()
            return result

        def processError(failure, callID):
            print "OBSERVER ERROR, RESTARTING"
            d = restart()
            d.addCallback(lambda _: False)

        def timeout(d_Confirm):
            print "OBSERVER TIMEOUT, RESTARTING"
            if not d_Confirm.called:
                self.confirmationRequestFIFO.pop(0)
                d_NewProcess = restart()
                # Fire the confirmation request's Deferred from the new process
                # startup's Deferred
                d_NewProcess.chainDeferred(d_Confirm)

        def restart():
            # The confirmation deferred is timing out, so the confirmation
            # xnee process must have gone into zombie mode. Start a new one.
            d = self.shutdown()
            d.addCallback(self.startup)
            # When the new process has started, report a failure to confirm
            # the key event that inspired the process restart
            d.addCallback(lambda _: False)
            return d
        
        if isinstance(keycode, int):
            # Let's return a Deferred to the confirmation...
            d = defer.Deferred()
            # ...that times out if it waits too long...
            callID = reactor.callLater(MAX_WAIT, timeout, d)
            d.addCallback(cancelTimeout, callID)
            d.addErrback(processError, callID)
            # ...and put the new confirmation request into the FIFO
            confirmationRequest = (keycode, d)
            self.confirmationRequestFIFO.append(confirmationRequest)
        else:
            # We can't confirm an invalid keycode
            d = defer.succeed(False)
        return d