<commit_before>#!/usr/bin/env python
#
# Copyright 2016 The Chromium OS Authors. All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#

"""
Runs the touchpad drag latency test using WALT Latency Timer
Usage example:
    $ python walt.py 11
    Input device   : /dev/input/event11
    Serial device  : /dev/ttyACM1
    Laser log file : /tmp/WALT_2016_06_23__1714_51_laser.log
    evtest log file: /tmp/WALT_2016_06_23__1714_51_evtest.log
    Clock zeroed at 1466716492 (rt 0.284ms)
    ........................................
    Processing data, may take a minute or two...
    Drag latency (min method) = 15.37 ms

Note, before running this script, check that evtest can grab the device.
On some systems it requires running as root.
"""

import argparse
import contextlib
import glob
import os
import random
import re
import socket
import subprocess
import sys
import tempfile
import threading
import time

import serial
import numpy

import evparser
import minimization
import screen_stats


# Time units
MS = 1e-3  # MS = 0.001 seconds
US = 1e-6  # US = 10^-6 seconds

# Globals
debug_mode = True


def log(msg):
    if debug_mode:
        print(msg)


class Walt(object):
    """ A class for communicating with Wlat device

    Usage:
    with Walt('/dev/ttyUSB0') as walt:
        body....


    """

    # Teensy commands, always singe char. Defined in WALT.ino
    # github.com/google/walt/blob/master/arduino/walt/walt.ino
    CMD_RESET = 'F'
    CMD_PING = 'P'
    CMD_SYNC_ZERO = 'Z'
    CMD_SYNC_SEND = 'I'
    CMD_SYNC_READOUT = 'R'
    CMD_TIME_NOW = 'T'
    CMD_AUTO_LASER_ON = 'L'
    CMD_AUTO_LASER_OFF = 'l'
    CMD_AUTO_SCREEN_ON = 'C'
    CMD_AUTO_SCREEN_OFF = 'c'
    CMD_GSHOCK = 'G'
    CMD_VERSION = 'V'
    CMD_SAMPLE_ALL = 'Q'
    CMD_BRIGHTNESS_CURVE = 'U'


    def __init__(self, serial_dev, timeout=None):
        self.serial_dev = serial_dev
        self.ser = serial.Serial(serial_dev, baudrate=115200, timeout=timeout)
        self.base_time = None
        self.min_lag = None
        self.max_lag = None
        self.median_latency = None

    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc_value, traceback):
        try:
            self.ser.close()
        except:
            pass

    def close(self):
        self.ser.close()

    def readline(self):
        return self.ser.readline()

    def sndrcv(self, data):
        """ Send a 1-char command.
        Return the reply and how long it took.

        """
        t0 = time.time()
        self.ser.write(data)
        reply = self.ser.readline()
        t1 = time.time()
        dt = (t1 - t0)
        log('sndrcv(): round trip %.3fms, reply=%s' % (dt / MS, reply.strip()))
        return dt, reply

    def read_shock_time(self):
        dt, s = self.sndrcv(Walt.CMD_GSHOCK)
        t_us = int(s.strip())
        return t_us


    def run_comm_stats(self, N=100):
        """
        Measure the USB serial round trip time.
        Send CMD_TIME_NOW to the Teensy N times measuring the round trip each time.
        Prints out stats (min, median, max).

        """
        log('Running USB comm stats...')
        self.ser.flushInput()
        self.sndrcv(Walt.CMD_SYNC_ZERO)
        tstart = time.time()
        times = numpy.zeros((N, 1))
        for i in range(N):
            dt, _ = self.sndrcv(Walt.CMD_TIME_NOW)
            times[i] = dt
        t_total = time.time() - tstart

        median = numpy.median(times)
        stats = (times.min() / MS, median / MS, times.max() / MS, N)
        self.median_latency = median
        log('USB comm round trip stats:')
        log('min=%.2fms, median=%.2fms, max=%.2fms N=%d' % stats)
        if (median > 2):
            print('ERROR: the median round trip is too high: %.2f ms' % (median / MS) )
            sys.exit(2)

    def zero_clock(self, max_delay=0.001, retries=10):
        """
        Tell the TeensyUSB to zero its clock (CMD_SYNC_ZERO).
        Returns the time when the command was sent.
        Verify that the response arrived within max_delay seconds.

        This is the simple zeroing used when the round trip is fast.
        It does not employ the same method as Android clock sync.
        """

        # Check that we get reasonable ping time with Teensy
        # this also 'warms up' the comms, first msg is often slower
        self.run_comm_stats(N=10)

        self.ser.flushInput()

        for i in range(retries):
            t0 = time.time()
            dt, _ = self.sndrcv(Walt.CMD_SYNC_ZERO)
            if dt < max_delay:
                print('Clock zeroed at %.0f (rt %.3f ms)' % (t0, dt / MS))
                self.base_time = t0
                self.max_lag = dt
                self.min_lag = 0
                return t0
        print('Error, failed to zero the clock after %d retries')
        return -1

    def read_remote_times(self):
        """ Helper func, see doc string in estimate_lage()
        Read out the timestamps taken recorded by the Teensy.
        """
        times = numpy.zeros(9)
        for i in range(9):
            dt, reply = self.sndrcv(Walt.CMD_SYNC_READOUT)
            num, tstamp = reply.strip().split(':')
            # TODO: verify that num is what we expect it to be
            log('read_remote_times() CMD_SYNC_READOUT > w >  = %s' % reply)
            t = float(tstamp) * US  # WALT sends timestamps in microseconds
            times[i] = t
        return times

    def estimate_lag(self):
        """ Estimate the difference between local and remote clocks

        This is based on:
        github.com/google/walt/blob/master/android/WALT/app/src/main/jni/README.md

        self.base_time needs to be set using self.zero_clock() before running
        this function.

        The result is saved as self.min_lag and self.max_lag. Assume that the
        remote clock lags behind the local by `lag` That is, at a given moment
        local_time = remote_time + lag
        where local_time = time.time() - self.base_time

        Immediately after this function completes the lag is guaranteed to be
        between min_lag and max_lag. But the lag change (drift) away with time.
        """
        self.ser.flushInput()

        # remote -> local
        times_local_received = numpy.zeros(9)
        self.ser.write(Walt.CMD_SYNC_SEND)
        for i in range(9):
            reply = self.ser.readline()
            times_local_received[i] = time.time() - self.base_time

        times_remote_sent = self.read_remote_times()
        max_lag = (times_local_received - times_remote_sent).min()

        # local -> remote
        times_local_sent = numpy.zeros(9)
        for i in range(9):
            s = '%d' % (i + 1)
            # Sleep between the messages to combat buffering
            t_sleep = US * random.randint(70, 700)
            time.sleep(t_sleep)
            times_local_sent[i] = time.time() - self.base_time
            self.ser.write(s)

        times_remote_received = self.read_remote_times()
        min_lag = (times_local_sent - times_remote_received).max()

        self.min_lag = min_lag
        self.max_lag = max_lag

    def parse_trigger(self, trigger_line):
        """ Parse a trigger line from WALT.

        Trigger events look like this: "G L 12902345 1 1"
        The parts:
         * G - common for all trigger events
         * L - means laser
         * 12902345 is timestamp in us since zeroed
         * 1st 1 or 0 is trigger value. 0 = changed to dark, 1 = changed to light,
         * 2nd 1 is counter of how many times this trigger happened since last
           readout, should always be 1 in our case

        """

        parts = trigger_line.strip().split()
        if len(parts) != 5:
            raise Exception('Malformed trigger line: "%s"\n' % trigger_line)
        t_us = int(parts[2])
        val = int(parts[3])
        return (t_us / 1e6, val)


