2

在 linux 机器（Debian wheezy）上，我正在尝试编写一个基于事件的服务器，它执行以下操作：

  1. 获取输入设备（特殊键盘）的独占输入，以防止击键进入通常的事件链。

  2. 注册扭曲反应堆中的事件

  3. 在等待事件返回的延迟处注册回调。然后，此回调将在收到特殊键序列后发送 HTTP 请求。

这是pyevdev包中的示例代码。它可以让我收到通知并相应地接收击键。

通过查看read_loop()命令的源代码，它也使用了select类似于 twisted 的语句。

我的问题

如何将此代码集成到 python Twisted 中？一个想法是查看底层字符设备/dev/input/event0并以非阻塞方式从中读取。如果 if 是一个常规文件，我会使用 inotify 类似的东西，但在这种情况下我不知道。

evdev 包中的示例代码

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')

# get exclusive access to input device
dev.grab()

for event in dev.read_loop():
    if event.type == ecodes.EV_KEY:
            print categorize(event)
4

1 回答 1

1

evdev.device.InputDevice有一个fileno()方法，这意味着你可以把它连接到一个 Twisted IReactorFDSet；几乎所有 Linux 上可用的反应器evdev都实现了这个接口。由于事件设备是一个带有文件描述符的对象，您通常只能从中读取，因此您需要IReadDescriptor包装它。

与您的示例大致相同逻辑的实现，但使用反应器处理事件，可能如下所示：

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()

请注意，此代码完全未经测试，因此一开始它可能无法正常工作，但它至少应该让您了解所需的内容。

于 2015-02-13T01:12:09.060 回答