Source code for evdev.eventio

import os
import fcntl
import select
import functools

from evdev import _input, _uinput, ecodes, util
from evdev.events import InputEvent


# --------------------------------------------------------------------------
class EvdevError(Exception):
    pass


[docs] class EventIO: """ Base class for reading and writing input events. This class is used by :class:`InputDevice` and :class:`UInput`. - On, :class:`InputDevice` it used for reading user-generated events (e.g. key presses, mouse movements) and writing feedback events (e.g. leds, beeps). - On, :class:`UInput` it used for writing user-generated events (e.g. key presses, mouse movements) and reading feedback events (e.g. leds, beeps). """
[docs] def fileno(self): """ Return the file descriptor to the open event device. This makes it possible to pass instances directly to :func:`select.select()` and :class:`asyncore.file_dispatcher`. """ return self.fd
[docs] def read_loop(self): """ Enter an endless :func:`select.select()` loop that yields input events. """ while True: r, w, x = select.select([self.fd], [], []) for event in self.read(): yield event
[docs] def read_one(self): """ Read and return a single input event as an instance of :class:`InputEvent <evdev.events.InputEvent>`. Return ``None`` if there are no pending input events. """ # event -> (sec, usec, type, code, val) event = _input.device_read(self.fd) if event: return InputEvent(*event)
[docs] def read(self): """ Read multiple input events from device. Return a generator object that yields :class:`InputEvent <evdev.events.InputEvent>` instances. Raises `BlockingIOError` if there are no available events at the moment. """ # events -> [(sec, usec, type, code, val), ...] events = _input.device_read_many(self.fd) for event in events: yield InputEvent(*event)
[docs] def need_write(func): """ Decorator that raises :class:`EvdevError` if there is no write access to the input device. """ @functools.wraps(func) def wrapper(*args): fd = args[0].fd if fcntl.fcntl(fd, fcntl.F_GETFL) & os.O_RDWR: return func(*args) msg = 'no write access to device "%s"' % args[0].path raise EvdevError(msg) return wrapper
[docs] def write_event(self, event): """ Inject an input event into the input subsystem. Events are queued until a synchronization event is received. Arguments --------- event: InputEvent InputEvent instance or an object with an ``event`` attribute (:class:`KeyEvent <evdev.events.KeyEvent>`, :class:`RelEvent <evdev.events.RelEvent>` etc). Example ------- >>> ev = InputEvent(1334414993, 274296, ecodes.EV_KEY, ecodes.KEY_A, 1) >>> ui.write_event(ev) """ if hasattr(event, "event"): event = event.event self.write(event.type, event.code, event.value)
[docs] @need_write def write(self, etype, code, value): """ Inject an input event into the input subsystem. Events are queued until a synchronization event is received. Arguments --------- etype event type (e.g. ``EV_KEY``). code event code (e.g. ``KEY_A``). value event value (e.g. 0 1 2 - depends on event type). Example --------- >>> ui.write(e.EV_KEY, e.KEY_A, 1) # key A - down >>> ui.write(e.EV_KEY, e.KEY_A, 0) # key A - up """ _uinput.write(self.fd, etype, code, value)
[docs] def close(self): pass