def array2str(a):
    a_strs = ['%0.2f' % x for x in a]
    s = ', '.join(a_strs)
    return '[' + s + ']'


def parse_args():
    temp_dir = tempfile.gettempdir()
    serial = '/dev/ttyACM0'

    # Try to autodetect the WALT serial port
    ls_ttyACM = glob.glob('/dev/ttyACM*')
    if len(ls_ttyACM) > 0:
        serial = ls_ttyACM[0]

    description = "Run the touchpad drag latency test using WALT Latency Timer"
    parser = argparse.ArgumentParser(
        description=description,
        formatter_class=argparse.ArgumentDefaultsHelpFormatter)

    parser.add_argument('-i', '--input', default='',
                        help='input device, e.g: 6 or /dev/input/event6')
    parser.add_argument('-s', '--serial', default=serial,
                        help='WALT serial port')
    parser.add_argument('-t', '--type', default='drag',
                        help='Test type: drag|tap|screen|sanity|curve|bridge')
    parser.add_argument('-l', '--logdir', default=temp_dir,
                        help='where to store logs')
    parser.add_argument('-n', default=40, type=int,
                        help='Number of laser toggles to read')
    parser.add_argument('-p', '--port', default=50007, type=int,
                        help='Port to listen on for the TCP bridge')
    parser.add_argument('-d', '--debug', action='store_true',
                        help='talk more')
    args = parser.parse_args()

    global debug_mode
    debug_mode = args.debug

    if args.input.isalnum():
        args.input = '/dev/input/event' + args.input

    return args


