forked from gvalkov/python-evdev
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy patheventio.py
More file actions
152 lines (117 loc) · 4.32 KB
/
eventio.py
File metadata and controls
152 lines (117 loc) · 4.32 KB
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
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
import fcntl
import functools
import os
import select
from typing import Iterator, Union
from . import _input, _uinput, ecodes
from .events import InputEvent
# --------------------------------------------------------------------------
class EvdevError(Exception):
pass
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).
"""
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) -> Iterator[InputEvent]:
"""
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) -> Union[InputEvent, None]:
"""
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) -> Iterator[InputEvent]:
"""
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)
# pylint: disable=no-self-argument
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:
# pylint: disable=not-callable
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: int, code: int, value: int):
"""
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)
def syn(self):
"""
Inject a ``SYN_REPORT`` event into the input subsystem. Events
queued by :func:`write()` will be fired. If possible, events
will be merged into an 'atomic' event.
"""
self.write(ecodes.EV_SYN, ecodes.SYN_REPORT, 0)
def close(self):
pass