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