def run_drag_latency_test(args):

    if not args.input:
        print('Error: --input argument is required for drag latency test')
        sys.exit(1)

    # Create names for log files
    prefix = time.strftime('WALT_%Y_%m_%d__%H%M_%S')
    laser_file_name = os.path.join(args.logdir,  prefix + '_laser.log')
    evtest_file_name = os.path.join(args.logdir,  prefix + '_evtest.log')

    print('Starting drag latency test')
    print('Input device   : ' + args.input)
    print('Serial device  : ' + args.serial)
    print('Laser log file : ' + laser_file_name)
    print('evtest log file: ' + evtest_file_name)

    with Walt(args.serial) as walt:
        walt.sndrcv(Walt.CMD_RESET)
        tstart = time.time()
        t_zero = walt.zero_clock()
        if t_zero < 0:
            print('Error: Couldn\'t zero clock, exitting')
            sys.exit(1)

        # Fire up the evtest process
        cmd = 'evtest %s > %s' % (args.input, evtest_file_name)
        evtest = subprocess.Popen(cmd, shell=True)

        # Turn on laser trigger auto-sending
        walt.sndrcv(Walt.CMD_AUTO_LASER_ON)
        trigger_count = 0
        while trigger_count < args.n:
            # The following line blocks until a message from WALT arrives
            trigger_line = walt.readline()
            trigger_count += 1
            log('#%d/%d - ' % (trigger_count, args.n) +
                trigger_line.strip())

            if not debug_mode:
                sys.stdout.write('.')
                sys.stdout.flush()

            t, val = walt.parse_trigger(trigger_line)
            t += t_zero
            with open(laser_file_name, 'at') as flaser:
                flaser.write('%.3f %d\n' % (t, val))
        walt.sndrcv(Walt.CMD_AUTO_LASER_OFF)

    # Send SIGTERM to evtest process
    evtest.terminate()

    print "\nProcessing data, may take a minute or two..."
    # lm.main(evtest_file_name, laser_file_name)
    minimization.minimize(evtest_file_name, laser_file_name)

def run_screen_curve(args):

    with Walt(args.serial, timeout=1) as walt:
        walt.sndrcv(Walt.CMD_RESET)

        t_zero = walt.zero_clock()
        if t_zero < 0:
            print('Error: Couldn\'t zero clock, exitting')
            sys.exit(1)

        # Fire up the walt_blinker process
        cmd = 'walt_blink 1'
        blinker = subprocess.Popen(cmd, shell=True)

        # Turn on laser trigger auto-sending
        walt.sndrcv(Walt.CMD_BRIGHTNESS_CURVE)
        s = 'dummy'
        while s:
            s = walt.readline()
            print(s.strip())


