# -*- coding: utf-8; tab-width: 4; python-indent: 4 -*-
# 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.


# DESCRIPTION :
#
# This is a factory test for audio quality. External equipment will send
# command through ethernet to configure the audio loop path. Note that it's
# External equipment's responsibility to capture and analyze the audio signal.

import binascii
import logging
import os
import re
import select
import socket
import subprocess
import threading
import time

from autotest_lib.client.bin import test
from autotest_lib.client.common_lib import error, utils
from autotest_lib.client.cros import factory_setup_modules #pylint:disable=W0611
from autotest_lib.client.cros.audio import audio_helper
try:
    # Workaround to avoid not finding jsonrpclib in buildbot.
    from cros.factory.goofy.goofy import CACHES_DIR
except:
    pass
from cros.factory.event_log import Log
from cros.factory.test import factory
from cros.factory.test import test_ui
from cros.factory.test import shopfloor
from cros.factory.utils import net_utils
from cros.factory.test import utils as factory_utils
from cros.factory.test.factory import FactoryTestFailure

SHOPFLOOR_TIMEOUT_SECS = 10 # Timeout for shopfloor connection.
SHOPFLOOR_RETRY_INTERVAL_SECS = 10 # Seconds to wait between retries.
INSERT_ETHERNET_DONGLE_TIMEOUT_SECS = 30 # Timeout for inserting dongle.
IP_SETUP_TIMEOUT_SECS = 10 # Timeout for setting IP address.
CHECK_FIXTURE_COMPLETE_SECS = 1 # Seconds to check fixture test.

# Host test machine crossover connected to DUT, fix local ip and port for
# communication in between.
_HOST = ''
_PORT = 8888
_LOCAL_IP = '192.168.1.2'

# Setting
_INIT_COUNTDOWN = 3
_REMOVE_ETHERNET_TIMEOUT_SECS = 30 # Timeout for inserting dongle.
_FIXTURE_PARAMETERS = ['audio/audio_md5', 'audio/audio.zip']

# Label strings.
_LABEL_SPACE_TO_START = test_ui.MakeLabel('Press \'Space\' to start test',
        u'按空白键开始测试')
_LABEL_CONNECTED = test_ui.MakeLabel('Connected', u'已连线')
_LABEL_WAITING = test_ui.MakeLabel('Waiting for command', u'等待指令中')
_LABEL_AUDIOLOOP = test_ui.MakeLabel('Audio looping', u'音源回放中')
_LABEL_SPEAKER_MUTE_OFF = test_ui.MakeLabel('Speaker on', u'喇叭开启')
_LABEL_DMIC_ON = test_ui.MakeLabel('Dmic on', u'LCD mic开启')
_LABEL_PLAYTONE_LEFT = test_ui.MakeLabel('Playing tone to left channel',
        u'播音至左声道')
_LABEL_PLAYTONE_RIGHT = test_ui.MakeLabel('Playing tone to right channel',
        u'播音至右声道')
_LABEL_WAITING_ETHERNET = test_ui.MakeLabel(
        'Waiting for Ethernet connectivity to ShopFloor',
        u'等待网路介面卡连接到 ShopFloor')
_LABEL_WAITING_IP = test_ui.MakeLabel('Waiting for IP address',
        u'等待 IP 设定')
_LABEL_CONNECT_SHOPFLOOR = test_ui.MakeLabel('Connecting to ShopFloor...',
        u'连接到 ShopFloor 中...')
_LABEL_DOWNLOADING_PARAMETERS = test_ui.MakeLabel(
        'Downloading parameters', u'下载测试规格中')
_LABEL_REMOVE_ETHERNET = test_ui.MakeLabel(
        'Remove Ethernet connectivity', u'移除网路介面卡')
_LABEL_WAITING_FIXTURE_ETHERNET = test_ui.MakeLabel(
        'Waiting for Ethernet connectivity to audio fixture',
        u'等待网路介面卡连接到 audio 置具')
_LABEL_READY = test_ui.MakeLabel('Ready for connection', u'準备完成,等待链接')
_LABEL_UPLOAD_AUXLOG = test_ui.MakeLabel('Upload log', u'上传记录档')

