# -*- coding: utf-8 -*-
# Copyright (c) 2011 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 the audio function. An external loopback dongle
# is required to automatically capture and detect the playback tones.

import glib
import gtk
import os
import re
import subprocess
import tempfile
import utils

from autotest_lib.client.bin import test
from autotest_lib.client.common_lib import error
from autotest_lib.client.cros import factory
from autotest_lib.client.cros.audio import audio_helper
from autotest_lib.client.cros.factory import ui as ful

_DEFAULT_FREQ_HZ = 1000
_DEFAULT_FREQ_THRESHOLD_HZ = 50
_DEFAULT_DURATION_SEC = 1

_LABEL_START_STR = 'Hit s to start loopback test\n' +\
        '按s鍵開始音源回放測試\n\n'
_LABEL_IN_PROGRESS_STR = 'Loopback testing...\n' +\
        '音源回放測試中...\n\n'
_LABEL_SUCCESS_RATE = 'Success rate %f\n'

# Regular expressions to match audiofuntest message.
_AUDIOFUNTEST_STOP_RE = re.compile('^Stop')
_AUDIOFUNTEST_SUCCESS_RATE_RE = re.compile('.*rate\s=\s(.*)$')

class factory_AudioLoop(test.test):
    version = 1

    def key_press_callback(self, widget, event):
        if event.keyval == ord('s'):
            self._status_box.remove(self._status_label)
            self._status_label = None
            self._status_label = ful.make_label(_LABEL_IN_PROGRESS_STR,
                                                fg = ful.WHITE)
            self._status_box.add(self._status_label)
        if self._audiofuntest:
            self._success_rate_box.remove(self._success_rate_label)
            self._success_rate_label = None
            self._success_rate_label = ful.make_label("", fg = ful.WHITE)
            self._success_rate_box.add(self._success_rate_label)

        self._main_widget.show_all()
        self._main_widget.queue_draw()
        return True

    def key_release_callback(self, widget, event):
        if event.keyval == ord('s'):
            if self._audiofuntest:
                self._proc = subprocess.Popen(self._audiofuntest_path,
                        stderr=subprocess.PIPE)
                self._gio_tag = glib.io_add_watch(self._proc.stderr, glib.IO_IN,
                        self.audiofuntest_cb, priority=glib.PRIORITY_LOW)
            else:
                self.audio_loopback()
                gtk.main_quit()

        return True

    def audio_loopback(self):
        for input_device in self._input_devices:
            self._ah = audio_helper.AudioHelper(self,
                    input_device=input_device,
                    record_duration=self._duration)
            # TODO(hychao): split deps and I/O devices to different
            # utils so we can setup deps only once.
            self._ah.setup_deps(['sox'])
            for output_device in self._output_devices:
                # Record a sample of "silence" to use as a noise profile.
                with tempfile.NamedTemporaryFile(mode='w+t') as noise_file:
                    factory.log('Noise file: %s' % noise_file.name)
                    self._ah.record_sample(noise_file.name)

                    # Playback sine tone and check the recorded audio frequency.
                    self._ah.loopback_test_channels(noise_file,
                            lambda ch: self.playback_sine(ch, output_device),
                            self.check_recorded_audio)

    def playback_sine(self, unused_channel, output_device='default'):
        cmd = '%s -n -t alsa %s synth %d sine %d' % (self._ah.sox_path,
                output_device, self._duration, self._freq)
        utils.system(cmd)

    def check_recorded_audio(self, sox_output):
        freq = self._ah.get_rough_freq(sox_output)
        if abs(freq - self._freq) > _DEFAULT_FREQ_THRESHOLD_HZ:
            raise error.TestError('Frequency %d not match' % freq)
        self._result = True
        factory.log('Got frequency %d' % freq)

    def audiofuntest_cb(self, source, cb_condition):
        '''
        Sample audiofuntest message:

        O: carrier = 41, delay =  6, success =  60, fail =   0, rate = 100.0
        Stop play tone
        Stop capturing data
        '''
        line = source.readline()

        m = _AUDIOFUNTEST_SUCCESS_RATE_RE.match(line)
        if m is not None:
            self._last_success_rate = float(m.group(1))

            self._success_rate_box.remove(self._success_rate_label)
            self._success_rate_label = None
            self._success_rate_label = ful.make_label(_LABEL_SUCCESS_RATE %
                    self._last_success_rate, fg = ful.WHITE)
            self._success_rate_box.add(self._success_rate_label)
            self._main_widget.show_all()
            self._main_widget.queue_draw()

        m = _AUDIOFUNTEST_STOP_RE.match(line)
        if m is not None:
            glib.source_remove(self._gio_tag)
            if (hasattr(self, '_last_success_rate') and
                    self._last_success_rate is not None):
                self._result = self._last_success_rate > 50.0
                gtk.main_quit()

        return True

    def register_callbacks(self, window):
        window.connect('key-press-event', self.key_press_callback)
        window.add_events(gtk.gdk.KEY_PRESS_MASK)
        window.connect('key-release-event', self.key_release_callback)
        window.add_events(gtk.gdk.KEY_RELEASE_MASK)

    def run_once(self, audiofuntest=False, duration=_DEFAULT_DURATION_SEC,
            input_devices=['default'], output_devices=['default'],
            mixer_controls=None):
        factory.log('%s run_once' % self.__class__)

        self._audiofuntest = audiofuntest
        self._duration = duration
        self._freq = _DEFAULT_FREQ_HZ
        self._input_devices = input_devices
        self._output_devices = output_devices

        # Create a default audio helper to do the setup jobs.
        self._ah = audio_helper.AudioHelper(self, record_duration=duration)
        if mixer_controls is not None:
            self._ah.set_mixer_controls(mixer_controls)

        # Setup dependencies
        self._ah.setup_deps(['sox', 'test_tones'])
        self._audiofuntest_path = os.path.join(self.autodir, 'deps',
                'test_tones', 'src', 'audiofuntest')
        if not (os.path.exists(self._audiofuntest_path) and
                os.access(self._audiofuntest_path, os.X_OK)):
            raise error.TestError(
                    '%s is not an executable' % self._audiofuntest_path)

        # Status box and label
        self._status_box = gtk.EventBox()
        self._status_box.modify_bg(gtk.STATE_NORMAL, ful.BLACK)
        self._status_label = ful.make_label(_LABEL_START_STR,
                                            fg = ful.WHITE)
        self._status_box.add(self._status_label)

        # Success rate box and label
        self._success_rate_box = gtk.EventBox()
        self._success_rate_box.modify_bg(gtk.STATE_NORMAL, ful.BLACK)
        self._success_rate_label = ful.make_label("", fg = ful.WHITE)
        self._success_rate_box.add(self._success_rate_label)

        self._label_list = gtk.VBox()
        self._label_list.pack_start(self._status_box, False, False)
        self._label_list.pack_end(self._success_rate_box, False, False)
        self._main_widget = gtk.EventBox()
        self._main_widget.modify_bg(gtk.STATE_NORMAL, ful.BLACK)
        self._main_widget.add(self._label_list)

        ful.run_test_widget(self.job, self._main_widget,
                window_registration_callback=self.register_callbacks)

        if not self._result:
            raise error.TestFail('ERROR: loopback test fail')