def run_screen_latency_test(args):

    # Create names for log files
    prefix = time.strftime('WALT_%Y_%m_%d__%H%M_%S')
    sensor_file_name = os.path.join(args.logdir,  prefix + '_screen_sensor.log')
    blinker_file_name = os.path.join(args.logdir,  prefix + '_blinker.log')

    print('Starting drag latency test')
    print('Input device   : ' + args.input)
    print('Serial device  : ' + args.serial)
    print('Laser log file : ' + sensor_file_name)
    print('evtest log file: ' + blinker_file_name)

    with Walt(args.serial, timeout=1) as walt:
        walt.sndrcv(Walt.CMD_RESET)

        t_zero = walt.zero_clock()
        if t_zero < 0:
            print('Error: Couldn\'t zero clock, exitting')
            sys.exit(1)

        # Fire up the walt_blinker process
        cmd = 'walt_blink %d > %s' % (args.n, blinker_file_name, )
        blinker = subprocess.Popen(cmd, shell=True)

        # Turn on laser trigger auto-sending
        walt.sndrcv(Walt.CMD_AUTO_SCREEN_ON)
        trigger_count = 0

        # Iterate while the blinker process is alive
        # TODO: re-sync clocks every once in a while
        while blinker.poll() is None:
            # The following line blocks until a message from WALT arrives
            trigger_line = walt.readline()
            if not trigger_line:
                # This usually happens when readline timeouts on last iteration
                continue
            trigger_count += 1
            log('#%d/%d - ' % (trigger_count, args.n) +
                trigger_line.strip())

            if not debug_mode:
                sys.stdout.write('.')
                sys.stdout.flush()

            t, val = walt.parse_trigger(trigger_line)
            t += t_zero
            with open(sensor_file_name, 'at') as flaser:
                flaser.write('%.3f %d\n' % (t, val))
        walt.sndrcv(Walt.CMD_AUTO_SCREEN_OFF)

    print("\nProcessing data ...")
    screen_stats.screen_stats(blinker_file_name, sensor_file_name)


def run_tap_latency_test(args):

    if not args.input:
        print('Error: --input argument is required for tap latency test')
        sys.exit(1)

    print('Starting tap latency test')

    with Walt(args.serial) as walt:
        walt.sndrcv(Walt.CMD_RESET)
        tstart = time.time()
        t_zero = walt.zero_clock()
        if t_zero < 0:
            print('Error: Couldn\'t zero clock, exitting')
            sys.exit(1)

        # Fire up the evtest process
        cmd = 'evtest ' + args.input
        evtest = subprocess.Popen(cmd, shell=True, stdout=subprocess.PIPE, bufsize=1, universal_newlines=True)
        walt.sndrcv(Walt.CMD_GSHOCK)

        taps_detected = 0
        taps = []
        while taps_detected < args.n:
            ev_line = evtest.stdout.readline()
            tap_info = evparser.parse_tap_line(ev_line)
            if not tap_info:
                continue

            # Just received a tap event from evtest
            taps_detected += 1

            t_tap_epoch, direction = tap_info
            shock_time_us = walt.read_shock_time()
            dt_tap_us = 1e6 * (t_tap_epoch - t_zero) - shock_time_us

            print ev_line
            print("shock t %d, tap t %f, tap val %d. dt=%0.1f" % (shock_time_us, t_tap_epoch, direction, dt_tap_us))

            if shock_time_us == 0:
                print "No shock detected, skipping this event"
                continue

            taps.append((dt_tap_us, direction))

    evtest.terminate()

    # Process data
    print("\nProcessing data...")
    dt_down = numpy.array([t[0] for t in taps if t[1] == 1]) / 1e3
    dt_up = numpy.array([t[0] for t in taps if t[1] == 0]) / 1e3

    print('dt_down = ' + array2str(dt_down))
    print('dt_up = ' + array2str(dt_up))

    median_down_ms = numpy.median(dt_down)
    meidan_up_ms = numpy.median(dt_up)

    print('Median latency, down: %0.1f, up: %0.1f' % (median_down_ms, meidan_up_ms))


def run_walt_sanity_test(args):
    print('Starting sanity test')

    with Walt(args.serial) as walt:
        walt.sndrcv(Walt.CMD_RESET)
<commit_msg>Fix printout of evtest events during tap test in pywalt<commit_after>#!/usr/bin/env python
#
# Copyright 2016 The Chromium OS Authors. All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#