# Regular expression to match external commands.
_LOOP_0_RE = re.compile("(?i)loop_0")
_LOOP_1_RE = re.compile("(?i)loop_1")
_LOOP_2_RE = re.compile("(?i)loop_2")
_LOOP_3_RE = re.compile("(?i)loop_3")
_XTALK_L_RE = re.compile("(?i)xtalk_l")
_XTALK_R_RE = re.compile("(?i)xtalk_r")
_MULTITONE_RE = re.compile("(?i)multitone")
_SEND_FILE_RE = re.compile("(?i)send_file")
_SWEEP_RE = re.compile("(?i)sweep")
_TEST_COMPLETE_RE = re.compile("(?i)test_complete")
_RESULT_PASS_RE = re.compile("(?i)result_pass")
_RESULT_FAIL_RE = re.compile("(?i)result_fail")
_VERSION_RE = re.compile("(?i)version")
_CONFIG_FILE_RE = re.compile("(?i)config_file")

# Common mixer settings for special audio configurations, board specific
# settings should goes to each test list.
_DMIC_SWITCH_MIXER_SETTINGS = []
_INIT_MIXER_SETTINGS = [{'name': '"HP/Speaker Playback Switch"',
                         'value': 'on'},
                        {'name': '"Master Playback Volume"',
                         'value': '90,90'}]
_UNMUTE_SPEAKER_MIXER_SETTINGS = [{'name': '"HP/Speaker Playback Switch"',
                                   'value': 'off'}]
_MUTE_LEFT_MIXER_SETTINGS = [{'name': '"Master Playback Switch"',
                              'value': 'off,on'}]
_MUTE_RIGHT_MIXER_SETTINGS = [{'name': '"Master Playback Switch"',
                               'value': 'on,off'}]

# Logs
_LABEL_FAIL_LOGS = 'Test fail, find more detail in log.'

