| # Copyright (c) 2014 The Chromium OS Authors. All rights reserved. |
| # Use of this source code is governed by a BSD-style license that can be |
| # found in the LICENSE file. |
| # |
| # |
| # Compute the touchpad drag latency |
| # |
| # This script computed the touch drag-latency given an evtest log and a |
| # Quickstep log that were collected while moving a finger at a fixed speed |
| # back and forth over the laser beam on the pad. It generates the numeric |
| # calculations (in seconds) as well as an image that visualizes the data |
| # for debugging purposes. This image is useful if the results are surprising, |
| # as it will often make it very obvious what the issue was. |
| # |
| # Usage: |
| # 1.) Align the Quickstep laser horizontally across the device. Make sure |
| # that the beam is low across the surface and squarely centered on the |
| # receiever. |
| # 2.) Begin evtest log collection while nothing is touching the device. |
| # evtest > evtest_log.txt |
| # 3.) Move a single finger at a constant speed back and forth over the laser |
| # on the pad at least 20 times. A robot will be able to produce more |
| # consistent results than a hand. |
| # 4.) Save a copy of the Quickstep log by accessing its "laser" sysfs entry |
| # find / -name 'laser' |
| # cat $the_file_you_found_above > quickstep_log.txt |
| # 5.) Crunch the numbers by passing both file names to this script |
| # python latency_measurement.py evtest_log.txt quickstep_log.txt |
| |
| import commands |
| import json |
| import numpy as np |
| import re |
| import sys |
| from collections import namedtuple |
| |
| import ppm |
| |
| |
| FingerPosition = namedtuple('FingerPosition', ['timestamp', 'x', 'y']) |
| LaserCrossing = namedtuple('LaserCrossing', ['timestamp', 'direction']) |
| |
| def get_evtest_timestamp(line): |
| return float(line.split(',')[0].split(' ')[-1]) |
| |
| def get_evtest_value(line): |
| return int(line.split(' ')[-1]) |
| |
| def get_finger_positions(filename): |
| """ Parse the finger positions out of an evtest log |
| This only works with a single contact. |
| Returns a list of FingerPosition objects |
| """ |
| log = open(filename, 'r') |
| if not log: |
| return None |
| |
| points = [] |
| fingers = {} |
| curr_id = -1 |
| last_x = last_y = 0 |
| |
| for line in log: |
| if not line.startswith('Event: '): |
| continue |
| |
| if 'ABS_MT_TRACKING_ID' in line: |
| curr_id = get_evtest_value(line) |
| if not fingers.get(curr_id, None): |
| fingers[curr_id] = {} |
| elif 'ABS_MT_POSITION_X' in line: |
| fingers[curr_id]['x'] = get_evtest_value(line) |
| elif 'ABS_MT_POSITION_Y' in line: |
| fingers[curr_id]['y'] = get_evtest_value(line) |
| elif 'SYN' in line and curr_id != -1 and curr_id in fingers: |
| t = get_evtest_timestamp(line) |
| last_x = x = fingers[curr_id].get('x', last_x) |
| last_y = y = fingers[curr_id].get('y', last_y) |
| points.append(FingerPosition(t, x, y)) |
| |
| log.close() |
| return points |
| |
| |
| def get_laser_crossings(filename): |
| """ Parse out the laser crossing events from a Quickstep log |
| Returns a list of LaserCrossing events |
| """ |
| QUICKSTEP_LOG_REGEX = '^\s*([\d.]+)\s+(\d)\s*$' |
| |
| status, raw_log = commands.getstatusoutput('cat %s' % filename) |
| if status: |
| return None |
| |
| laser_crossings = [] |
| for line in raw_log.splitlines(): |
| matches = re.match(QUICKSTEP_LOG_REGEX, line) |
| if matches and len(matches.groups()) == 2: |
| timestamp = float(matches.group(1)) |
| direction = int(matches.group(2)) |
| laser_crossings.append(LaserCrossing(timestamp, direction)) |
| |
| return laser_crossings |
| |
| |
| def clip_timeframes(positions, laser_crossings): |
| """ Clip both lists of events so they only contain events that occurred |
| during the period of time that *both* logs were being generated. |
| Returns the clipped lists |
| """ |
| if not positions or not laser_crossings: |
| return [], [] |
| |
| touchpad_start = positions[0].timestamp |
| touchpad_end = positions[-1].timestamp |
| laser_start = laser_crossings[0].timestamp |
| laser_end = laser_crossings[-1].timestamp |
| |
| positions = [p for p in positions |
| if (p.timestamp <= laser_end and p.timestamp >= laser_start)] |
| laser_crossings = [q for q in laser_crossings |
| if (q.timestamp <= touchpad_end and |
| q.timestamp >= touchpad_start)] |
| |
| return positions, laser_crossings |
| |
| |
| def get_laser_crossing_points(positions, laser_crossings): |
| """ Find the touchpad readings at the moment the laser was crossed for each |
| laser crossing event. Since the timestamps will not likely line up |
| perfectly, the exact position is interpolated between the two positions |
| read before and after the laser was crossed. |
| Returns a list of TouchpadPosition events |
| """ |
| crossing_points = [] |
| |
| for crossing in laser_crossings: |
| for i, position in enumerate(positions): |
| if i == 0: |
| continue |
| |
| if position.timestamp > crossing.timestamp: |
| # interpolate, since they won't line up perfectly |
| |
| time_gap = (position.timestamp - positions[i - 1].timestamp) |
| before_weight = ((position.timestamp - crossing.timestamp) / |
| time_gap) |
| after_weight = 1.0 - before_weight |
| |
| crossing_points.append( |
| FingerPosition(crossing.timestamp, |
| (positions[i - 1].x * before_weight + |
| position.x * after_weight), |
| (positions[i - 1].y * before_weight + |
| position.y * after_weight) |
| )) |
| break |
| |
| return crossing_points |
| |
| |
| def estimate_laser_line(laser_crossings): |
| """ Estimate the position of the line defined by the laser on the pad |
| Given the touchpad readings on either side of the line that were generated |
| at the moment the laser was crossed, you can fit a line to them and |
| discover where the line is inbetween then. |
| Returns a np.poly1d representation of the line |
| """ |
| line = np.poly1d(np.polyfit([x for t, x, y in laser_crossings], |
| [y for t, x, y in laser_crossings], |
| deg=1)) |
| return line |
| |
| |
| def which_side(line, coords): |
| """ Indicates which side of a line a given point lies on """ |
| distance = line(coords.x) - coords.y |
| if abs(distance) <= 2: return 0 |
| elif distance > 0: return 1 |
| else: return -1 |
| |
| |
| def get_touchpad_crossing_points(positions, line): |
| """ Find the touchpad readings of where it finally crossed the line formed |
| by the laser. These are the points that the user would see, and are likely |
| delayed somewhat, which is what we will try to measure. |
| Returns a list of TouchpadPosition events along the line |
| """ |
| last_side = which_side(line, positions[0]) |
| points = [] |
| for i, position in enumerate(positions): |
| current_side = which_side(line, position) |
| if current_side != last_side and last_side != 0: |
| points.append(position) |
| last_side = current_side |
| return points |
| |
| |
| def compute_latency(line_positions, laser_positions): |
| """ Measure how much time passed between the laser being triggered and the |
| touchpad reporting an event passed the line. |
| Returns a list of floats corresponding to the latency for each crossing |
| """ |
| latencies = [] |
| for (probe_timestamp, _, _), (laser_timestamp, _, _) \ |
| in zip(line_positions, laser_positions): |
| latencies.append(probe_timestamp - laser_timestamp) |
| return latencies |
| |
| def measure_latencies(finger_positions, laser_crossings, filename=None): |
| positions, laser_crossings = \ |
| clip_timeframes(finger_positions, laser_crossings) |
| if not positions or not laser_crossings: |
| print 'ERROR: There are no overlapping events in the input' |
| return [] |
| |
| # Find the points where the laser was crossed |
| laser_crossing_points = \ |
| get_laser_crossing_points(positions, laser_crossings) |
| |
| # Separate those points of two "lines" (the laser essentially defines two) |
| laser_crossing_points0 = [p for i, p in enumerate(laser_crossing_points) |
| if int((i + 1) / 2) % 2 == 0] |
| laser_crossing_points1 = [p for i, p in enumerate(laser_crossing_points) |
| if int((i + 1) / 2) % 2 == 1] |
| |
| # Fit a line to them to estimate where the actual laser is |
| line0 = estimate_laser_line(laser_crossing_points0) |
| line1 = estimate_laser_line(laser_crossing_points1) |
| |
| # Find the touchpad positions where the probe actually crossed the "line" |
| touchpad_crossing_points0 = get_touchpad_crossing_points(positions, line0) |
| touchpad_crossing_points1 = get_touchpad_crossing_points(positions, line1) |
| |
| # Actually do the timings against the ideal line |
| latencies0 = compute_latency(touchpad_crossing_points0, |
| laser_crossing_points0) |
| latencies1 = compute_latency(touchpad_crossing_points1, |
| laser_crossing_points1) |
| |
| if filename: |
| # Draw everything, and make a picture of the data for debugging |
| maxx = max([x for t, x, y in positions]) |
| maxy = max([y for t, x, y in positions]) |
| img = ppm.Image(maxx + 50, maxy + 50) |
| |
| last_x, last_y = None, None |
| for t, x, y in positions: |
| img.Circle((x, y), ppm.BLACK, radius=2) |
| if not last_x is None: |
| img.Line((last_x, last_y), (x, y), ppm.BLACK) |
| last_x, last_y = x, y |
| |
| for t, x, y in touchpad_crossing_points0: |
| img.Circle((x, y), ppm.BLACK, radius=6) |
| for t, x, y in touchpad_crossing_points1: |
| img.Circle((x, y), ppm.BLACK, radius=6) |
| |
| for i, (t, x, y) in enumerate(laser_crossing_points): |
| img.Circle((x, y), ppm.BLACK, radius=6) |
| for i, (t, x, y) in enumerate(laser_crossing_points0): |
| img.Circle((x, y), ppm.MAGENTA if i % 2 else ppm.RED) |
| for i, (t, x, y) in enumerate(laser_crossing_points1): |
| img.Circle((x, y), ppm.YELLOW if i % 2 else ppm.CYAN) |
| |
| img.Line((0, line0(0)), (maxx + 50, line0(maxx + 50)), ppm.BLUE) |
| img.Line((0, line1(0)), (maxx + 50, line1(maxx + 50)), ppm.GREEN) |
| img.Save(filename) |
| |
| return latencies0 + latencies1 |
| |
| |
| def main(evtest_log_filename, quickstep_log_filename): |
| positions = get_finger_positions(evtest_log_filename) |
| laser_crossings = get_laser_crossings(quickstep_log_filename) |
| |
| latencies = measure_latencies(positions, laser_crossings, 'out.ppm') |
| if not latencies: |
| print 'ERROR: Unable to compute any latencies' |
| return |
| |
| # Display the results |
| print 'Average', 'Maximum', 'Minimum' |
| print np.average(latencies), max(latencies), min(latencies) |
| print |
| print 'Latency result: %f ms' % (np.average(latencies) * 1000.0) |
| |
| |
| if __name__ == '__main__': |
| if len(sys.argv) != 3: |
| print 'Usage: %s evtest_log.txt quickstep_log.txt' % sys.argv[0] |
| sys.exit(1) |
| evtest_log_filename = sys.argv[1] |
| quickstep_log_filename = sys.argv[2] |
| main(evtest_log_filename, quickstep_log_filename) |