"""
Runs the touchpad drag latency test using WALT Latency Timer
Usage example:
    $ python walt.py 11
    Input device   : /dev/input/event11
    Serial device  : /dev/ttyACM1
    Laser log file : /tmp/WALT_2016_06_23__1714_51_laser.log
    evtest log file: /tmp/WALT_2016_06_23__1714_51_evtest.log
    Clock zeroed at 1466716492 (rt 0.284ms)
    ........................................
    Processing data, may take a minute or two...
    Drag latency (min method) = 15.37 ms

Note, before running this script, check that evtest can grab the device.
On some systems it requires running as root.
"""

import argparse
import contextlib
import glob
import os
import random
import re
import socket
import subprocess
import sys
import tempfile
import threading
import time

import serial
import numpy

import evparser
import minimization
import screen_stats


# Time units
MS = 1e-3  # MS = 0.001 seconds
US = 1e-6  # US = 10^-6 seconds

# Globals
debug_mode = True


def log(msg):
    if debug_mode:
        print(msg)


class Walt(object):
    """ A class for communicating with Wlat device

    Usage:
    with Walt('/dev/ttyUSB0') as walt:
        body....


    """

    # Teensy commands, always singe char. Defined in WALT.ino
    # github.com/google/walt/blob/master/arduino/walt/walt.ino
    CMD_RESET = 'F'
    CMD_PING = 'P'
    CMD_SYNC_ZERO = 'Z'
    CMD_SYNC_SEND = 'I'
    CMD_SYNC_READOUT = 'R'
    CMD_TIME_NOW = 'T'
    CMD_AUTO_LASER_ON = 'L'
    CMD_AUTO_LASER_OFF = 'l'
    CMD_AUTO_SCREEN_ON = 'C'
    CMD_AUTO_SCREEN_OFF = 'c'
    CMD_GSHOCK = 'G'
    CMD_VERSION = 'V'
    CMD_SAMPLE_ALL = 'Q'
    CMD_BRIGHTNESS_CURVE = 'U'


    def __init__(self, serial_dev, timeout=None):
        self.serial_dev = serial_dev
        self.ser = serial.Serial(serial_dev, baudrate=115200, timeout=timeout)
        self.base_time = None
        self.min_lag = None
        self.max_lag = None
        self.median_latency = None

    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc_value, traceback):
        try:
            self.ser.close()
        except:
            pass

    def close(self):
        self.ser.close()

    def readline(self):
        return self.ser.readline()

    def sndrcv(self, data):
        """ Send a 1-char command.
        Return the reply and how long it took.

        """
        t0 = time.time()
        self.ser.write(data)
        reply = self.ser.readline()
        t1 = time.time()
        dt = (t1 - t0)
        log('sndrcv(): round trip %.3fms, reply=%s' % (dt / MS, reply.strip()))
        return dt, reply

    def read_shock_time(self):
        dt, s = self.sndrcv(Walt.CMD_GSHOCK)
        t_us = int(s.strip())
        return t_us


    def run_comm_stats(self, N=100):
        """
        Measure the USB serial round trip time.
        Send CMD_TIME_NOW to the Teensy N times measuring the round trip each time.
        Prints out stats (min, median, max).

        """
        log('Running USB comm stats...')
        self.ser.flushInput()
        self.sndrcv(Walt.CMD_SYNC_ZERO)
        tstart = time.time()
        times = numpy.zeros((N, 1))
        for i in range(N):
            dt, _ = self.sndrcv(Walt.CMD_TIME_NOW)
            times[i] = dt
        t_total = time.time() - tstart

        median = numpy.median(times)
        stats = (times.min() / MS, median / MS, times.max() / MS, N)
        self.median_latency = median
        log('USB comm round trip stats:')
        log('min=%.2fms, median=%.2fms, max=%.2fms N=%d' % stats)
        if (median > 2):
            print('ERROR: the median round trip is too high: %.2f ms' % (median / MS) )
            sys.exit(2)

    def zero_clock(self, max_delay=0.001, retries=10):
        """
        Tell the TeensyUSB to zero its clock (CMD_SYNC_ZERO).
        Returns the time when the command was sent.
        Verify that the response arrived within max_delay seconds.

        This is the simple zeroing used when the round trip is fast.
        It does not employ the same method as Android clock sync.
        """

        # Check that we get reasonable ping time with Teensy
        # this also 'warms up' the comms, first msg is often slower
        self.run_comm_stats(N=10)

        self.ser.flushInput()

        for i in range(retries):
            t0 = time.time()
            dt, _ = self.sndrcv(Walt.CMD_SYNC_ZERO)
            if dt < max_delay:
                print('Clock zeroed at %.0f (rt %.3f ms)' % (t0, dt / MS))
                self.base_time = t0
                self.max_lag = dt
                self.min_lag = 0
                return t0
        print('Error, failed to zero the clock after %d retries')
        return -1

    def read_remote_times(self):
        """ Helper func, see doc string in estimate_lage()
        Read out the timestamps taken recorded by the Teensy.
        """
        times = numpy.zeros(9)
        for i in range(9):
            dt, reply = self.sndrcv(Walt.CMD_SYNC_READOUT)
            num, tstamp = reply.strip().split(':')
            # TODO: verify that num is what we expect it to be
            log('read_remote_times() CMD_SYNC_READOUT > w >  = %s' % reply)
            t = float(tstamp) * US  # WALT sends timestamps in microseconds
            times[i] = t
        return times

    def estimate_lag(self):
        """ Estimate the difference between local and remote clocks

        This is based on:
        github.com/google/walt/blob/master/android/WALT/app/src/main/jni/README.md

        self.base_time needs to be set using self.zero_clock() before running
        this function.

        The result is saved as self.min_lag and self.max_lag. Assume that the
        remote clock lags behind the local by `lag` That is, at a given moment
        local_time = remote_time + lag
        where local_time = time.time() - self.base_time

        Immediately after this function completes the lag is guaranteed to be
        between min_lag and max_lag. But the lag change (drift) away with time.
        """
        self.ser.flushInput()

        # remote -> local
        times_local_received = numpy.zeros(9)
        self.ser.write(Walt.CMD_SYNC_SEND)
        for i in range(9):
            reply = self.ser.readline()
            times_local_received[i] = time.time() - self.base_time

        times_remote_sent = self.read_remote_times()
        max_lag = (times_local_received - times_remote_sent).min()

        # local -> remote
        times_local_sent = numpy.zeros(9)
        for i in range(9):
            s = '%d' % (i + 1)
            # Sleep between the messages to combat buffering
            t_sleep = US * random.randint(70, 700)
            time.sleep(t_sleep)
            times_local_sent[i] = time.time() - self.base_time
            self.ser.write(s)

        times_remote_received = self.read_remote_times()
        min_lag = (times_local_sent - times_remote_received).max()

        self.min_lag = min_lag
        self.max_lag = max_lag

    def parse_trigger(self, trigger_line):
        """ Parse a trigger line from WALT.

        Trigger events look like this: "G L 12902345 1 1"
        The parts:
         * G - common for all trigger events
         * L - means laser
         * 12902345 is timestamp in us since zeroed
         * 1st 1 or 0 is trigger value. 0 = changed to dark, 1 = changed to light,
         * 2nd 1 is counter of how many times this trigger happened since last
           readout, should always be 1 in our case

        """

        parts = trigger_line.strip().split()
        if len(parts) != 5:
            raise Exception('Malformed trigger line: "%s"\n' % trigger_line)
        t_us = int(parts[2])
        val = int(parts[3])
        return (t_us / 1e6, val)


