| # -*- 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. |
| |
| |
| # 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 gobject |
| import gtk |
| import os |
| import re |
| import socket |
| import subprocess |
| import tempfile |
| |
| from autotest_lib.client.bin import test, utils |
| 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 |
| |
| # 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' |
| |
| # Label strings. |
| _LABEL_CONNECTED = 'Connected\n已連線\n' |
| _LABEL_WAITING = 'Waiting for command\n等待指令中\n' |
| _LABEL_AUDIOLOOP = 'Audio looping\n音源回放中\n' |
| _LABEL_SPEAKER_MUTE_OFF = 'Speaker on\n喇叭開啟\n' |
| _LABEL_PLAYTONE_LEFT = ('Playing tone to left channel\n' |
| '播音至左聲道\n') |
| _LABEL_PLAYTONE_RIGHT = ('Playing tone to right channel\n' |
| '播音至右聲道\n') |
| |
| # 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") |
| _SEND_FILE_RE = re.compile("(?i)send_file\,\s*[^\,]+\,\s*(\d)+$") |
| _TEST_COMPLETE_RE = re.compile("(?i)test_complete") |
| _RESULT_PASS_RE = re.compile("(?i)result_pass") |
| _RESULT_FAIL_RE = re.compile("(?i)result_fail") |
| |
| class factory_AudioQuality(test.test): |
| version = 1 |
| |
| def handle_connection(self, conn, *args): |
| ''' |
| Asynchronous handler for socket connection. |
| ''' |
| line = conn.recv(1024) |
| if line: |
| factory.log("Received command %s" % line) |
| else: |
| return False |
| |
| for key in self._handlers.iterkeys(): |
| if key.match(line): |
| self._handlers[key](conn, line) |
| break |
| |
| # Respond by the received command with '_OK' postfix. |
| conn.send(line + '_OK') |
| return False |
| |
| def start_loop(self): |
| ''' |
| Starts the internal audio loopback. |
| ''' |
| self._loop_process = subprocess.Popen( |
| [self._ah.sox_path, '-d', '-d']) |
| |
| def play_tone(self): |
| ''' |
| Plays a single tone. |
| ''' |
| cmdargs = [self._ah.sox_path, '-t', 'null', '/dev/null', '-d', 'synth', |
| '20.0', 'sine', '1000.0'] |
| self._play_tone_process = subprocess.Popen(cmdargs) |
| |
| def restore_configuration(self): |
| ''' |
| Stops all the running process and restore the mute settings. |
| ''' |
| if hasattr(self, '_play_tone_process') and self._play_tone_process: |
| self._play_tone_process.kill() |
| if hasattr(self, '_loop_process') and self._loop_process: |
| self._loop_process.kill() |
| factory.log("Stopped audio loop process") |
| self.mute_headphone_left_right(False, False) |
| self.set_auto_mute(True) |
| |
| def handle_send_file(self, *args): |
| conn = args[0] |
| conn.send('OK') |
| |
| params = args[1].split(',') |
| file_name = params[1] |
| size = int(params[2]) |
| |
| with tempfile.NamedTemporaryFile(mode='w+t') as tmp_file: |
| factory.log("created tmp_file: %s\n" % tmp_file.name) |
| tmp_file.write('File name: %s\n' % file_name) |
| |
| # A message DONE will be concatenated to the end of detailed log. |
| left = size + 4 |
| while left > 0: |
| data = conn.recv(1024) |
| left -= len(data) |
| tmp_file.write(data) |
| tmp_file.seek(0) |
| for line in tmp_file: |
| self._detail_log += line |
| |
| factory.log("Received file %s with size %d" % (file_name, size)) |
| |
| def handle_result_pass(self, *args): |
| self._test_passed = True |
| |
| def handle_result_fail(self, *args): |
| self._test_passed = False |
| |
| def handle_test_complete(self, *args): |
| gtk.main_quit() |
| |
| def handle_loop_none(self, *args): |
| self.restore_configuration() |
| self._loop_status_label.set_text(_LABEL_WAITING) |
| |
| def handle_loop(self, *args): |
| self.restore_configuration() |
| self._loop_status_label.set_text(_LABEL_AUDIOLOOP) |
| self.start_loop() |
| |
| def handle_loop_speaker_unmute(self, *args): |
| self.handle_loop() |
| self._loop_status_label.set_text(_LABEL_AUDIOLOOP + |
| _LABEL_SPEAKER_MUTE_OFF) |
| self.set_auto_mute(False) |
| |
| def handle_xtalk_left(self, *args): |
| self.restore_configuration() |
| self._loop_status_label.set_text(_LABEL_PLAYTONE_LEFT) |
| self.mute_headphone_left_right(False, True) |
| self.play_tone() |
| |
| def handle_xtalk_right(self, *args): |
| self.restore_configuration() |
| self._loop_status_label.set_text(_LABEL_PLAYTONE_RIGHT) |
| self.mute_headphone_left_right(True, False) |
| self.play_tone() |
| |
| def listen(self, sock, *args): |
| ''' |
| Listens for connection and start handler for it. |
| ''' |
| conn, addr = sock.accept() |
| self._loop_status_label.set_text("Connected") |
| gobject.io_add_watch(conn, gobject.IO_IN, self.handle_connection) |
| return True |
| |
| def start_server(self): |
| ''' |
| Initialize server and start listening for external commands. |
| ''' |
| sock = socket.socket() |
| sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) |
| sock.bind((_HOST, _PORT)) |
| sock.listen(1) |
| factory.log("Listening at port %d" % _PORT) |
| gobject.io_add_watch(sock, gobject.IO_IN, self.listen) |
| |
| def mute_headphone_left_right(self, left=False, right=False): |
| ''' |
| Mutes specified headphone channels. |
| ''' |
| left_vol = 100 if left else 0 |
| right_vol = 100 if right else 0 |
| mixer_settings = [{'name': "'Headphone Playback Volume'", |
| 'value': ('%d%%,%d%%' % (left_vol, right_vol))}] |
| self._ah.set_mixer_controls(mixer_settings) |
| |
| def set_auto_mute(self, enable=True): |
| ''' |
| Sets the auto-mute mode. When auto-mute is enabled, the speaker will be |
| muted automatically when an external mic detected. |
| ''' |
| mixer_settings = [{'name': "'Auto-Mute Mode'", |
| 'value': ('Enabled' if enable else 'Disabled')}] |
| self._ah.set_mixer_controls(mixer_settings) |
| |
| def on_test_complete(self): |
| ''' |
| Restores the original state before exiting the test. |
| ''' |
| os.system('iptables -D INPUT -p tcp --dport %s -j ACCEPT' % _PORT) |
| os.system('ifconfig eth0 down') |
| os.system('ifconfig eth0 up') |
| self.restore_configuration() |
| self._ah.cleanup_deps(['sox']) |
| |
| def key_release_callback(self, widget, event): |
| # Hit Q to force quit this test. |
| if event.keyval == ord('Q'): |
| self._test_passed = False |
| gtk.main_quit() |
| |
| def register_callbacks(self, window): |
| window.connect('key-release-event', self.key_release_callback) |
| |
| def check_eth_state(self): |
| path = '/sys/class/net/eth0/carrier' |
| output = None |
| try: |
| if os.path.exists(path): |
| output = open(path).read() |
| finally: |
| if output: |
| return output == '1\n' |
| else: |
| return False |
| |
| def run_once(self, audio_sample_path=None, audio_init_volume=None): |
| factory.log('%s run_once' % self.__class__) |
| |
| self._ah = audio_helper.AudioHelper(self) |
| self._ah.setup_deps(['sox']) |
| self._detail_log = '' |
| |
| factory.log('Checking eth0 state....') |
| utils.poll_for_condition(self.check_eth_state, |
| timeout=30, desc='Checking eth0') |
| factory.log('Checking eth0 done!') |
| |
| # Configure local network environment to accept command from test host. |
| os.system('ifconfig eth0 %s netmask 255.255.255.0 up' % _LOCAL_IP) |
| os.system('iptables -A INPUT -p tcp --dport %s -j ACCEPT' % _PORT) |
| |
| # 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 |
| self._handlers[_LOOP_2_RE] = self.handle_loop_speaker_unmute |
| self._handlers[_LOOP_3_RE] = self.handle_loop |
| self._handlers[_XTALK_L_RE] = self.handle_xtalk_left |
| self._handlers[_XTALK_R_RE] = self.handle_xtalk_right |
| |
| self.start_server() |
| |
| self._main_widget = gtk.EventBox() |
| self._main_widget.modify_bg(gtk.STATE_NORMAL, ful.BLACK) |
| self._loop_status_label = ful.make_label('No audio loop', fg=ful.WHITE) |
| self._main_widget.add(self._loop_status_label) |
| |
| ful.run_test_widget(self.job, self._main_widget, |
| window_registration_callback=self.register_callbacks, |
| cleanup_callback=self.on_test_complete) |
| if not self._test_passed: |
| factory.log(self._detail_log) |
| raise error.TestError('Test failed.') |
| |
| factory.log('%s run_once finished' % self.__class__) |
