Restore playback port after each test

client/site_tests/audiovideo_PlaybackRecordSemiAuto/audiovideo_PlaybackRecordSemiAuto.py

 # Copyright (c) 2010 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.
 
 import copy, logging, os, pprint, re, threading, time, urllib
 
 from autotest_lib.client.bin import site_ui_test
 from autotest_lib.client.common_lib import error, site_httpd, site_ui, utils
 
 # HTML templates.
@@ skipping 71 matching lines @@
 <td>%(name)s</td>
 <td>%(index)d</td>
 <td>%(channels)d</td>
 <td>%(is_hardware)d</td>
 <td>%(sample_format)s</td>
 <td>%(sample_rate)s</td>
 <td>%(ports)s</td>
 </tr>
 '''
 
+_DEVICE_SECTION_ENTRY_PORT_START_TMPL = '''<tr>
+<td>%(name)s</td>
+<td>%(index)d</td>
+<td>%(channels)d</td>
+<td>%(is_hardware)d</td>
+<td>%(sample_format)s</td>
+<td>%(sample_rate)s</td>
+'''
+_DEVICE_SECTION_ENTRY_PORT_END_TMPL = '''<td>%s</td>
+</tr>
+'''
+
 _DEVICE_LIST_TEST = '''
 <tr><td><table> <tr>
 <td>Device List Looks Correct?</td>
 <td class="action" id="summary"><a href="?test=summary&result=pass">pass</a>
 <a href="?test=summary&result=fail">fail</a>
 </td>
 </tr></table></td></tr>
 '''
 
 _TEST_CONTROL_START = '<table><th>Test description</th><th>Invoke Link</th>'
@@ skipping 102 matching lines @@
 _STANZA_START_RE = re.compile('  (\*| ) index: (\d+)')
 _NAME_RE = re.compile('\tname:\s+<(.+)>')
 _FLAGS_RE = re.compile('\tflags:\s+(.+)')
 _MAX_VOLUME_RE = re.compile('\tvolume steps:\s+(\d+)')
 _BASE_VOLUME_RE = re.compile('\tbase volume:\s+(\d+)%')
 _SAMPLE_SPEC_RE = re.compile('\tsample spec:\s+(\S+) (\d+)ch (\d+)Hz')
 _CHANNEL_MAP_RE = re.compile('\tchannel map:\s+(.+)')
 _TOP_LEVEL_RE = re.compile('\t\S')
 _PORTS_RE = re.compile('\tports:')
 _PORT_SPEC_RE = re.compile('\t\t(\S+): .* \(priority \d+\)')
-
+_ACTIVE_PORT_RE = re.compile('\tactive port: <(.+)>')
 
 class ToneThread(threading.Thread):
     """Wraps the running of test_tones in a thread."""
     def __init__(self, audio, config):
         threading.Thread.__init__(self)
         self.audio = audio
         self.config = config
 
     def run(self):
         self.audio.run_test_tones(self.config)
@@ skipping 88 matching lines @@
 
         # Run test server.
         self._server_root = 'http://localhost:8000/'
         self._testServer = site_httpd.HTTPListener(port=8000,
                                                    docroot=self.bindir)
         self._testServer.run()
         site_ui_test.UITest.initialize(self, creds)
 
 
     def cleanup(self):
+        for device in self._playback_devices['info']:
+            if device['is_hardware']:
+                self.restore_playback_port(device)
         self._testServer.stop()
         site_ui_test.UITest.cleanup(self)
 
 
     def run_once(self, timeout=10000):
         self._testServer.add_url_handler(
                 '/%s' % _CONTROL_ENDPOINT,
                 lambda server, form, o=self: o.handle_control(server, form))
         self._testServer.add_url_handler(
                 '/%s' % _LIST_ENDPOINT,
@@ skipping 106 matching lines @@
         test_data = {
                 'test': _VOLUME_ENDPOINT,
                 'device': device_num,
                 'port': port_num
                 }
         server.wfile.write(
                 self.get_pass_fail_div(_CONTROL_ENDPOINT, test_data))
 
         server.wfile.write(_VOLUME_INSTRUCTIONS)
 
-        self.render_single_device_summary(server, device)
+        self.render_single_device_summary(server, device, port)
 
         server.wfile.write(_VOLUME_TEST_DETAILS % device)
         if device.has_key('channel_map'):
             server.wfile.write('<p>Channels are: %s' %
                     self._pp.pformat(device['channel_map']))
 
         # End Page.
         server.wfile.write(_HTML_FOOTER)
 
         self._running_test = threading.Thread(
@@ skipping 16 matching lines @@
         test_data = {
                 'test': _PLAYBACK_ENDPOINT,
                 'device': device_num,
                 'port': port_num
                 }
         server.wfile.write(
                 self.get_pass_fail_div(_CONTROL_ENDPOINT, test_data))
 
         server.wfile.write(_PLAYBACK_INSTRUCTIONS)
 
-        self.render_single_device_summary(server, device)
+        self.render_single_device_summary(server, device, port)
         self.render_channel_test_order(server, device, port)
 
         # End Page.
         server.wfile.write(_HTML_FOOTER)
 
         self.wait_for_current_test()
         self._running_test = threading.Thread(
                 target=lambda d=device,p=port: self.do_playback_test(d,p))
         self._running_test.start()
 
@@ skipping 13 matching lines @@
         test_data = {
                 'test': _RECORD_ENDPOINT,
                 'device': device_num,
                 'port': port_num
                 }
         server.wfile.write(
                 self.get_pass_fail_div(_CONTROL_ENDPOINT, test_data))
 
         server.wfile.write(_RECORD_INSTRUCTIONS)
 
-        self.render_single_device_summary(server, device)
+        self.render_single_device_summary(server, device, port)
         self.render_channel_test_order(server, device, port)
 
         # End Page.
         server.wfile.write(_HTML_FOOTER)
 
         self.wait_for_current_test()
         self._running_test = threading.Thread(
                 target=lambda d=device,p=port: self.do_record_test(d,p))
         self._running_test.start()
 
@@ skipping 62 matching lines @@
                         server.wfile.write(
                                 self.get_record_item(device_num, port_num))
                 else:
                     server.wfile.write(self.get_record_item(device_num))
         server.wfile.write(_TEST_CONTROL_END)
 
         # End Page.
         server.wfile.write(_HTML_FOOTER)
 
 
-    def render_single_device_summary(self, server, device):
+    def render_single_device_summary(self, server, device, port):
         """Output a HTML table with information on a single device"""
         server.wfile.write(_DEVICE_LIST_START)
         server.wfile.write(_DEVICE_SECTION_START)
-        server.wfile.write(_DEVICE_SECTION_ENTRY_TMPL % device)
+        server.wfile.write(_DEVICE_SECTION_ENTRY_PORT_START_TMPL % device)
+        server.wfile.write(_DEVICE_SECTION_ENTRY_PORT_END_TMPL % port)
         server.wfile.write(_DEVICE_SECTION_END)

[davejcool, 2010/09/04 00:45:44]

On the performing test page, it was listing ALL ports instead of the specific
port being tested, which could be confusing (it may list the 'speaker' port when
testing the 'headphone' port, where no audio would be heard out the speaker).

         server.wfile.write(_DEVICE_LIST_END)
 
 
     def render_channel_test_order(self, server, device, port):
         """Output HTML a table with device channel ordering info."""
         if port != None:
             server.wfile.write('<p>Active port on device: %s' % port)
         else:
             server.wfile.write('<p>Use default (only) port.')
 
@@ skipping 141 matching lines @@
             current_sink['sample_format'] = m.group(1)
             current_sink['channels'] = int(m.group(2))
             current_sink['sample_rate'] = int(m.group(3))
 
         m = _CHANNEL_MAP_RE.match(line)
         if m is not None:
             channel_map = []
             for channel in m.group(1).split(','):
                 channel_map.append(channel)
             current_sink['channel_map'] = channel_map
+        
+        m = _ACTIVE_PORT_RE.match(line)
+        if m is not None:
+            current_sink['active_port'] = m.group(1)
 
 
     def parse_device_info(self, device_info_output):
         """Parses the output of a pacmd list-sources or list-sinks call."""
         device_info = { 'info' : [] }
         current_device = None
         port_parsing_mode = False
         for line in device_info_output.split('\n'):
             # Leave port_parsing_mode if we find a top-level attribute.
             if port_parsing_mode and _TOP_LEVEL_RE.match(line) is not None:
@@ skipping 63 matching lines @@
         """
         self.do_pacmd('set-default-%s %d' % (type, device['index']))
         logging.info(
             '* Testing device %d (%s)' % (device['index'], device['name']))
 
         if port is not None:
             self.do_pacmd('set-%s-port %d %s' % (type, device['index'], port))
             logging.info('-- setting port %s' % port)
 
 
+    def restore_playback_port(self, device):
+        """Restores the sink's active port to what it was before testing.
+
+        Args:
+          device: A dictionary with the parsed device information.
+        """
+        if device['active_port'] is not None:
+            self.do_pacmd('set-sink-port %d %s' %
+                (device['index'], device['active_port']))
+            logging.info('Restoring device %d to port %s' %
+                (device['index'], device['active_port']))
+
+
     def do_signal_test_end(self):
         """Play 3 short 1000Hz tones to signal a test case's completion.
 
         Playback is done on whatever the current default device is.
         """
         config = self.default_tone_config()
         config['tone_length_sec'] = 0.3
         self.play_tone(config, 1000)
         self.play_tone(config, 1000)
         self.play_tone(config, 1000)
 
 
     def do_record_test(self, device, port):
         """Performs a record test for the given device and port.
 
         This sets the default playback device is set to whatever device
         is returned first in enumerate_playback_devices().  The playback
-        device is set to use its first port, unmuted, and set to the
-        test_volume.
+        device is set to use the default port (active port before running
+        tests), unmuted, and set to the test_volume.
 
         For each channel on the given device and port, a sample is recorded
         and played-back at max source volume. Then once again with all
         channel enabled.
 
         Next, a sample is taken at base volume (no amplification) if
         that is available, and played back.
 
         This is followed by a sample at 1/2 amplifcation, and again with
         the record device muted.
 
         Args:
             device: device info dictionary gotten from
                     enumerate_record devices()
             port: String with the name of the port to use on the device.
                   Can be None if the device does not have multiple ports.
         """
         # Configure the playback device to something normal.
         playback_device = self._playback_devices['info'][0]
         playback_volume = self.get_test_volume(playback_device)
         playback_port = None

[davejcool, 2010/09/04 00:45:44]

Setting playback port the the system default, most likely to be heard rather
than the first port in the list, which on some devices may be one where you can
not hear the audio.

-        if len(playback_device['ports']):
+        if playback_device['active_port'] is not None:
+            self.restore_playback_port(playback_device)
+        elif len(playback_device['ports']):
             playback_port = playback_device['ports'][0]
         self.set_default_device_and_port('sink', playback_device,
                                          playback_port)
         self.do_set_mute('sink', playback_device['index'], False)
         self.do_set_volume('sink', playback_device['index'], playback_volume)
 
         # Set record device.
         self.set_default_device_and_port('source', device, port)
 
         # Set to max hardware amplification volume.
@@ skipping 338 matching lines @@
         cmd = ('%(exec)s '
                '-t %(type)s -h %(frequency)f -l %(tone_length_sec)f '
                '-c %(channels)d' % args)
         if args['active_channel'] is not None:
             cmd += ' -a %s' % args['active_channel']
         if args['type'] == 'tone':
             logging.info('[tone %dHz]' % args['frequency'])
         elif args['type'] == 'scale':
             logging.info('[A# harmonic minor scale]')
         utils.system(self.pacmd(cmd))