def array2str(a):
    a_strs = ['%0.2f' % x for x in a]
    s = ', '.join(a_strs)
    return '[' + s + ']'


def parse_args():
    temp_dir = tempfile.gettempdir()
    serial = '/dev/ttyACM0'

    # Try to autodetect the WALT serial port
    ls_ttyACM = glob.glob('/dev/ttyACM*')
    if len(ls_ttyACM) > 0:
        serial = ls_ttyACM[0]

    description = "Run the touchpad drag latency test using WALT Latency Timer"
    parser = argparse.ArgumentParser(
        description=description,
        formatter_class=argparse.ArgumentDefaultsHelpFormatter)

    parser.add_argument('-i', '--input', default='',
                        help='input device, e.g: 6 or /dev/input/event6')
    parser.add_argument('-s', '--serial', default=serial,
                        help='WALT serial port')
    parser.add_argument('-t', '--type', default='drag',
                        help='Test type: drag|tap|screen|sanity|curve|bridge')
    parser.add_argument('-l', '--logdir', default=temp_dir,
                        help='where to store logs')
    parser.add_argument('-n', default=40, type=int,
                        help='Number of laser toggles to read')
    parser.add_argument('-p', '--port', default=50007, type=int,
                        help='Port to listen on for the TCP bridge')
    parser.add_argument('-d', '--debug', action='store_true',
                        help='talk more')
    args = parser.parse_args()

    global debug_mode
    debug_mode = args.debug

    if args.input.isalnum():
        args.input = '/dev/input/event' + args.input

    return args