class factory_AudioQuality(test.test):
    """This test case is to analysis audio quality including THD and R&B of
    digital microphone and speaker.

    There are two loops:
    1. from digital mic to headphone.
    2. from external mic to speaker.
    """
    version = 3
    preserve_srcdir = True

    def handle_connection(self, conn, *args):
        """Asynchronous handler for socket connection.

        Command Protocol:
          Command1[\x05]Data1[\x05]Data2[\x04][\x03]
        One line may contains many commands. and the last character must
        be Ascii code \x03.

        Use Ascii code \x05 to seperate command and data.
        Use Ascii code \x04 to present the end of command.
        Use Ascii code \x03 to present the end of list of command.

        When DUT received command, DUT should reply Active status immediately.
        Format is
          Command[\x05]Active[\x04][\x03]

        When DUT executed command, DUT should return result.
        Format is
          Command[\x05]Active_Status[\x05]Result[\x05]Result_String[\x05]
          Error_Code[\x04][\x03]
        Active_Status may be:
          Active_End: executed commmand successfully
          Active_Timeout: executed command timeout
        Result may be:
          Pass: result of command is pass
          Fail: result of command is fail
        Result_String and Error_Code could be any plaintext.
        If Result_String and Error_Code are empty, you can omit these.
        For Example: Command[\x05]Active_End[\x05]Pass[\x04][\x03]

        @param conn: socket connection
        """
        next_commands = ''
        while True:
            commands = next_commands
            while True:
                buf = conn.recv(1024)
                commands += buf
                if not buf or '\x03' in commands:
                    break

            commands, next_commands = commands.split('\x03', 1)
            if commands:
                logging.info("Received command %s", repr(commands))
            else:
                break

            command_list = commands[0:-1].split('\x04')
            for command in command_list:
                if not command:
                    continue
                attr_list = command.split('\x05')
                instruction = attr_list[0]
                conn.send(instruction + '\x05' + 'Active' + '\x04\x03')

                match_command = False
                for key in self._handlers.iterkeys():
                    if key.match(instruction):
                        match_command = True
                        factory.console.info('match command %s', instruction)
                        self._handlers[key](conn, attr_list)
                        break
                if not match_command:
                    factory.console.error("Command %s cannot find", instruction)
                    conn.send(instruction + '\x05' + 'Active_End' + '\x05' +
                        'Fail' + '\x04\x03')

            if self._test_complete:
                factory.console.info('Test completed')
                break
        factory.console.info("Connection disconnect")
        return False

    def restore_configuration(self):
        """Stops all the running process and restore the mute settings."""
        if self._multitone_job:
            utils.nuke_subprocess(self._multitone_job.sp)
            utils.join_bg_jobs([self._multitone_job], timeout=1)
            self._multitone_job = None

        if self._sweep_job:
            utils.nuke_subprocess(self._sweep_job.sp)
            utils.join_bg_jobs([self._sweep_job], timeout=1)
            self._sweep_job = None

        if self._tone_job:
            utils.nuke_subprocess(self._tone_job.sp)
            utils.join_bg_jobs([self._tone_job], timeout=1)
            self._tone_job = None

        if self._loop_process:
            self._loop_process.kill()
            self._loop_process = None
            logging.info("Stopped audio loop process")
        self._ah.set_mixer_controls(self._init_mixer_settings)

    def send_response(self, response, args):
        """Sends response to DUT for each command.

        @param response: response string
        @param args: This parameter is omitted when we test from FA-utility.
                     Otherwise, this parameter is passing from
                     handle_connection
                     args[0] is socket connection
                     args[1] is attr_list of handle_connection
        """
        # because there will not have args from mock_command
        if not args or not args[0] or not args[1][0]:
            return
        conn = args[0]
        command = args[1][0]
        if response:
            conn.send(command + '\x05' + 'Active_End' + '\x05' +
                'Pass' + '\x05' + response + '\x04\x03')
        else:
            conn.send(command + '\x05' + 'Active_End' + '\x05' +
                'Pass' + '\x04\x03')
        logging.info('Respond %s OK', command)

    def handle_version(self, *args):
        """Returns the md5 checksum of configuration file."""
        file_path = os.path.join(self._caches_dir, self._parameters[0])
        try:
            with open(file_path, "rb") as md5_file:
                rawstring = md5_file.read()
                self.send_response(rawstring.strip(), args)
        except IOError:
          factory.console.error('No such file or directory: %s', file_path)
        return

    def handle_config_file(self, *args):
        """Return the content of configuration file."""
        file_path = os.path.join(self._caches_dir, self._parameters[1])
        try:
            with open(file_path, "rb") as config_file:
                rawstring = config_file.read()
                """
                The format of file content is
                'file_name;file_size;file_content'.
                The file size is real file size instead of the size
                after b2a_hex.
                Using b2a_hex is to avoid the file content including special
                character such as '\x03', '\x04', and '\x05'.
                """
                rawdata = (self._parameters[1] + ';' +
                           str(len(rawstring)) + ';' +
                           binascii.b2a_hex(rawstring))

                self.send_response(rawdata, args)
        except IOError:
          factory.console.error('No such file or directory: %s', file_path)
        return

    def handle_send_file(self, *args):
        """This function is used to save test results from DUT.

        Also uploads the parsed data to log.
        """
        conn = args[0]

        attr_list = args[1]
        file_name = attr_list[1]
        size = int(attr_list[2])
        received_data = attr_list[3].replace('\x00', ' ')

        write_path = os.path.join(factory.get_log_root(), 'aux',
                'audio', file_name)
        factory_utils.TryMakeDirs(os.path.dirname(write_path))
        factory.console.info('save file: %s', write_path)
        with open(write_path, 'wb') as f:
          f.write(received_data)
        self._auxlogs.append(write_path)

        logging.info("Received file %s with size %d" , file_name, size)

        # Dump another copy of logs
        logging.info(repr(received_data))

        """
        The result logs are stored in filename ending in _[0-9]+.txt.

        Its content looks like:
        Freq [Hz]   dBV         Phase [Deg]
        100.00      -60.01      3.00
        105.93      -64.04      33.85
        112.20      -68.47      92.10
        ...

        The column count is variable. There may be up to ten columns in the
        results. Each column contains 12 characters. Because the spaces on
        the right side of last colume are stripped. So the number of column
        is the length of line divides by 12 and plus one.

        Unfortunately, we cannot read the column names in the header row
        by splitting with spaces.
        """

        match = re.search(r"(\d+)_(\d+)_(\d+).txt", file_name)
        match2 = re.search(r"(\d+)_(\d+).txt", file_name)

        if match:
            """
            serial_number and timestamp are generated by camerea test fixture.
            We can use these two strings to lookup the raw logs on fixture.
            """
            serial_number, timestamp, test_index = match.groups()

            lines = received_data.splitlines()
            header_row = lines[0]

            table = []
            """
            record the maximum column_number, to add sufficient 'nan' to
            the end of list if the spaces in the end of line are stripped.
            """
            column_number = max([len(line)/12 + 1 for line in lines[1:]])
            for line in lines[1:]:
                x = []
                for i in range(column_number):
                    x.append(float(line[i*12:i*12 + 12].strip() or 'nan'))
                table.append(x)

            test_result = {}
            """
            Remarks:
            1. cros.factory.event_log requires special format for key string
            2. because the harmonic of some frequencies are not valid, we may
               have empty values in certain fields
            3. The missing fields are always in the last columns
            """
            frequencies = dict((row[0], row[1:]) for row in table)
            test_result['frequencies'] = frequencies
            test_result['header_row'] = header_row
            test_result['serial_number'] = serial_number
            test_result['timestamp'] = timestamp

            Log(('audio_quality_test_%s' % test_index), **test_result)
        elif match2:
            serial_number, timestamp = match2.groups()

            final_result = {}
            final_result['serial_number'] = serial_number
            final_result['timestamp'] = timestamp
            final_result['data'] = received_data.replace('\r', '')

            Log('audio_quality_final_result', **final_result)
        else:
            logging.info("Unrecognizable filename %s", file_name)

        self.send_response(None, args)

    def handle_result_pass(self, *args):
        """Mark pass of this test case."""
        self._test_passed = True
        self.send_response(None, args)

    def handle_result_fail(self, *args):
        """Mark fail of this test case."""
        self._test_passed = False
        self.send_response(None, args)

    def handle_test_complete(self, *args):
        """Handles test completion.

        Runs post test script before ends this test
        """
        self.on_test_complete()
        self._test_complete = True
        logging.info('%s run_once finished', self.__class__)
        self.send_response(None, args)

    def handle_loop_none(self, *args):
        """Restore amixer configuration to default."""
        self.restore_configuration()
        self.ui.CallJSFunction('setMessage', _LABEL_WAITING)
        self.send_response(None, args)

    def handle_loop(self, *args):
        """Starts the internal audio loopback."""
        self.restore_configuration()
        self.ui.CallJSFunction('setMessage', _LABEL_AUDIOLOOP)
        if self._use_sox_loop:
            cmdargs = [self._ah.sox_path, '-t', 'alsa', self._input_dev, '-t',
                    'alsa', self._output_dev]
            self._loop_process = subprocess.Popen(cmdargs)
        else:
            cmdargs = [self._ah.audioloop_path, '-i', self._input_dev, '-o',
                    self._output_dev, '-c', str(self._loop_buffer_count)]
            self._loop_process = subprocess.Popen(cmdargs)

    def handle_multitone(self, *args):
        """Plays the multi-tone wav file."""
        self.restore_configuration()
        wav_path = os.path.join(self.srcdir, '10SEC.wav')
        cmdargs = ['aplay', wav_path]
        self._multitone_job = utils.BgJob(' '.join(cmdargs))
        self.send_response(None, args)

    def handle_sweep(self, *args):
        """Plays the sweep wav file."""
        self.restore_configuration()
        wav_path = os.path.join(self.srcdir, 'sweep.wav')
        cmdargs = ['aplay', wav_path]
        self._sweep_job = utils.BgJob(' '.join(cmdargs))
        self.send_response(None, args)

    def handle_loop_jack(self, *args):
        """External mic loop to headphone."""
        if self._use_multitone:
            self.handle_multitone()
        else:
            self.handle_loop()
        self.ui.CallJSFunction('setMessage', _LABEL_AUDIOLOOP)
        self.send_response(None, args)

    def handle_loop_from_dmic(self, *args):
        """Digital mic loop to headphone."""
        self.handle_loop()
        self.ui.CallJSFunction('setMessage', _LABEL_AUDIOLOOP +
                _LABEL_DMIC_ON)
        self._ah.set_mixer_controls(self._dmic_switch_mixer_settings)
        self.send_response(None, args)

    def handle_loop_speaker_unmute(self, *args):
        """External mic loop to speaker."""
        if self._use_multitone:
            self.handle_multitone()
        else:
            self.handle_loop()
        self.ui.CallJSFunction('setMessage', _LABEL_AUDIOLOOP +
                _LABEL_SPEAKER_MUTE_OFF)
        self._ah.set_mixer_controls(self._unmute_speaker_mixer_settings)
        self.send_response(None, args)

    def handle_xtalk_left(self, *args):
        """Cross talk left."""
        self.restore_configuration()
        self.ui.CallJSFunction('setMessage', _LABEL_PLAYTONE_LEFT)
        self._ah.set_mixer_controls(self._mute_left_mixer_settings)
        cmdargs = self._ah.get_play_sine_args(1, self._output_dev)
        self._tone_job = utils.BgJob(' '.join(cmdargs))
        self.send_response(None, args)

    def handle_xtalk_right(self, *args):
        """Cross talk right."""
        self.restore_configuration()
        self.ui.CallJSFunction('setMessage', _LABEL_PLAYTONE_RIGHT)
        self._ah.set_mixer_controls(self._mute_right_mixer_settings)
        cmdargs = self._ah.get_play_sine_args(0, self._output_dev)
        self._tone_job = utils.BgJob(' '.join(cmdargs))
        self.send_response(None, args)

    def listen_forever(self, sock):
        """Thread function to handle socket.

        @param sock: socket object.
        """
        fd = sock.fileno()
        while True:
            _rl, _, _ = select.select([fd], [], [])
            if fd in _rl:
                conn, addr = sock.accept()
                self.handle_connection(conn)

    def on_test_complete(self):
        """Restores the original state before exiting the test."""
        utils.system('iptables -D INPUT -p tcp --dport %s -j ACCEPT' % _PORT)
        utils.system('ifconfig %s down' % self._eth)
        utils.system('ifconfig %s up' % self._eth)
        self.restore_configuration()
        self._ah.cleanup_deps(['sox', 'audioloop'])

    def mock_command(self, event):
        """Receive test command from FA-utility.

        @param event: event from UI.
        """
        logging.info('Get event %s', event)
        cmd = event.data.get('cmd', '')
        for key in self._handlers.iterkeys():
            if key.match(cmd):
                self._handlers[key]()
                break

    def remove_network(self):
        """Detect and wait ethernet remove."""
        while True:
            try:
                self.ui.CallJSFunction('setMessage', _LABEL_REMOVE_ETHERNET)
                logging.info('Removing Ethernet device...')
                net_utils.PollForCondition(condition=(
                    lambda: False if net_utils.FindUsableEthDevice() else True),
                    timeout=_REMOVE_ETHERNET_TIMEOUT_SECS,
                    condition_name='Remove Ethernet device')
                break
            except:  # pylint: disable=W0702
                exception_string = factory_utils.FormatExceptionOnly()
                factory.console.error('Remove Ethernet Exception: %s',
                                      exception_string)

    def prepare_network(self, force_ip, msg):
        """Blocks forever until network is prepared.

        @param force_ip: If true, set _LOCAL_IP. Otherwise, use DHCP
        @param msg: The message will be shown in UI
        """
        def ObtainIp():
            """ Setup IP address """
            if force_ip is False:
                net_utils.SendDhcpRequest()
            else:
                net_utils.SetEthernetIp(_LOCAL_IP, force=True)
            return True if net_utils.GetEthernetIp() else False

        while True:
            self.ui.CallJSFunction('setMessage', msg)
            factory.console.info('Detecting Ethernet device...')
            try:
                net_utils.PollForCondition(condition=(
                        lambda: True if net_utils.FindUsableEthDevice()
                        else False),
                        timeout=INSERT_ETHERNET_DONGLE_TIMEOUT_SECS,
                        condition_name='Detect Ethernet device')

                # Only setup the IP if required so.
                current_ip = net_utils.GetEthernetIp(
                        net_utils.FindUsableEthDevice())
                if not current_ip or force_ip:
                    self.ui.CallJSFunction('setMessage', _LABEL_WAITING_IP)
                    factory.console.info('Setting up IP address...')
                    net_utils.PollForCondition(condition=ObtainIp,
                            timeout=IP_SETUP_TIMEOUT_SECS,
                            condition_name='Setup IP address')
                    break
                else:
                    break
            except:  # pylint: disable=W0702
                exception_string = factory_utils.FormatExceptionOnly()
                factory.console.error('Unable to setup network: %s',
                                      exception_string)
        factory.console.info('Network prepared. IP: %r',
                net_utils.GetEthernetIp())
        self._eth = net_utils.FindUsableEthDevice()

    def get_shopfloor_connection(
            self, timeout_secs=SHOPFLOOR_TIMEOUT_SECS,
            retry_interval_secs=SHOPFLOOR_RETRY_INTERVAL_SECS):
        """Returns a shopfloor client object.

        Try forever until a connection of shopfloor is established.

        @param timeout_secs: Timeout for shopfloor connection.
        @param retry_interval_secs: Seconds to wait between retries.
        """
        factory.console.info('Connecting to shopfloor...')
        while True:
            try:
                shopfloor_client = shopfloor.get_instance(
                    detect=True, timeout=timeout_secs)
                break
            except:  # pylint: disable=W0702
                exception_string = factory_utils.FormatExceptionOnly()
                logging.info('Unable to sync with shopfloor server: %s',
                             exception_string)
            time.sleep(retry_interval_secs)
        return shopfloor_client

    def init_audio_parameter(self):
        """Downloads parameters from shopfloor and saved to state/caches.

        The parameters include a ZIP file and a md5 checksum file.
        ZIP file is including all the files which are needed by Audio
        analysis software.
        md5 checksum file is used to check ZIP file version.
        If the version is mismatch, analysis software can download
        latest parameter and apply it.
        """
        self.prepare_network(False, _LABEL_WAITING_ETHERNET)
        factory.console.info('Start downloading parameters...')
        self.ui.CallJSFunction('setMessage', _LABEL_CONNECT_SHOPFLOOR)
        shopfloor_client = self.get_shopfloor_connection()
        logging.info('Syncing time with shopfloor...')
        goofy = factory.get_state_instance()
        goofy.SyncTimeWithShopfloorServer()

        self.ui.CallJSFunction('setMessage', _LABEL_DOWNLOADING_PARAMETERS)
        download_list = []
        for glob_expression in self._parameters:
            logging.info('Listing %s', glob_expression)
            download_list.extend(
                    shopfloor_client.ListParameters(glob_expression))
        factory.console.info('Download list prepared:\n%s',
                '\n'.join(download_list))
        assert len(download_list) > 0, 'No parameters found on shopfloor'
        """Download the list and saved to caches in state directory."""
        for filepath in download_list:
            utils.system('mkdir -p ' + os.path.join(
                    self._caches_dir, os.path.dirname(filepath)))
            binary_obj = shopfloor_client.GetParameter(filepath)
            with open(os.path.join(self._caches_dir, filepath), 'wb') as fd:
                fd.write(binary_obj.data)
        self.remove_network()

    def run_audio_server(self):
        """Initializes server and starts listening for external commands."""
        self.prepare_network(True, _LABEL_WAITING_FIXTURE_ETHERNET)
        sock = socket.socket()
        sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        sock.bind((_HOST, _PORT))
        sock.listen(1)
        logging.info("Listening at port %d", _PORT)

        self._listen_thread = threading.Thread(target=self.listen_forever,
                args=(sock,))
        self._listen_thread.start()
        self.ui.CallJSFunction('setMessage', _LABEL_READY)

        while True:
            if self._test_complete:
                break
            time.sleep(CHECK_FIXTURE_COMPLETE_SECS)
        self.remove_network()

    def upload_auxlog(self):
        """Uploads files which are sent from DUT by send_file command to
        shopfloor.
        """
        self.prepare_network(False, _LABEL_WAITING_ETHERNET)
        factory.console.info('Start uploading logs...')
        self.ui.CallJSFunction('setMessage', _LABEL_UPLOAD_AUXLOG)
        shopfloor.UploadAuxLogs(self._auxlogs)

    def start_run(self, event):
        """Runs the testing flow after user press 'space'.

        @param event: event from UI.
        """
        if self._use_shopfloor:
            self.init_audio_parameter()

        self.run_audio_server()

        if self._use_shopfloor:
            self.upload_auxlog()

        if self._test_passed:
            self.ui.Pass()
        else:
            if self._use_shopfloor:
                factory.console.info(
                        'Test failed. Force to flush event logs...')
                goofy = factory.get_state_instance()
                goofy.FlushEventLogs()
            self.ui.Fail(_LABEL_FAIL_LOGS)

    def run_once(self, input_dev='hw:0,0', output_dev='hw:0,0', eth='eth0',
            dmic_switch_mixer_settings=_DMIC_SWITCH_MIXER_SETTINGS,
            init_mixer_settings=_INIT_MIXER_SETTINGS,
            unmute_speaker_mixer_settings=_UNMUTE_SPEAKER_MIXER_SETTINGS,
            mute_right_mixer_settings=_MUTE_RIGHT_MIXER_SETTINGS,
            mute_left_mixer_settings=_MUTE_LEFT_MIXER_SETTINGS,
            use_sox_loop=False, use_multitone=False, loop_buffer_count=10,
            parameters=_FIXTURE_PARAMETERS, use_shopfloor=True):
        logging.info('%s run_once', self.__class__)

        self._ah = audio_helper.AudioHelper(self)
        self._ah.setup_deps(['sox', 'audioloop'])
        self._input_dev = input_dev
        self._output_dev = output_dev
        self._eth = None
        self._test_complete = False
        self._test_passed = False
        self._use_sox_loop = use_sox_loop
        self._use_multitone = use_multitone
        self._loop_buffer_count = loop_buffer_count

        self._multitone_job = None
        self._sweep_job = None
        self._tone_job = None
        self._loop_process = None
        self._parameters = parameters
        self._caches_dir = os.path.join(CACHES_DIR, 'parameters')
        self._use_shopfloor = use_shopfloor
        self._auxlogs = []

        # Mixer settings for different configurations.
        self._init_mixer_settings = init_mixer_settings
        self._unmute_speaker_mixer_settings = unmute_speaker_mixer_settings
        self._mute_left_mixer_settings = mute_left_mixer_settings
        self._mute_right_mixer_settings = mute_right_mixer_settings
        self._dmic_switch_mixer_settings = dmic_switch_mixer_settings

        self.ui = test_ui.UI()

        # Register commands to corresponding handlers.
        self._handlers = {}
        self._handlers[_SEND_FILE_RE] = self.handle_send_file
        self._handlers[_RESULT_PASS_RE] = self.handle_result_pass
        self._handlers[_RESULT_FAIL_RE] = self.handle_result_fail
        self._handlers[_TEST_COMPLETE_RE] = self.handle_test_complete
        self._handlers[_LOOP_0_RE] = self.handle_loop_none
        self._handlers[_LOOP_1_RE] = self.handle_loop_from_dmic
        self._handlers[_LOOP_2_RE] = self.handle_loop_speaker_unmute
        self._handlers[_LOOP_3_RE] = self.handle_loop_jack
        self._handlers[_XTALK_L_RE] = self.handle_xtalk_left
        self._handlers[_XTALK_R_RE] = self.handle_xtalk_right
        self._handlers[_MULTITONE_RE] = self.handle_multitone
        self._handlers[_SWEEP_RE] = self.handle_sweep
        self._handlers[_VERSION_RE] = self.handle_version
        self._handlers[_CONFIG_FILE_RE] = self.handle_config_file

        self.ui.CallJSFunction('setMessage', _LABEL_SPACE_TO_START)
        self.ui.AddEventHandler('start_run', self.start_run)
        self.ui.AddEventHandler('mock_command', self.mock_command)
        utils.system('iptables -A INPUT -p tcp --dport %s -j ACCEPT' % _PORT)
        try:
            self.ui.Run()
        except FactoryTestFailure as e:
            raise error.TestError(e.message)