• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1# Lint as: python2, python3
2# Copyright (c) 2012 The Chromium Authors. All rights reserved.
3# Use of this source code is governed by a BSD-style license that can be
4# found in the LICENSE file.
5
6import logging
7import os
8import subprocess
9import tempfile
10
11from autotest_lib.client.bin import utils
12from autotest_lib.client.common_lib import error
13from autotest_lib.client.cros.audio import audio_helper
14from autotest_lib.client.cros.audio import cmd_utils
15from autotest_lib.client.cros.audio import cras_utils
16from autotest_lib.client.cros.audio import sox_utils
17
18
19_TEST_TONE_ONE = 440
20_TEST_TONE_TWO = 523
21
22class audio_CRASFormatConversion(audio_helper.cras_rms_test):
23    """Checks that sample rate conversion works in CRAS."""
24
25    version = 1
26
27
28    def play_sine_tone(self, frequency, rate):
29        """Plays a sine tone by cras and returns the processes.
30
31        @param frequency: the frequency of the sine wave.
32        @param rate: the sampling rate.
33        """
34        p1 = cmd_utils.popen(
35            sox_utils.generate_sine_tone_cmd(
36                    filename='-', rate=rate, frequencies=frequency, gain=-6),
37            stdout=subprocess.PIPE)
38        p2 = cras_utils.playback(blocking=False,
39                                 stdin=p1.stdout,
40                                 playback_file='-',
41                                 rate=rate)
42        return [p1, p2]
43
44
45    def wait_for_active_stream_count(self, expected_count):
46        """Waits until the number of active streams matches the requested
47        number or until a timeout occurs.
48
49        @param expected_count: the exact number of streams required to
50                               be active for execution to continue.
51
52        @raise TestError: if a timeout occurs.
53        """
54
55        utils.poll_for_condition(
56                lambda: cras_utils.get_active_stream_count() == expected_count,
57                exception=error.TestError(
58                        'Timeout waiting active stream count to become %d' %
59                        expected_count),
60                timeout=1, sleep_interval=0.05)
61
62    def loopback(self, noise_profile, primary, secondary):
63        """Gets the rms value of the recorded audio of playing two different
64        tones (the 440 and 523 Hz sine wave) at the specified sampling rate.
65
66        @param noise_profile: The noise profile which is used to reduce the
67                              noise of the recored audio.
68        @param primary: The first sample rate, HW will be set to this.
69        @param secondary: The second sample rate, will be SRC'd to the first.
70        """
71        popens = []
72
73        record_file = os.path.join(self.resultsdir,
74                'record-%s-%s.wav' % (primary, secondary))
75
76        # There should be no other active streams.
77        self.wait_for_active_stream_count(0)
78
79        # Start with the primary sample rate, then add the secondary.  This
80        # causes the secondary to be SRC'd to the primary rate.
81        try:
82            # Play the first audio stream and make sure it has been played
83            popens += self.play_sine_tone(_TEST_TONE_ONE, primary)
84            self.wait_for_active_stream_count(1)
85
86            # Play the second audio stream and make sure it has been played
87            popens += self.play_sine_tone(_TEST_TONE_TWO, secondary)
88            self.wait_for_active_stream_count(2)
89
90            cras_utils.capture(record_file, duration=1, rate=44100)
91
92            # Make sure the playback is still in good shape
93            if any(p.poll() is not None for p in popens):
94                # We will log more details later in finally.
95                raise error.TestFail('process unexpectly stopped')
96
97            reduced_file = tempfile.NamedTemporaryFile()
98            sox_utils.noise_reduce(
99                    record_file, reduced_file.name, noise_profile, rate=44100)
100
101            sox_stat = sox_utils.get_stat(reduced_file.name, rate=44100)
102
103            logging.info('The sox stat of (%d, %d) is %s',
104                         primary, secondary, str(sox_stat))
105
106            return sox_stat.rms
107
108        finally:
109            cmd_utils.kill_or_log_returncode(*popens)
110
111    def run_once(self, test_sample_rates):
112        """Runs the format conversion test.
113        """
114
115        rms_values = {}
116
117        # Record silence to use as the noise profile.
118        noise_file = os.path.join(self.resultsdir, "noise.wav")
119        noise_profile = tempfile.NamedTemporaryFile()
120        cras_utils.capture(noise_file, duration=1)
121        sox_utils.noise_profile(noise_file, noise_profile.name)
122
123        # Try all sample rate pairs.
124        for primary in test_sample_rates:
125            for secondary in test_sample_rates:
126                key = 'rms_value_%d_%d' % (primary, secondary)
127                rms_values[key] = self.loopback(
128                        noise_profile.name, primary, secondary)
129
130        # Record at all sample rates
131        record_file = tempfile.NamedTemporaryFile()
132        for rate in test_sample_rates:
133            cras_utils.capture(record_file.name, duration=1, rate=rate)
134
135        # Add min_rms_value to the result
136        rms_values['min_rms_value'] = min(rms_values.values())
137
138        self.write_perf_keyval(rms_values)
139