1#!/usr/local/bin/python3.8
2# vim: set expandtab shiftwidth=4:
3# -*- Mode: python; coding: utf-8; indent-tabs-mode: nil -*- */
4#
5# Copyright © 2017 Red Hat, Inc.
6#
7# Permission is hereby granted, free of charge, to any person obtaining a
8# copy of this software and associated documentation files (the "Software"),
9# to deal in the Software without restriction, including without limitation
10# the rights to use, copy, modify, merge, publish, distribute, sublicense,
11# and/or sell copies of the Software, and to permit persons to whom the
12# Software is furnished to do so, subject to the following conditions:
13#
14# The above copyright notice and this permission notice (including the next
15# paragraph) shall be included in all copies or substantial portions of the
16# Software.
17#
18# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
19# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
20# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.  IN NO EVENT SHALL
21# THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
22# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
23# FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
24# DEALINGS IN THE SOFTWARE.
25#
26
27import sys
28import subprocess
29import argparse
30try:
31    import evdev
32    import evdev.ecodes
33    import pyudev
34except ModuleNotFoundError as e:
35    print('Error: {}'.format(str(e)), file=sys.stderr)
36    print('One or more python modules are missing. Please install those '
37          'modules and re-run this tool.')
38    sys.exit(1)
39
40
41class Range(object):
42    """Class to keep a min/max of a value around"""
43    def __init__(self):
44        self.min = float('inf')
45        self.max = float('-inf')
46
47    def update(self, value):
48        self.min = min(self.min, value)
49        self.max = max(self.max, value)
50
51
52class Touch(object):
53    """A single data point of a sequence (i.e. one event frame)"""
54
55    def __init__(self, pressure=None):
56        self.pressure = pressure
57
58
59class TouchSequence(object):
60    """A touch sequence from beginning to end"""
61
62    def __init__(self, device, tracking_id):
63        self.device = device
64        self.tracking_id = tracking_id
65        self.points = []
66
67        self.is_active = True
68
69        self.is_down = False
70        self.was_down = False
71        self.is_palm = False
72        self.was_palm = False
73        self.is_thumb = False
74        self.was_thumb = False
75
76        self.prange = Range()
77
78    def append(self, touch):
79        """Add a Touch to the sequence"""
80        self.points.append(touch)
81        self.prange.update(touch.pressure)
82
83        if touch.pressure < self.device.up:
84            self.is_down = False
85        elif touch.pressure > self.device.down:
86            self.is_down = True
87            self.was_down = True
88
89        self.is_palm = touch.pressure > self.device.palm
90        if self.is_palm:
91            self.was_palm = True
92
93        self.is_thumb = touch.pressure > self.device.thumb
94        if self.is_thumb:
95            self.was_thumb = True
96
97    def finalize(self):
98        """Mark the TouchSequence as complete (finger is up)"""
99        self.is_active = False
100
101    def avg(self):
102        """Average pressure value of this sequence"""
103        return int(sum([p.pressure for p in self.points])/len(self.points))
104
105    def median(self):
106        """Median pressure value of this sequence"""
107        ps = sorted([p.pressure for p in self.points])
108        idx = int(len(self.points)/2)
109        return ps[idx]
110
111    def __str__(self):
112        return self._str_state() if self.is_active else self._str_summary()
113
114    def _str_summary(self):
115        if not self.points:
116            return "{:78s}".format("Sequence: no pressure values recorded")
117
118        s = "Sequence {} pressure: "\
119            "min: {:3d} max: {:3d} avg: {:3d} median: {:3d} tags:" \
120            .format(
121                self.tracking_id,
122                self.prange.min,
123                self.prange.max,
124                self.avg(),
125                self.median()
126            )
127        if self.was_down:
128            s += " down"
129        if self.was_palm:
130            s += " palm"
131        if self.was_thumb:
132            s += " thumb"
133
134        return s
135
136    def _str_state(self):
137        s = "Touchpad pressure: {:3d} min: {:3d} max: {:3d} tags: {} {} {}" \
138            .format(
139                self.points[-1].pressure,
140                self.prange.min,
141                self.prange.max,
142                "down" if self.is_down else "    ",
143                "palm" if self.is_palm else "    ",
144                "thumb" if self.is_thumb else "     "
145            )
146        return s
147
148
149class InvalidDeviceError(Exception):
150    pass
151
152
153class Device(object):
154    def __init__(self, path):
155        if path is None:
156            self.path = self.find_touchpad_device()
157        else:
158            self.path = path
159
160        self.device = evdev.InputDevice(self.path)
161
162        print("Using {}: {}\n".format(self.device.name, self.path))
163
164        # capabilities rturns a dict with the EV_* codes as key,
165        # each of which is a list of tuples of (code, AbsInfo)
166        #
167        # Get the abs list first (or empty list if missing),
168        # then extract the pressure absinfo from that
169        all_caps = self.device.capabilities(absinfo=True)
170        caps = all_caps.get(evdev.ecodes.EV_ABS, [])
171        p = [cap[1] for cap in caps if cap[0] == evdev.ecodes.ABS_MT_PRESSURE]
172        if not p:
173            p = [cap[1] for cap in caps if cap[0] == evdev.ecodes.ABS_PRESSURE]
174            if not p:
175                raise InvalidDeviceError("device does not have ABS_PRESSURE/ABS_MT_PRESSURE")
176            self.has_mt_pressure = False
177        else:
178            self.has_mt_pressure = True
179
180        p = p[0]
181        prange = p.max - p.min
182
183        # libinput defaults
184        self.down = int(p.min + 0.12 * prange)
185        self.up = int(p.min + 0.10 * prange)
186        self.palm = 130  # the libinput default
187        self.thumb = p.max
188
189        self._init_thresholds_from_quirks()
190        self.sequences = []
191
192    def find_touchpad_device(self):
193        context = pyudev.Context()
194        for device in context.list_devices(subsystem='input'):
195            if not device.get('ID_INPUT_TOUCHPAD', 0):
196                continue
197
198            if not device.device_node or \
199               not device.device_node.startswith('/dev/input/event'):
200                continue
201
202            return device.device_node
203        print("Unable to find a touchpad device.", file=sys.stderr)
204        sys.exit(1)
205
206    def _init_thresholds_from_quirks(self):
207        command = ['libinput', 'quirks', 'list', self.path]
208        cmd = subprocess.run(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
209        if cmd.returncode != 0:
210            print("Error querying quirks: {}".format(cmd.stderr.decode('utf-8')), file=sys.stderr)
211            return
212
213        stdout = cmd.stdout.decode('utf-8')
214        quirks = [q.split('=') for q in stdout.split('\n')]
215
216        for q in quirks:
217            if q[0] == 'AttrPalmPressureThreshold':
218                self.palm = int(q[1])
219            elif q[0] == 'AttrPressureRange':
220                self.down, self.up = colon_tuple(q[1])
221            elif q[0] == 'AttrThumbPressureThreshold':
222                self.thumb = int(q[1])
223
224    def start_new_sequence(self, tracking_id):
225        self.sequences.append(TouchSequence(self, tracking_id))
226
227    def current_sequence(self):
228        return self.sequences[-1]
229
230
231def handle_key(device, event):
232    tapcodes = [
233        evdev.ecodes.BTN_TOOL_DOUBLETAP,
234        evdev.ecodes.BTN_TOOL_TRIPLETAP,
235        evdev.ecodes.BTN_TOOL_QUADTAP,
236        evdev.ecodes.BTN_TOOL_QUINTTAP
237    ]
238    if event.code in tapcodes and event.value > 0:
239        print("\rThis tool cannot handle multiple fingers, "
240              "output will be invalid", file=sys.stderr)
241
242
243def handle_abs(device, event):
244    if event.code == evdev.ecodes.ABS_MT_TRACKING_ID:
245        if event.value > -1:
246            device.start_new_sequence(event.value)
247        else:
248            try:
249                s = device.current_sequence()
250                s.finalize()
251                print("\r{}".format(s))
252            except IndexError:
253                # If the finger was down at startup
254                pass
255    elif ((event.code == evdev.ecodes.ABS_MT_PRESSURE) or
256          (event.code == evdev.ecodes.ABS_PRESSURE and not device.has_mt_pressure)):
257        try:
258            s = device.current_sequence()
259            s.append(Touch(pressure=event.value))
260            print("\r{}".format(s), end="")
261        except IndexError:
262            # If the finger was down at startup
263            pass
264
265
266def handle_event(device, event):
267    if event.type == evdev.ecodes.EV_ABS:
268        handle_abs(device, event)
269    elif event.type == evdev.ecodes.EV_KEY:
270        handle_key(device, event)
271
272
273def loop(device):
274    print("Ready for recording data.")
275    print("Pressure range used: {}:{}".format(device.down, device.up))
276    print("Palm pressure range used: {}".format(device.palm))
277    print("Thumb pressure range used: {}".format(device.thumb))
278    print("Place a single finger on the touchpad to measure pressure values.\n"
279          "Ctrl+C to exit\n")
280
281    for event in device.device.read_loop():
282        handle_event(device, event)
283
284
285def colon_tuple(string):
286    try:
287        ts = string.split(':')
288        t = tuple([int(x) for x in ts])
289        if len(t) == 2 and t[0] >= t[1]:
290            return t
291    except:
292        pass
293
294    msg = "{} is not in format N:M (N >= M)".format(string)
295    raise argparse.ArgumentTypeError(msg)
296
297
298def main(args):
299    parser = argparse.ArgumentParser(
300        description="Measure touchpad pressure values"
301    )
302    parser.add_argument(
303        'path', metavar='/dev/input/event0', nargs='?', type=str,
304        help='Path to device (optional)'
305    )
306    parser.add_argument(
307        '--touch-thresholds', metavar='down:up', type=colon_tuple,
308        help='Thresholds when a touch is logically down or up'
309    )
310    parser.add_argument(
311        '--palm-threshold', metavar='t', type=int,
312        help='Threshold when a touch is a palm'
313    )
314    args = parser.parse_args()
315
316    try:
317        device = Device(args.path)
318
319        if args.touch_thresholds is not None:
320            device.down, device.up = args.touch_thresholds
321
322        if args.palm_threshold is not None:
323            device.palm = args.palm_threshold
324
325        loop(device)
326    except KeyboardInterrupt:
327        pass
328    except (PermissionError, OSError):
329        print("Error: failed to open device")
330    except InvalidDeviceError as e:
331        print("Error: {}".format(e))
332
333
334if __name__ == "__main__":
335    main(sys.argv)
336