Integrating HID Access With Evdev On Linux With Python Twisted

On a linux machine (Debian wheezy) I am trying to write an event-based server that does the following: Grab exclusive input to the input device (a special keyboard) to prevent the

Solution 1:

evdev.device.InputDevice has a fileno() method , which means that you can hook it up to a Twisted IReactorFDSet; pretty much all reactors available on Linux where evdev is relevant implement this interface. Since an event device is an object with a file descriptor that you can mostly just read from, you need an IReadDescriptor to wrap it.

An implementation of roughly the same logic as your example, but using the reactor to process the events, might look like so:

from zope.interface import implementer
from twisted.internet.interfaces import IReadDescriptor
from twisted.logger import Logger

log = Logger()

@implementer(IReadDescriptor)
class InputDescriptor(object):
    def __init__(self, reactor, inputDevice, eventReceiver):
        self._reactor = reactor
        self._dev = inputDevice
        self._receiver = eventReceiver

    def fileno(self):
        return self._dev.fileno()

    def logPrefix(self):
        return "Input Device: " + repr(self._dev)

    def doRead(self):
        evt = self._dev.read_one()
        try:
            self._receiver.eventReceived(evt)
        except:
            log.failure("while dispatching HID event")

    def connectionLost(self, reason):
        self.stop()
        self._receiver.connectionLost(reason)

    def start(self):
        self._dev.grab()
        self._reactor.addReader(self)

    def stop(self):
        self._reactor.removeReader(self)
        self._dev.ungrab()

from evdev import InputDevice, categorize, ecodes, list_devices

devices = [InputDevice(fn) for fn in list_devices()]
for dev in devices:
   print(dev.fn, dev.name, dev.phys)

dev = InputDevice('/dev/input/event0')

class KeyReceiver(object):
    def eventReceived(self, event):
        if event.type == ecodes.EV_KEY:
            print(categorize(event))

    def connectionLost(self, reason):
        print("Event device lost!!", reason)

from twisted.internet import reactor
InputDescriptor(reactor, dev, KeyReceiver()).start()
reactor.run()

Please note that this code is totally untested, so it may not work quite right at first, but it should at least give you an idea of what is required.