1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
|
# encoding: utf-8
import os
import fcntl
import select
import functools
from evdev import _input, _uinput, ecodes, util
from evdev.events import InputEvent
#--------------------------------------------------------------------------
class EvdevError(Exception):
pass
class EventIO(object):
'''
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).
'''
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
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
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)
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)
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
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)
@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)
|
