• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1# Copyright 2019 The Chromium Authors. All rights reserved.
2# Use of this source code is governed by a BSD-style license that can be
3# found in the LICENSE file.
4
5import logging
6import os
7import time
8
9from autotest_lib.client.bin import utils
10from autotest_lib.client.common_lib import error
11from autotest_lib.client.cros.audio import audio_helper
12from autotest_lib.client.cros.audio import audio_spec
13from autotest_lib.client.cros.audio import audio_test_data
14from autotest_lib.client.cros.audio import cmd_utils
15from autotest_lib.client.cros.audio import cras_utils
16
17TEST_DURATION = 1
18
19
20class audio_CrasPinnedStream(audio_helper.cras_rms_test):
21    """Verifies audio capture function on multiple devices."""
22    version = 1
23
24    @staticmethod
25    def wait_for_active_stream_count(expected_count):
26        """Wait until the active stream count is correct"""
27
28        utils.poll_for_condition(
29                lambda: cras_utils.get_active_stream_count() == expected_count,
30                exception=error.TestError(
31                        'Timeout waiting active stream count to become %d' %
32                        expected_count))
33
34    @staticmethod
35    def contains_all_zero(path):
36        """Check the recorded sample contains none zero data"""
37        with open(path, 'rb') as f:
38            samples = f.read()
39            for sample in samples:
40                if sample != '\x00':
41                    return False
42        return True
43
44    def run_once(self):
45        """Entry point of this test."""
46
47        # The test requires internal mic as second capture device.
48        if not audio_spec.has_internal_microphone(utils.get_board_type()):
49            logging.info("No internal mic. Skipping the test.")
50            return
51
52        raw_file = audio_test_data.GenerateAudioTestData(
53                path=os.path.join(self.bindir, '5SEC.raw'),
54                duration_secs=5,
55                frequencies=[440, 440],
56                volume_scale=0.9)
57
58        loopback_recorded_file = os.path.join(self.resultsdir,
59                                              'loopback_recorded.raw')
60        internal_mic_recorded_file = os.path.join(self.resultsdir,
61                                                  'internal_mic_recorded.raw')
62        node_type = audio_spec.get_internal_mic_node(utils.get_board_type(),
63                                                     utils.get_board(),
64                                                     utils.get_platform(),
65                                                     utils.get_sku())
66        device_id = int(
67                cras_utils.get_device_id_from_node_type(node_type, True))
68
69        self.wait_for_active_stream_count(0)
70        p = cmd_utils.popen(cras_utils.playback_cmd(raw_file.path))
71        try:
72            loop_c = cmd_utils.popen(
73                    cras_utils.capture_cmd(
74                            loopback_recorded_file, duration=TEST_DURATION))
75            int_c = cmd_utils.popen(
76                    cras_utils.capture_cmd(
77                            internal_mic_recorded_file,
78                            duration=TEST_DURATION,
79                            pin_device=device_id))
80
81            # Make sure the audio is still playing.
82            if p.poll() != None:
83                raise error.TestError('playback stopped')
84            # Make sure the recordings finish.
85            time.sleep(2 * TEST_DURATION)
86        finally:
87            cmd_utils.kill_or_log_returncode(p, int_c, loop_c)
88            raw_file.delete()
89
90        rms_value = audio_helper.get_rms(loopback_recorded_file)[0]
91        self.write_perf_keyval({'rms_value': rms_value})
92
93        audio_helper.recorded_filesize_check(
94                os.path.getsize(internal_mic_recorded_file), TEST_DURATION)
95        audio_helper.recorded_filesize_check(
96                os.path.getsize(loopback_recorded_file), TEST_DURATION)
97
98        if self.contains_all_zero(internal_mic_recorded_file):
99            raise error.TestError('Record all zero from internal mic')
100