def run_drag_latency_test(args):

    if not args.input:
        print('Error: --input argument is required for drag latency test')
        sys.exit(1)

    # Create names for log files
    prefix = time.strftime('WALT_%Y_%m_%d__%H%M_%S')
    laser_file_name = os.path.join(args.logdir,  prefix + '_laser.log')
    evtest_file_name = os.path.join(args.logdir,  prefix + '_evtest.log')

    print('Starting drag latency test')
    print('Input device   : ' + args.input)
    print('Serial device  : ' + args.serial)
    print('Laser log file : ' + laser_file_name)
    print('evtest log file: ' + evtest_file_name)

    with Walt(args.serial) as walt:
        walt.sndrcv(Walt.CMD_RESET)
        tstart = time.time()
        t_zero = walt.zero_clock()
        if t_zero < 0:
            print('Error: Couldn\'t zero clock, exitting')
            sys.exit(1)

        # Fire up the evtest process
        cmd = 'evtest %s > %s' % (args.input, evtest_file_name)
        evtest = subprocess.Popen(cmd, shell=True)

        # Turn on laser trigger auto-sending
        walt.sndrcv(Walt.CMD_AUTO_LASER_ON)
        trigger_count = 0
        while trigger_count < args.n:
            # The following line blocks until a message from WALT arrives
            trigger_line = walt.readline()
            trigger_count += 1
            log('#%d/%d - ' % (trigger_count, args.n) +
                trigger_line.strip())

            if not debug_mode:
                sys.stdout.write('.')
                sys.stdout.flush()

            t, val = walt.parse_trigger(trigger_line)
            t += t_zero
            with open(laser_file_name, 'at') as flaser:
                flaser.write('%.3f %d\n' % (t, val))
        walt.sndrcv(Walt.CMD_AUTO_LASER_OFF)

    # Send SIGTERM to evtest process
    evtest.terminate()

    print "\nProcessing data, may take a minute or two..."
    # lm.main(evtest_file_name, laser_file_name)
    minimization.minimize(evtest_file_name, laser_file_name)

def run_screen_curve(args):

    with Walt(args.serial, timeout=1) as walt:
        walt.sndrcv(Walt.CMD_RESET)

        t_zero = walt.zero_clock()
        if t_zero < 0:
            print('Error: Couldn\'t zero clock, exitting')
            sys.exit(1)

        # Fire up the walt_blinker process
        cmd = 'walt_blink 1'
        blinker = subprocess.Popen(cmd, shell=True)

        # Turn on laser trigger auto-sending
        walt.sndrcv(Walt.CMD_BRIGHTNESS_CURVE)
        s = 'dummy'
        while s:
            s = walt.readline()
            print(s.strip())


