xref: /external/autotest/client/site_tests/firmware_TouchMTB/test_flow.py

# -*- coding: utf-8 -*-

# Copyright (c) 2012 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.

"""Guide the user to perform gestures. Record and validate the gestures."""

import fcntl
import glob
import os
import subprocess
import sys

import common_util
import firmware_log
import firmware_utils
import fuzzy
import mini_color
import mtb
import touchbotII_robot_wrapper as robot_wrapper
import test_conf as conf
import validators

from firmware_utils import GestureList

sys.path.append('../../bin/input')
import input_device

# Include some constants
from firmware_constants import DEV, GV, MODE, OPTIONS, TFK


class TestFlow:
    """Guide the user to perform gestures. Record and validate the gestures."""

    def __init__(self, device_geometry, device, keyboard, win, parser, output,
                 test_version, board, firmware_version, options):
        self.device_geometry = device_geometry
        self.device = device
        self.device_node = self.device.device_node
        self.keyboard = keyboard
        self.firmware_version = firmware_version
        self.output = output
        self.board = board
        self.test_version = test_version
        self.output.print_report('%s' % test_version)
        self._get_record_cmd()
        self.win = win
        self.parser = parser
        self.packets = None
        self.gesture_file_name = None
        self.prefix_space = self.output.get_prefix_space()
        self.scores = []
        self.mode = options[OPTIONS.MODE]
        self.fngenerator_only = options[OPTIONS.FNGENERATOR]
        self.iterations = options[OPTIONS.ITERATIONS]
        self.replay_dir = options[OPTIONS.REPLAY]
        self.resume_dir = options[OPTIONS.RESUME]
        self.recording = not any([bool(self.replay_dir), bool(self.resume_dir)])
        self.device_type = (DEV.TOUCHSCREEN if options[OPTIONS.TOUCHSCREEN]
                                            else DEV.TOUCHPAD)

        self.robot = robot_wrapper.RobotWrapper(self.board, options)
        self.robot_waiting = False

        self.gv_count = float('infinity')
        gesture_names = self._get_gesture_names()
        order = None
        if self._is_robot_mode():
            order = lambda x: conf.finger_tips_required[x.name]
        self.gesture_list = GestureList(gesture_names).get_gesture_list(order)
        self._get_all_gesture_variations(options[OPTIONS.SIMPLIFIED])

        self.init_flag = False
        self.system_device = self._non_blocking_open(self.device_node)
        self.evdev_device = input_device.InputEvent()
        self.screen_shot = firmware_utils.ScreenShot(self.geometry_str)
        self.mtb_evemu = mtb.MtbEvemu(device)

        self._rename_old_log_and_html_files()
        self._set_static_prompt_messages()
        self.gesture_image_name = None
        self.gesture_continues_flag = False
        self.use_existent_event_file_flag = False

    def __del__(self):
        self.system_device.close()

    def _rename_old_log_and_html_files(self):
        """When in replay or resume mode, rename the old log and html files."""
        if self.replay_dir or self.resume_dir:
            for file_type in ['*.log', '*.html']:
                path_names = os.path.join(self.output.log_dir, file_type)
                for old_path_name in glob.glob(path_names):
                    new_path_name = '.'.join([old_path_name, 'old'])
                    os.rename(old_path_name, new_path_name)

    def _is_robot_mode(self):
        return self.robot.is_robot_action_mode() or self.mode == MODE.ROBOT_SIM

    def _get_gesture_names(self):
        """Determine the gesture names based on the mode."""
        if self.mode == MODE.QUICKSTEP:
            return conf.gesture_names_quickstep
        elif self.mode == MODE.NOISE:
            return conf.gesture_names_noise_extended
        elif self._is_robot_mode():
            # The mode could be MODE.ROBOT or MODE.ROBOT_SIM.
            # The same gesture names list is used in both modes.
            return conf.gesture_names_robot[self.device_type]
        elif self.mode == MODE.MANUAL:
            # Define the manual list which is gesture_names_complete:
            # gesture_names_robot - gesture_names_equipment_required
            manual_set = (set(conf.gesture_names_complete[self.device_type]) -
                          set(conf.gesture_names_robot[self.device_type]))
            return list(manual_set - set(conf.gesture_names_fngenerator_required))

        elif self.mode == MODE.CALIBRATION:
            return conf.gesture_names_calibration
        else:
            # Filter out tests that need a function generator for COMPLETE mode
            # unless they've indicated that they have one
            return [n for n in conf.gesture_names_complete[self.device_type]
                    if (self.fngenerator_only or
                        n not in conf.gesture_names_fngenerator_required)]

    def _non_blocking_open(self, filename):
        """Open the file in non-blocing mode."""
        fd = open(filename)
        fcntl.fcntl(fd, fcntl.F_SETFL, os.O_NONBLOCK)
        return fd

    def _non_blocking_read(self, dev, fd):
        """Non-blocking read on fd."""
        try:
            dev.read(fd)
            event = (dev.tv_sec, dev.tv_usec, dev.type, dev.code, dev.value)
        except Exception, e:
            event = None
        return event

    def _reopen_system_device(self):
        """Close the device and open a new one."""
        self.system_device.close()
        self.system_device = open(self.device_node)
        self.system_device = self._non_blocking_open(self.device_node)

    def _set_static_prompt_messages(self):
        """Set static prompt messages."""
        # Prompt for next gesture.
        self._prompt_next = (
                "Press SPACE to save this file and go to next test,\n"
                "      'm'   to save this file and record again,\n"
                "      'd'   to delete this file and try again,\n"
                "      'x'   to discard this file and exit.")

        # Prompt to see test result through timeout callback.
        self._prompt_result = (
                "Perform the gesture now.\n"
                "See the test result on the right after finger lifted.\n"
                "Or press 'x' to exit.")

    def _get_prompt_abnormal_gestures(self, warn_msg):
        """Prompt for next gesture."""
        prompt = '\n'.join(
                ["It is very likely that you perform a WRONG gesture!",
                 warn_msg,
                 "Press 'd'   to delete this file and try again (recommended),",
                 "      SPACE to save this file if you are sure it's correct,",
                 "      'x'   to discard this file and exit."])
        return prompt

    def _get_prompt_no_data(self):
        """Prompt to remind user of performing gestures."""
        prompt = ("You need to perform the specified gestures "
                  "before pressing SPACE.\n")
        return prompt + self._prompt_result

    def _get_record_cmd(self):
        """Get the device event record command."""
        # Run mtplot with settings to disable clearing the display if the robot
        # clicks the pad, and adding a visible click indicator in the output
        self.record_program = 'mtplot -s1 -c0 -m0'
        if not common_util.program_exists(self.record_program):
            msg = 'Error: the program "%s" does not exist in $PATH.'
            self.output.print_report(msg % self.record_program)
            exit(1)

        display_name = firmware_utils.get_display_name()
        self.geometry_str = '%dx%d+%d+%d' % self.device_geometry
        format_str = '%s %s -d %s -g %s'
        self.record_cmd = format_str % (self.record_program,
                                        self.device_node,
                                        display_name,
                                        self.geometry_str)
        self.output.print_report('Record program: %s' % self.record_cmd)

    def _span_seq(self, seq1, seq2):
        """Span sequence seq1 over sequence seq2.

        E.g., seq1 = (('a', 'b'), 'c')
              seq2 = ('1', ('2', '3'))
              res = (('a', 'b', '1'), ('a', 'b', '2', '3'),
                     ('c', '1'), ('c', '2', '3'))
        E.g., seq1 = ('a', 'b')
              seq2 = ('1', '2', '3')
              res  = (('a', '1'), ('a', '2'), ('a', '3'),
                      ('b', '1'), ('b', '2'), ('b', '3'))
        E.g., seq1 = (('a', 'b'), ('c', 'd'))
              seq2 = ('1', '2', '3')
              res  = (('a', 'b', '1'), ('a', 'b', '2'), ('a', 'b', '3'),
                      ('c', 'd', '1'), ('c', 'd', '2'), ('c', 'd', '3'))
        """
        to_list = lambda s: list(s) if isinstance(s, tuple) else [s]
        return tuple(tuple(to_list(s1) + to_list(s2)) for s1 in seq1
                                                      for s2 in seq2)

    def span_variations(self, seq):
        """Span the variations of a gesture."""
        if seq is None:
            return (None,)
        elif isinstance(seq[0], tuple):
            return reduce(self._span_seq, seq)
        else:
            return seq

    def _stop(self):
        """Terminate the recording process."""
        self.record_proc.poll()
        # Terminate the process only when it was not terminated yet.
        if self.record_proc.returncode is None:
            self.record_proc.terminate()
            self.record_proc.wait()
        self.output.print_window('')

    def _get_gesture_image_name(self):
        """Get the gesture file base name without file extension."""
        filepath = os.path.splitext(self.gesture_file_name)[0]
        self.gesture_image_name = filepath + '.png'
        return filepath

    def _close_gesture_file(self):
        """Close the gesture file."""
        if self.gesture_file.closed:
            return

        filename = self.gesture_file.name
        self.gesture_file.close()

        # Strip off the header of the gesture file.
        #
        # Input driver version is 1.0.1
        # Input device ID: bus 0x18 vendor 0x0 product 0x0 version 0x0
        # Input device name: "Atmel maXTouch Touchpad"
        # ...
        # Testing ... (interrupt to exit)
        # Event: time 519.855, type 3 (EV_ABS), code 57 (ABS_MT_TRACKING_ID),
        #                                       value 884
        #
        tmp_filename = filename + '.tmp'
        os.rename(filename, tmp_filename)
        with open(tmp_filename) as src_f:
            with open(filename, 'w') as dst_f:
                for line in src_f:
                    if line.startswith('Event:'):
                        dst_f.write(line)
        os.remove(tmp_filename)

    def _stop_record_and_post_image(self):
        """Terminate the recording process."""
        if self.record_new_file:
            self._close_gesture_file()
            self.screen_shot.dump_root(self._get_gesture_image_name())
            self.record_proc.terminate()
            self.record_proc.wait()
        else:
            self._get_gesture_image_name()
        self.win.set_image(self.gesture_image_name)

    def _create_prompt(self, test, variation):
        """Create a color prompt."""
        prompt = test.prompt
        if isinstance(variation, tuple):
            subprompt = reduce(lambda s1, s2: s1 + s2,
                               tuple(test.subprompt[s] for s in variation))
        elif variation is None or test.subprompt is None:
            subprompt = None
        else:
            subprompt = test.subprompt[variation]

        if subprompt is None:
            color_prompt = prompt
            monochrome_prompt = prompt
        else:
            color_prompt = mini_color.color_string(prompt, '{', '}', 'green')
            color_prompt = color_prompt.format(*subprompt)
            monochrome_prompt = prompt.format(*subprompt)

        color_msg_format = mini_color.color_string('\n<%s>:\n%s%s', '<', '>',
                                                   'blue')
        color_msg = color_msg_format % (test.name, self.prefix_space,
                                        color_prompt)
        msg = '%s: %s' % (test.name, monochrome_prompt)

        glog = firmware_log.GestureLog()
        glog.name = test.name
        glog.variation = variation
        glog.prompt = monochrome_prompt

        return (msg, color_msg, glog)

    def _choice_exit(self):
        """Procedure to exit."""
        self._stop()
        if os.path.exists(self.gesture_file_name):
            os.remove(self.gesture_file_name)
            self.output.print_report(self.deleted_msg)

    def _stop_record_and_rm_file(self):
        """Stop recording process and remove the current gesture file."""
        self._stop()
        if os.path.exists(self.gesture_file_name):
            os.remove(self.gesture_file_name)
            self.output.print_report(self.deleted_msg)

    def _create_gesture_file_name(self, gesture, variation):
        """Create the gesture file name based on its variation.

        Examples of different levels of file naming:
            Primary name:
                pinch_to_zoom.zoom_in-lumpy-fw_11.27
            Root name:
                pinch_to_zoom.zoom_in-lumpy-fw_11.27-manual-20130221_050510
            Base name:
                pinch_to_zoom.zoom_in-lumpy-fw_11.27-manual-20130221_050510.dat
        """
        if variation is None:
            gesture_name = gesture.name
        else:
            if type(variation) is tuple:
                name_list = [gesture.name,] + list(variation)
            else:
                name_list = [gesture.name, variation]
            gesture_name = '.'.join(name_list)

        self.primary_name = conf.filename.sep.join([
                gesture_name,
                self.board,
                conf.fw_prefix + self.firmware_version])
        root_name = conf.filename.sep.join([
                self.primary_name,
                self.mode,
                firmware_utils.get_current_time_str()])
        basename = '.'.join([root_name, conf.filename.ext])
        return basename

    def _add_scores(self, new_scores):
        """Add the new scores of a single gesture to the scores list."""
        if new_scores is not None:
            self.scores += new_scores

    def _final_scores(self, scores):
        """Print the final score."""
        # Note: conf.score_aggregator uses a function in fuzzy module.
        final_score = eval(conf.score_aggregator)(scores)
        self.output.print_report('\nFinal score: %s\n' % str(final_score))

    def _robot_action(self):
        """Control the robot to perform the action."""
        if self._is_robot_mode() or self.robot.is_manual_noise_test_mode():
            self.robot.configure_noise(self.gesture, self.variation)

        if self._is_robot_mode():
            self.robot.control(self.gesture, self.variation)
            # Once the script terminates start a timeout to clean up if one
            # hasn't already been set to keep the test suite from hanging.
            if not self.gesture_begins_flag:
                self.win.register_timeout_add(self.gesture_timeout_callback,
                                              self.gesture.timeout)

    def _handle_user_choice_save_after_parsing(self, next_gesture=True):
        """Handle user choice for saving the parsed gesture file."""
        self.output.print_window('')
        if self.use_existent_event_file_flag or self.recording:
            if self.saved_msg:
                self.output.print_report(self.saved_msg)
            if self.new_scores:
                self._add_scores(self.new_scores)
            self.output.report_html.insert_image(self.gesture_image_name)
            self.output.report_html.flush()
        # After flushing to report_html, reset the gesture_image_name so that
        # it will not be reused by next gesture variation accidentally.
        self.gesture_image_name = None

        if self._pre_setup_this_gesture_variation(next_gesture=next_gesture):
            # There are more gestures.
            self._setup_this_gesture_variation()
            self._robot_action()
        else:
            # No more gesture.
            self._final_scores(self.scores)
            self.output.stop()
            self.output.report_html.stop()
            self.win.stop()
        self.packets = None

    def _handle_user_choice_discard_after_parsing(self):
        """Handle user choice for discarding the parsed gesture file."""
        self.output.print_window('')
        self._setup_this_gesture_variation()
        self._robot_action()
        self.packets = None

    def _handle_user_choice_exit_after_parsing(self):
        """Handle user choice to exit after the gesture file is parsed."""
        self._stop_record_and_rm_file()
        self.output.stop()
        self.output.report_html.stop()
        self.win.stop()

    def check_for_wrong_number_of_fingers(self, details):
        flag_found = False
        try:
            position = details.index('CountTrackingIDValidator')
        except ValueError as e:
            return None

        # An example of the count of tracking IDs:
        #     '    count of trackid IDs: 1'
        number_tracking_ids = int(details[position + 1].split()[-1])
        # An example of the criteria_str looks like:
        #     '    criteria_str: == 2'
        criteria = int(details[position + 2].split()[-1])
        if number_tracking_ids < criteria:
            print '  CountTrackingIDValidator: '
            print '  number_tracking_ids: ', number_tracking_ids
            print '  criteria: ', criteria
            print '  number_tracking_ids should be larger!'
            msg = 'Number of Tracking IDs should be %d instead of %d'
            return msg % (criteria, number_tracking_ids)
        return None

    def _empty_packets_is_legal_result(self):
        return ('tap' in self.gesture.name and self._is_robot_mode())

    def _handle_user_choice_validate_before_parsing(self):
        """Handle user choice for validating before gesture file is parsed."""
        # Parse the device events. Make sure there are events.
        self.packets = self.parser.parse_file(self.gesture_file_name)
        if self.packets or self._empty_packets_is_legal_result():
            # Validate this gesture and get the results.
            (self.new_scores, msg_list, vlogs) = validators.validate(
                    self.packets, self.gesture, self.variation)

            # If the number of tracking IDs is less than the expected value,
            # the user probably made a wrong gesture.
            error = self.check_for_wrong_number_of_fingers(msg_list)
            if error:
                prompt = self._get_prompt_abnormal_gestures(error)
                color = 'red'
            else:
                prompt = self._prompt_next
                color = 'black'

            self.output.print_window(msg_list)
            self.output.buffer_report(msg_list)
            self.output.report_html.insert_validator_logs(vlogs)
            self.win.set_prompt(prompt, color=color)
            print prompt
            self._stop_record_and_post_image()
        else:
            self.win.set_prompt(self._get_prompt_no_data(), color='red')

    def _handle_user_choice_exit_before_parsing(self):
        """Handle user choice to exit before the gesture file is parsed."""
        self._close_gesture_file()
        self._handle_user_choice_exit_after_parsing()

    def _is_parsing_gesture_file_done(self):
        """Is parsing the gesture file done?"""
        return self.packets is not None

    def _is_arrow_key(self, choice):
        """Is this an arrow key?"""
        return (choice in TFK.ARROW_KEY_LIST)

    def user_choice_callback(self, fd, condition):
        """A callback to handle the key pressed by the user.

        This is the primary GUI event-driven method handling the user input.
        """
        choice = self.keyboard.get_key_press_event(fd)
        if choice:
            self._handle_keyboard_event(choice)
        return True

    def _handle_keyboard_event(self, choice):
        """Handle the keyboard event."""
        if self._is_arrow_key(choice):
            self.win.scroll(choice)
        elif self.robot_waiting:
            # The user wants the robot to start its action.
            if choice in (TFK.SAVE, TFK.SAVE2):
                self.robot_waiting = False
                self._robot_action()
            # The user wants to exit.
            elif choice == TFK.EXIT:
                self._handle_user_choice_exit_after_parsing()
        elif self._is_parsing_gesture_file_done():
            # Save this gesture file and go to next gesture.
            if choice in (TFK.SAVE, TFK.SAVE2):
                self._handle_user_choice_save_after_parsing()
            # Save this file and perform the same gesture again.
            elif choice == TFK.MORE:
                self._handle_user_choice_save_after_parsing(next_gesture=False)
            # Discard this file and perform the gesture again.
            elif choice == TFK.DISCARD:
                self._handle_user_choice_discard_after_parsing()
            # The user wants to exit.
            elif choice == TFK.EXIT:
                self._handle_user_choice_exit_after_parsing()
            # The user presses any wrong key.
            else:
                self.win.set_prompt(self._prompt_next, color='red')
        else:
            if choice == TFK.EXIT:
                self._handle_user_choice_exit_before_parsing()
            # The user presses any wrong key.
            else:
                self.win.set_prompt(self._prompt_result, color='red')

    def _get_all_gesture_variations(self, simplified):
        """Get all variations for all gestures."""
        gesture_variations_list = []
        self.variations_dict = {}
        for gesture in self.gesture_list:
            variations_list = []
            variations = self.span_variations(gesture.variations)
            for variation in variations:
                gesture_variations_list.append((gesture, variation))
                variations_list.append(variation)
                if simplified:
                    break
            self.variations_dict[gesture.name] = variations_list
        self.gesture_variations = iter(gesture_variations_list)

    def gesture_timeout_callback(self):
        """A callback watching whether a gesture has timed out."""
        if self.replay_dir:
            # There are event files to replay for this gesture variation.
            if self.use_existent_event_file_flag:
                self._handle_user_choice_validate_before_parsing()
            self._handle_user_choice_save_after_parsing(next_gesture=True)
            return False

        # A gesture is stopped only when two conditions are met simultaneously:
        # (1) there are no reported packets for a timeout interval, and
        # (2) the number of tracking IDs is 0.
        elif (self.gesture_continues_flag or
            not self.mtb_evemu.all_fingers_leaving()):
            self.gesture_continues_flag = False
            return True

        else:
            self._handle_user_choice_validate_before_parsing()
            self.win.remove_event_source(self.gesture_file_watch_tag)
            if self._is_robot_mode():
                self._handle_keyboard_event(TFK.SAVE)
            return False

    def gesture_file_watch_callback(self, fd, condition, evdev_device):
        """A callback to watch the device input."""
        # Read the device node continuously until end
        event = True
        while event:
            event = self._non_blocking_read(evdev_device, fd)
            if event:
                self.mtb_evemu.process_event(event)

        self.gesture_continues_flag = True
        if (not self.gesture_begins_flag):
            self.gesture_begins_flag = True
            self.win.register_timeout_add(self.gesture_timeout_callback,
                                          self.gesture.timeout)
        return True

    def init_gesture_setup_callback(self, widget, event):
        """A callback to set up environment before a user starts a gesture."""
        if not self.init_flag:
            self.init_flag = True
            self._pre_setup_this_gesture_variation()
            self._setup_this_gesture_variation()
            self._robot_action()

    def _get_existent_event_files(self):
        """Get the existent event files that starts with the primary_name."""
        primary_pathnames = os.path.join(self.output.log_dir,
                                         self.primary_name + '*.dat')
        self.primary_gesture_files = glob.glob(primary_pathnames)
        # Reverse sorting the file list so that we could pop from the tail.
        self.primary_gesture_files.sort()
        self.primary_gesture_files.reverse()

    def _use_existent_event_file(self):
        """If the replay flag is set in the command line, and there exists a
        file(s) with the same primary name, then use the existent file(s)
        instead of recording a new one.
        """
        if self.primary_gesture_files:
            self.gesture_file_name = self.primary_gesture_files.pop()
            return True
        return False

    def _pre_setup_this_gesture_variation(self, next_gesture=True):
        """Get gesture, variation, filename, prompt, etc."""
        next_gesture_first_time = False
        if next_gesture:
            if self.gv_count < self.iterations:
                self.gv_count += 1
            else:
                self.gv_count = 1
                gesture_variation = next(self.gesture_variations, None)
                if gesture_variation is None:
                    return False
                self.gesture, self.variation = gesture_variation
                next_gesture_first_time = True

        basename = self._create_gesture_file_name(self.gesture, self.variation)
        if next_gesture_first_time:
            self._get_existent_event_files()

        if self.replay_dir or self.resume_dir:
            self.use_existent_event_file_flag = self._use_existent_event_file()

        if ((not self.replay_dir and not self.resume_dir) or
                (self.resume_dir and not self.use_existent_event_file_flag)):
            self.gesture_file_name = os.path.join(self.output.log_dir, basename)
            self.saved_msg = '(saved: %s)\n' % self.gesture_file_name
            self.deleted_msg = '(deleted: %s)\n' % self.gesture_file_name
        else:
            self.saved_msg = None
            self.deleted_msg = None
        self.new_scores = None

        if self.robot.is_robot_action_mode() or self.robot.is_manual_noise_test_mode():
            self.robot.turn_off_noise()

        (msg, color_msg, glog) = self._create_prompt(self.gesture,
                                                     self.variation)
        self.win.set_gesture_name(msg)
        self.output.report_html.insert_gesture_log(glog)
        print color_msg
        self.output.print_report(color_msg)
        return True

    def _setup_this_gesture_variation(self):
        """Set up the recording process or use an existent event data file."""
        if self.replay_dir:
            self.record_new_file = False
            self.win.register_timeout_add(self.gesture_timeout_callback, 0)
            return

        if self.resume_dir and self.use_existent_event_file_flag:
            self.record_new_file = False
            self._handle_user_choice_validate_before_parsing()
            self._handle_keyboard_event(TFK.SAVE)
            return

        # Initiate the MtbSanityValidator. Note that this should be done each
        # time just before recording the gesture file since it requires a
        # snapshot of the input device before any finger touching the device.
        self.gesture.mtb_sanity_validator = validators.MtbSanityValidator()

        # Now, we will record a new gesture event file.
        # Fork a new process for mtplot. Add io watch for the gesture file.
        self.record_new_file = True
        self.gesture_file = open(self.gesture_file_name, 'w')
        self.record_proc = subprocess.Popen(self.record_cmd.split(),
                                            stdout=self.gesture_file)

        # Watch if data come in to the monitored file.
        self.gesture_begins_flag = False
        self._reopen_system_device()
        self.gesture_file_watch_tag = self.win.register_io_add_watch(
                self.gesture_file_watch_callback, self.system_device,
                self.evdev_device)