def run_screen_latency_test(args):

    # Create names for log files
    prefix = time.strftime('WALT_%Y_%m_%d__%H%M_%S')
    sensor_file_name = os.path.join(args.logdir,  prefix + '_screen_sensor.log')
    blinker_file_name = os.path.join(args.logdir,  prefix + '_blinker.log')

    print('Starting drag latency test')
    print('Input device   : ' + args.input)
    print('Serial device  : ' + args.serial)
    print('Laser log file : ' + sensor_file_name)
    print('evtest log file: ' + blinker_file_name)

    with Walt(args.serial, timeout=1) as walt:
        walt.sndrcv(Walt.CMD_RESET)

        t_zero = walt.zero_clock()
        if t_zero < 0:
            print('Error: Couldn\'t zero clock, exitting')
            sys.exit(1)

        # Fire up the walt_blinker process
        cmd = 'walt_blink %d > %s' % (args.n, blinker_file_name, )
        blinker = subprocess.Popen(cmd, shell=True)

        # Turn on laser trigger auto-sending
        walt.sndrcv(Walt.CMD_AUTO_SCREEN_ON)
        trigger_count = 0

        # Iterate while the blinker process is alive
        # TODO: re-sync clocks every once in a while
        while blinker.poll() is None:
            # The following line blocks until a message from WALT arrives
            trigger_line = walt.readline()
            if not trigger_line:
                # This usually happens when readline timeouts on last iteration
                continue
            trigger_count += 1
            log('#%d/%d - ' % (trigger_count, args.n) +
                trigger_line.strip())

            if not debug_mode:
                sys.stdout.write('.')
                sys.stdout.flush()

            t, val = walt.parse_trigger(trigger_line)
            t += t_zero
            with open(sensor_file_name, 'at') as flaser:
                flaser.write('%.3f %d\n' % (t, val))
        walt.sndrcv(Walt.CMD_AUTO_SCREEN_OFF)

    print("\nProcessing data ...")
    screen_stats.screen_stats(blinker_file_name, sensor_file_name)


def run_tap_latency_test(args):

    if not args.input:
        print('Error: --input argument is required for tap latency test')
        sys.exit(1)

    print('Starting tap latency test')

    with Walt(args.serial) as walt:
        walt.sndrcv(Walt.CMD_RESET)
        tstart = time.time()
        t_zero = walt.zero_clock()
        if t_zero < 0:
            print('Error: Couldn\'t zero clock, exitting')
            sys.exit(1)

        # Fire up the evtest process
        cmd = 'evtest ' + args.input
        evtest = subprocess.Popen(cmd, shell=True, stdout=subprocess.PIPE, bufsize=1, universal_newlines=True)
        walt.sndrcv(Walt.CMD_GSHOCK)

        taps_detected = 0
        taps = []
        while taps_detected < args.n:
            ev_line = evtest.stdout.readline()
            tap_info = evparser.parse_tap_line(ev_line)
            if not tap_info:
                continue

            # Just received a tap event from evtest
            taps_detected += 1

            t_tap_epoch, direction = tap_info
            shock_time_us = walt.read_shock_time()
            dt_tap_us = 1e6 * (t_tap_epoch - t_zero) - shock_time_us

            print(ev_line.strip())
            print("shock t %d, tap t %f, tap val %d. dt=%0.1f" % (shock_time_us, t_tap_epoch, direction, dt_tap_us))

            if shock_time_us == 0:
                print "No shock detected, skipping this event"
                continue

            taps.append((dt_tap_us, direction))

    evtest.terminate()

    # Process data
    print("\nProcessing data...")
    dt_down = numpy.array([t[0] for t in taps if t[1] == 1]) / 1e3
    dt_up = numpy.array([t[0] for t in taps if t[1] == 0]) / 1e3

    print('dt_down = ' + array2str(dt_down))
    print('dt_up = ' + array2str(dt_up))

    median_down_ms = numpy.median(dt_down)
    meidan_up_ms = numpy.median(dt_up)

    print('Median latency, down: %0.1f, up: %0.1f' % (median_down_ms, meidan_up_ms))


def run_walt_sanity_test(args):
    print('Starting sanity test')

    with Walt(args.serial) as walt:
        walt.sndrcv(Walt.CMD_RESET)

{'fraction_non_alphanumeric': 0.10605741136781228, 'fraction_numerical': 0.01265098346335733, 'mean_word_length': 2.1475161167993932, 'pattern_counts': {'":': 0, '<': 17, '<?xml version=': 0, '>': 19, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3774665', 'n_tokens_mistral': 11097, 'n_tokens_neox': 10497, 'n_words': 2920}