1 # Copyright 2015 The Chromium OS 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 5 """This module provides the test utilities for audio tests using chameleon.""" 6 7 # TODO (cychiang) Move test utilities from chameleon_audio_helpers 8 # to this module. 9 10 import logging 11 import multiprocessing 12 import os 13 import pprint 14 import re 15 import time 16 from contextlib import contextmanager 17 18 from autotest_lib.client.common_lib import error 19 from autotest_lib.client.cros import constants 20 from autotest_lib.client.cros.audio import audio_analysis 21 from autotest_lib.client.cros.audio import audio_data 22 from autotest_lib.client.cros.audio import audio_helper 23 from autotest_lib.client.cros.audio import audio_quality_measurement 24 from autotest_lib.client.cros.chameleon import chameleon_audio_ids 25 26 CHAMELEON_AUDIO_IDS_TO_CRAS_NODE_TYPES = { 27 chameleon_audio_ids.CrosIds.HDMI: 'HDMI', 28 chameleon_audio_ids.CrosIds.HEADPHONE: 'HEADPHONE', 29 chameleon_audio_ids.CrosIds.EXTERNAL_MIC: 'MIC', 30 chameleon_audio_ids.CrosIds.SPEAKER: 'INTERNAL_SPEAKER', 31 chameleon_audio_ids.CrosIds.INTERNAL_MIC: 'INTERNAL_MIC', 32 chameleon_audio_ids.CrosIds.BLUETOOTH_HEADPHONE: 'BLUETOOTH', 33 chameleon_audio_ids.CrosIds.BLUETOOTH_MIC: 'BLUETOOTH', 34 chameleon_audio_ids.CrosIds.USBIN: 'USB', 35 chameleon_audio_ids.CrosIds.USBOUT: 'USB', 36 } 37 38 39 def cros_port_id_to_cras_node_type(port_id): 40 """Gets Cras node type from Cros port id. 41 42 @param port_id: A port id defined in chameleon_audio_ids.CrosIds. 43 44 @returns: A Cras node type defined in cras_utils.CRAS_NODE_TYPES. 45 46 """ 47 return CHAMELEON_AUDIO_IDS_TO_CRAS_NODE_TYPES[port_id] 48 49 50 def check_output_port(audio_facade, port_id): 51 """Checks selected output node on Cros device is correct for a port. 52 53 @param port_id: A port id defined in chameleon_audio_ids.CrosIds. 54 55 """ 56 output_node_type = cros_port_id_to_cras_node_type(port_id) 57 check_audio_nodes(audio_facade, ([output_node_type], None)) 58 59 60 def check_input_port(audio_facade, port_id): 61 """Checks selected input node on Cros device is correct for a port. 62 63 @param port_id: A port id defined in chameleon_audio_ids.CrosIds. 64 65 """ 66 input_node_type = cros_port_id_to_cras_node_type(port_id) 67 check_audio_nodes(audio_facade, (None, [input_node_type])) 68 69 70 def check_audio_nodes(audio_facade, audio_nodes): 71 """Checks the node selected by Cros device is correct. 72 73 @param audio_facade: A RemoteAudioFacade to access audio functions on 74 Cros device. 75 76 @param audio_nodes: A tuple (out_audio_nodes, in_audio_nodes) containing 77 expected selected output and input nodes. 78 79 @raises: error.TestFail if the nodes selected by Cros device are not expected. 80 81 """ 82 curr_out_nodes, curr_in_nodes = audio_facade.get_selected_node_types() 83 out_audio_nodes, in_audio_nodes = audio_nodes 84 if (in_audio_nodes != None and 85 sorted(curr_in_nodes) != sorted(in_audio_nodes)): 86 raise error.TestFail('Wrong input node(s) selected: %s ' 87 'expected: %s' % (str(curr_in_nodes), str(in_audio_nodes))) 88 89 # Treat line-out node as headphone node in Chameleon test since some 90 # Cros devices detect audio board as lineout. This actually makes sense 91 # because 3.5mm audio jack is connected to LineIn port on Chameleon. 92 if (out_audio_nodes == ['HEADPHONE'] and curr_out_nodes == ['LINEOUT']): 93 return 94 95 if (out_audio_nodes != None and 96 sorted(curr_out_nodes) != sorted(out_audio_nodes)): 97 raise error.TestFail('Wrong output node(s) selected %s ' 98 'expected: %s' % (str(curr_out_nodes), str(out_audio_nodes))) 99 100 101 def check_plugged_nodes(audio_facade, audio_nodes): 102 """Checks the nodes that are currently plugged on Cros device are correct. 103 104 @param audio_facade: A RemoteAudioFacade to access audio functions on 105 Cros device. 106 107 @param audio_nodes: A tuple (out_audio_nodes, in_audio_nodes) containing 108 expected plugged output and input nodes. 109 110 @raises: error.TestFail if the plugged nodes on Cros device are not expected. 111 112 """ 113 curr_out_nodes, curr_in_nodes = audio_facade.get_plugged_node_types() 114 out_audio_nodes, in_audio_nodes = audio_nodes 115 if (in_audio_nodes != None and 116 sorted(curr_in_nodes) != sorted(in_audio_nodes)): 117 raise error.TestFail('Wrong input node(s) plugged: %s ' 118 'expected: %s!' % (str(curr_in_nodes), str(in_audio_nodes))) 119 if (out_audio_nodes != None and 120 sorted(curr_out_nodes) != sorted(out_audio_nodes)): 121 raise error.TestFail('Wrong output node(s) plugged: %s ' 122 'expected: %s!' % (str(curr_out_nodes), str(out_audio_nodes))) 123 124 125 def bluetooth_nodes_plugged(audio_facade): 126 """Checks bluetooth nodes are plugged. 127 128 @param audio_facade: A RemoteAudioFacade to access audio functions on 129 Cros device. 130 131 @raises: error.TestFail if either input or output bluetooth node is 132 not plugged. 133 134 """ 135 curr_out_nodes, curr_in_nodes = audio_facade.get_plugged_node_types() 136 return 'BLUETOOTH' in curr_out_nodes and 'BLUETOOTH' in curr_in_nodes 137 138 139 def _get_board_name(host): 140 """Gets the board name. 141 142 @param host: The CrosHost object. 143 144 @returns: The board name. 145 146 """ 147 return host.get_board().split(':')[1] 148 149 150 def has_internal_speaker(host): 151 """Checks if the Cros device has speaker. 152 153 @param host: The CrosHost object. 154 155 @returns: True if Cros device has internal speaker. False otherwise. 156 157 """ 158 board_name = _get_board_name(host) 159 if host.get_board_type() == 'CHROMEBOX' and board_name != 'stumpy': 160 logging.info('Board %s does not have speaker.', board_name) 161 return False 162 return True 163 164 165 def has_internal_microphone(host): 166 """Checks if the Cros device has internal microphone. 167 168 @param host: The CrosHost object. 169 170 @returns: True if Cros device has internal microphone. False otherwise. 171 172 """ 173 board_name = _get_board_name(host) 174 if host.get_board_type() == 'CHROMEBOX': 175 logging.info('Board %s does not have internal microphone.', board_name) 176 return False 177 return True 178 179 180 def has_headphone(host): 181 """Checks if the Cros device has headphone. 182 183 @param host: The CrosHost object. 184 185 @returns: True if Cros device has headphone. False otherwise. 186 187 """ 188 board_name = _get_board_name(host) 189 if host.get_board_type() == 'CHROMEBIT': 190 logging.info('Board %s does not have headphone.', board_name) 191 return False 192 return True 193 194 195 def suspend_resume(host, suspend_time_secs, resume_network_timeout_secs=50): 196 """Performs the suspend/resume on Cros device. 197 198 @param suspend_time_secs: Time in seconds to let Cros device suspend. 199 @resume_network_timeout_secs: Time in seconds to let Cros device resume and 200 obtain network. 201 """ 202 def action_suspend(): 203 """Calls the host method suspend.""" 204 host.suspend(suspend_time=suspend_time_secs) 205 206 boot_id = host.get_boot_id() 207 proc = multiprocessing.Process(target=action_suspend) 208 logging.info("Suspending...") 209 proc.daemon = True 210 proc.start() 211 host.test_wait_for_sleep(suspend_time_secs / 3) 212 logging.info("DUT suspended! Waiting to resume...") 213 host.test_wait_for_resume( 214 boot_id, suspend_time_secs + resume_network_timeout_secs) 215 logging.info("DUT resumed!") 216 217 218 def dump_cros_audio_logs(host, audio_facade, directory, suffix=''): 219 """Dumps logs for audio debugging from Cros device. 220 221 @param host: The CrosHost object. 222 @param audio_facade: A RemoteAudioFacade to access audio functions on 223 Cros device. 224 @directory: The directory to dump logs. 225 226 """ 227 def get_file_path(name): 228 """Gets file path to dump logs. 229 230 @param name: The file name. 231 232 @returns: The file path with an optional suffix. 233 234 """ 235 file_name = '%s.%s' % (name, suffix) if suffix else name 236 file_path = os.path.join(directory, file_name) 237 return file_path 238 239 audio_facade.dump_diagnostics(get_file_path('audio_diagnostics.txt')) 240 241 host.get_file('/var/log/messages', get_file_path('messages')) 242 243 host.get_file(constants.MULTIMEDIA_XMLRPC_SERVER_LOG_FILE, 244 get_file_path('multimedia_xmlrpc_server.log')) 245 246 247 def examine_audio_diagnostics(path): 248 """Examines audio diagnostic content. 249 250 @param path: Path to audio diagnostic file. 251 252 @returns: Warning messages or ''. 253 254 """ 255 warning_msgs = [] 256 line_number = 1 257 258 underrun_pattern = re.compile('num_underruns: (\d*)') 259 260 with open(path) as f: 261 for line in f.readlines(): 262 263 # Check for number of underruns. 264 search_result = underrun_pattern.search(line) 265 if search_result: 266 num_underruns = int(search_result.group(1)) 267 if num_underruns != 0: 268 warning_msgs.append( 269 'Found %d underrun at line %d: %s' % ( 270 num_underruns, line_number, line)) 271 272 # TODO(cychiang) add other check like maximum client reply delay. 273 line_number = line_number + 1 274 275 if warning_msgs: 276 return ('Found issue in audio diganostics result : %s' % 277 '\n'.join(warning_msgs)) 278 279 logging.info('audio_diagnostic result looks fine') 280 return '' 281 282 283 @contextmanager 284 def monitor_no_nodes_changed(audio_facade, callback=None): 285 """Context manager to monitor nodes changed signal on Cros device. 286 287 Starts the counter in the beginning. Stops the counter in the end to make 288 sure there is no NodesChanged signal during the try block. 289 290 E.g. with monitor_no_nodes_changed(audio_facade): 291 do something on playback/recording 292 293 @param audio_facade: A RemoteAudioFacade to access audio functions on 294 Cros device. 295 @param fail_callback: The callback to call before raising TestFail 296 when there is unexpected NodesChanged signals. 297 298 @raises: error.TestFail if there is NodesChanged signal on 299 Cros device during the context. 300 301 """ 302 try: 303 audio_facade.start_counting_signal('NodesChanged') 304 yield 305 finally: 306 count = audio_facade.stop_counting_signal() 307 if count: 308 message = 'Got %d unexpected NodesChanged signal' % count 309 logging.error(message) 310 if callback: 311 callback() 312 raise error.TestFail(message) 313 314 315 # The second dominant frequency should have energy less than -26dB of the 316 # first dominant frequency in the spectrum. 317 _DEFAULT_SECOND_PEAK_RATIO = 0.05 318 319 # Tolerate more noise for bluetooth audio using HSP. 320 _HSP_SECOND_PEAK_RATIO = 0.2 321 322 # Tolerate more noise for speaker. 323 _SPEAKER_SECOND_PEAK_RATIO = 0.1 324 325 # Tolerate more noise for internal microphone. 326 _INTERNAL_MIC_SECOND_PEAK_RATIO = 0.2 327 328 # maximum tolerant noise level 329 DEFAULT_TOLERANT_NOISE_LEVEL = 0.01 330 331 # If relative error of two durations is less than 0.2, 332 # they will be considered equivalent. 333 DEFAULT_EQUIVALENT_THRESHOLD = 0.2 334 335 # The frequency at lower than _DC_FREQ_THRESHOLD should have coefficient 336 # smaller than _DC_COEFF_THRESHOLD. 337 _DC_FREQ_THRESHOLD = 0.001 338 _DC_COEFF_THRESHOLD = 0.01 339 340 def get_second_peak_ratio(source_id, recorder_id, is_hsp=False): 341 """Gets the second peak ratio suitable for use case. 342 343 @param source_id: ID defined in chameleon_audio_ids for source widget. 344 @param recorder_id: ID defined in chameleon_audio_ids for recorder widget. 345 @param is_hsp: For bluetooth HSP use case. 346 347 @returns: A float for proper second peak ratio to be used in 348 check_recorded_frequency. 349 """ 350 if is_hsp: 351 return _HSP_SECOND_PEAK_RATIO 352 elif source_id == chameleon_audio_ids.CrosIds.SPEAKER: 353 return _SPEAKER_SECOND_PEAK_RATIO 354 elif recorder_id == chameleon_audio_ids.CrosIds.INTERNAL_MIC: 355 return _INTERNAL_MIC_SECOND_PEAK_RATIO 356 else: 357 return _DEFAULT_SECOND_PEAK_RATIO 358 359 360 # The deviation of estimated dominant frequency from golden frequency. 361 DEFAULT_FREQUENCY_DIFF_THRESHOLD = 5 362 363 def check_recorded_frequency( 364 golden_file, recorder, 365 second_peak_ratio=_DEFAULT_SECOND_PEAK_RATIO, 366 frequency_diff_threshold=DEFAULT_FREQUENCY_DIFF_THRESHOLD, 367 ignore_frequencies=None, check_anomaly=False, check_artifacts=False, 368 mute_durations=None, volume_changes=None, 369 tolerant_noise_level=DEFAULT_TOLERANT_NOISE_LEVEL): 370 """Checks if the recorded data contains sine tone of golden frequency. 371 372 @param golden_file: An AudioTestData object that serves as golden data. 373 @param recorder: An AudioWidget used in the test to record data. 374 @param second_peak_ratio: The test fails when the second dominant 375 frequency has coefficient larger than this 376 ratio of the coefficient of first dominant 377 frequency. 378 @param frequency_diff_threshold: The maximum difference between estimated 379 frequency of test signal and golden 380 frequency. This value should be small for 381 signal passed through line. 382 @param ignore_frequencies: A list of frequencies to be ignored. The 383 component in the spectral with frequency too 384 close to the frequency in the list will be 385 ignored. The comparison of frequencies uses 386 frequency_diff_threshold as well. 387 @param check_anomaly: True to check anomaly in the signal. 388 @param check_artifacts: True to check artifacts in the signal. 389 @param mute_durations: Each duration of mute in seconds in the signal. 390 @param volume_changes: A list containing alternative -1 for decreasing 391 volume and +1 for increasing volume. 392 @param tolerant_noise_level: The maximum noise level can be tolerated 393 394 @returns: A list containing tuples of (dominant_frequency, coefficient) for 395 valid channels. Coefficient can be a measure of signal magnitude 396 on that dominant frequency. Invalid channels where golden_channel 397 is None are ignored. 398 399 @raises error.TestFail if the recorded data does not contain sine tone of 400 golden frequency. 401 402 """ 403 if not ignore_frequencies: 404 ignore_frequencies = [] 405 406 # Also ignore harmonics of ignore frequencies. 407 ignore_frequencies_harmonics = [] 408 for ignore_freq in ignore_frequencies: 409 ignore_frequencies_harmonics += [ignore_freq * n for n in xrange(1, 4)] 410 411 data_format = recorder.data_format 412 recorded_data = audio_data.AudioRawData( 413 binary=recorder.get_binary(), 414 channel=data_format['channel'], 415 sample_format=data_format['sample_format']) 416 417 errors = [] 418 dominant_spectrals = [] 419 420 for test_channel, golden_channel in enumerate(recorder.channel_map): 421 if golden_channel is None: 422 logging.info('Skipped channel %d', test_channel) 423 continue 424 425 signal = recorded_data.channel_data[test_channel] 426 saturate_value = audio_data.get_maximum_value_from_sample_format( 427 data_format['sample_format']) 428 logging.debug('Channel %d max signal: %f', test_channel, max(signal)) 429 normalized_signal = audio_analysis.normalize_signal( 430 signal, saturate_value) 431 logging.debug('saturate_value: %f', saturate_value) 432 logging.debug('max signal after normalized: %f', max(normalized_signal)) 433 spectral = audio_analysis.spectral_analysis( 434 normalized_signal, data_format['rate']) 435 logging.debug('spectral: %s', spectral) 436 437 if not spectral: 438 errors.append( 439 'Channel %d: Can not find dominant frequency.' % 440 test_channel) 441 442 golden_frequency = golden_file.frequencies[golden_channel] 443 logging.debug('Checking channel %s spectral %s against frequency %s', 444 test_channel, spectral, golden_frequency) 445 446 dominant_frequency = spectral[0][0] 447 448 if (abs(dominant_frequency - golden_frequency) > 449 frequency_diff_threshold): 450 errors.append( 451 'Channel %d: Dominant frequency %s is away from golden %s' % 452 (test_channel, dominant_frequency, golden_frequency)) 453 454 if check_anomaly: 455 detected_anomaly = audio_analysis.anomaly_detection( 456 signal=normalized_signal, 457 rate=data_format['rate'], 458 freq=golden_frequency) 459 if detected_anomaly: 460 errors.append( 461 'Channel %d: Detect anomaly near these time: %s' % 462 (test_channel, detected_anomaly)) 463 else: 464 logging.info( 465 'Channel %d: Quality is good as there is no anomaly', 466 test_channel) 467 468 if check_artifacts or mute_durations or volume_changes: 469 result = audio_quality_measurement.quality_measurement( 470 normalized_signal, 471 data_format['rate'], 472 dominant_frequency=dominant_frequency) 473 logging.debug('Quality measurement result:\n%s', pprint.pformat(result)) 474 if check_artifacts: 475 if len(result['artifacts']['noise_before_playback']) > 0: 476 errors.append( 477 'Channel %d: Detects artifacts before playing near' 478 ' these time and duration: %s' % 479 (test_channel, 480 str(result['artifacts']['noise_before_playback']))) 481 482 if len(result['artifacts']['noise_after_playback']) > 0: 483 errors.append( 484 'Channel %d: Detects artifacts after playing near' 485 ' these time and duration: %s' % 486 (test_channel, 487 str(result['artifacts']['noise_after_playback']))) 488 489 if mute_durations: 490 delays = result['artifacts']['delay_during_playback'] 491 delay_durations = [] 492 for x in delays: 493 delay_durations.append(x[1]) 494 mute_matched, delay_matched = longest_common_subsequence( 495 mute_durations, 496 delay_durations, 497 DEFAULT_EQUIVALENT_THRESHOLD) 498 499 # updated delay list 500 new_delays = [delays[i] 501 for i in delay_matched if not delay_matched[i]] 502 503 result['artifacts']['delay_during_playback'] = new_delays 504 505 unmatched_mutes = [mute_durations[i] 506 for i in mute_matched if not mute_matched[i]] 507 508 if len(unmatched_mutes) > 0: 509 errors.append( 510 'Channel %d: Unmatched mute duration: %s' % 511 (test_channel, unmatched_mutes)) 512 513 if check_artifacts: 514 if len(result['artifacts']['delay_during_playback']) > 0: 515 errors.append( 516 'Channel %d: Detects delay during playing near' 517 ' these time and duration: %s' % 518 (test_channel, 519 result['artifacts']['delay_during_playback'])) 520 521 if len(result['artifacts']['burst_during_playback']) > 0: 522 errors.append( 523 'Channel %d: Detects burst/pop near these time: %s' % 524 (test_channel, 525 result['artifacts']['burst_during_playback'])) 526 527 if result['equivalent_noise_level'] > tolerant_noise_level: 528 errors.append( 529 'Channel %d: noise level is higher than tolerant' 530 ' noise level: %f > %f' % 531 (test_channel, 532 result['equivalent_noise_level'], 533 tolerant_noise_level)) 534 535 if volume_changes: 536 matched = True 537 volume_changing = result['volume_changes'] 538 if len(volume_changing) != len(volume_changes): 539 matched = False 540 else: 541 for i in xrange(len(volume_changing)): 542 if volume_changing[i][1] != volume_changes[i]: 543 matched = False 544 break 545 if not matched: 546 errors.append( 547 'Channel %d: volume changing is not as expected, ' 548 'found changing time and events are: %s while ' 549 'expected changing events are %s'% 550 (test_channel, 551 volume_changing, 552 volume_changes)) 553 554 # Filter out the harmonics resulted from imperfect sin wave. 555 # This list is different for different channels. 556 harmonics = [dominant_frequency * n for n in xrange(2, 10)] 557 558 def should_be_ignored(frequency): 559 """Checks if frequency is close to any frequency in ignore list. 560 561 The ignore list is harmonics of frequency to be ignored 562 (like power noise), plus harmonics of dominant frequencies, 563 plus DC. 564 565 @param frequency: The frequency to be tested. 566 567 @returns: True if the frequency should be ignored. False otherwise. 568 569 """ 570 for ignore_frequency in (ignore_frequencies_harmonics + harmonics 571 + [0.0]): 572 if (abs(frequency - ignore_frequency) < 573 frequency_diff_threshold): 574 logging.debug('Ignore frequency: %s', frequency) 575 return True 576 577 # Checks DC is small enough. 578 for freq, coeff in spectral: 579 if freq < _DC_FREQ_THRESHOLD and coeff > _DC_COEFF_THRESHOLD: 580 errors.append( 581 'Channel %d: Found large DC coefficient: ' 582 '(%f Hz, %f)' % (test_channel, freq, coeff)) 583 584 # Filter out the frequencies to be ignored. 585 spectral = [x for x in spectral if not should_be_ignored(x[0])] 586 587 if len(spectral) > 1: 588 first_coeff = spectral[0][1] 589 second_coeff = spectral[1][1] 590 if second_coeff > first_coeff * second_peak_ratio: 591 errors.append( 592 'Channel %d: Found large second dominant frequencies: ' 593 '%s' % (test_channel, spectral)) 594 595 dominant_spectrals.append(spectral[0]) 596 597 if errors: 598 raise error.TestFail(', '.join(errors)) 599 600 return dominant_spectrals 601 602 603 def longest_common_subsequence(list1, list2, equivalent_threshold): 604 """Finds longest common subsequence of list1 and list2 605 606 Such as list1: [0.3, 0.4], 607 list2: [0.001, 0.299, 0.002, 0.401, 0.001] 608 equivalent_threshold: 0.001 609 it will return matched1: [True, True], 610 matched2: [False, True, False, True, False] 611 612 @param list1: a list of integer or float value 613 @param list2: a list of integer or float value 614 @param equivalent_threshold: two values are considered equivalent if their 615 relative error is less than 616 equivalent_threshold. 617 618 @returns: a tuple of list (matched_1, matched_2) indicating each item 619 of list1 and list2 are matched or not. 620 621 """ 622 length1, length2 = len(list1), len(list2) 623 matching = [[0] * (length2 + 1)] * (length1 + 1) 624 # matching[i][j] is the maximum number of matched pairs for first i items 625 # in list1 and first j items in list2. 626 for i in xrange(length1): 627 for j in xrange(length2): 628 # Maximum matched pairs may be obtained without 629 # i-th item in list1 or without j-th item in list2 630 matching[i + 1][j + 1] = max(matching[i + 1][j], 631 matching[i][j + 1]) 632 diff = abs(list1[i] - list2[j]) 633 relative_error = diff / list1[i] 634 # If i-th item in list1 can be matched to j-th item in list2 635 if relative_error < equivalent_threshold: 636 matching[i + 1][j + 1] = matching[i][j] + 1 637 638 # Backtracking which item in list1 and list2 are matched 639 matched1 = [False] * length1 640 matched2 = [False] * length2 641 i, j = length1, length2 642 while i > 0 and j > 0: 643 # Maximum number is obtained by matching i-th item in list1 644 # and j-th one in list2. 645 if matching[i][j] == matching[i - 1][j - 1] + 1: 646 matched1[i - 1] = True 647 matched2[j - 1] = True 648 i, j = i - 1, j - 1 649 elif matching[i][j] == matching[i - 1][j]: 650 i -= 1 651 else: 652 j -= 1 653 return (matched1, matched2) 654 655 656 def switch_to_hsp(audio_facade): 657 """Switches to HSP profile. 658 659 Selects bluetooth microphone and runs a recording process on Cros device. 660 This triggers bluetooth profile be switched from A2DP to HSP. 661 Note the user can call stop_recording on audio facade to stop the recording 662 process, or let multimedia_xmlrpc_server terminates it in its cleanup. 663 664 """ 665 audio_facade.set_chrome_active_node_type(None, 'BLUETOOTH') 666 check_audio_nodes(audio_facade, (None, ['BLUETOOTH'])) 667 audio_facade.start_recording( 668 dict(file_type='raw', sample_format='S16_LE', channel=2, 669 rate=48000)) 670 671 672 def compare_recorded_correlation(golden_file, recorder, parameters=None): 673 """Checks recorded audio in an AudioInputWidget against a golden file. 674 675 Compares recorded data with golden data by cross correlation method. 676 Refer to audio_helper.compare_data for details of comparison. 677 678 @param golden_file: An AudioTestData object that serves as golden data. 679 @param recorder: An AudioInputWidget that has recorded some audio data. 680 @param parameters: A dict containing parameters for method. 681 682 """ 683 logging.info('Comparing recorded data with golden file %s ...', 684 golden_file.path) 685 audio_helper.compare_data_correlation( 686 golden_file.get_binary(), golden_file.data_format, 687 recorder.get_binary(), recorder.data_format, recorder.channel_map, 688 parameters) 689 690 691 def check_and_set_chrome_active_node_types(audio_facade, output_type=None, 692 input_type=None): 693 """Check the target types are available, and set them to be active nodes. 694 695 @param audio_facade: An AudioFacadeNative or AudioFacadeAdapter object. 696 @output_type: An output node type defined in cras_utils.CRAS_NODE_TYPES. 697 None to skip. 698 @input_type: An input node type defined in cras_utils.CRAS_NODE_TYPES. 699 None to skip. 700 701 @raises: error.TestError if the expected node type is missing. We use 702 error.TestError here because usually this step is not the main 703 purpose of the test, but a setup step. 704 705 """ 706 output_types, input_types = audio_facade.get_plugged_node_types() 707 logging.debug('Plugged types: output: %r, input: %r', 708 output_types, input_types) 709 if output_type and output_type not in output_types: 710 raise error.TestError( 711 'Target output type %s not present' % output_type) 712 if input_type and input_type not in input_types: 713 raise error.TestError( 714 'Target input type %s not present' % input_type) 715 audio_facade.set_chrome_active_node_type(output_type, input_type) 716 717 718 def check_hp_or_lineout_plugged(audio_facade): 719 """Checks whether line-out or headphone is plugged. 720 721 @param audio_facade: A RemoteAudioFacade to access audio functions on 722 Cros device. 723 724 @returns: 'LINEOUT' if line-out node is plugged. 725 'HEADPHONE' if headphone node is plugged. 726 727 @raises: error.TestFail if the plugged nodes does not contain one of 728 'LINEOUT' and 'HEADPHONE'. 729 730 """ 731 # Checks whether line-out or headphone is detected. 732 output_nodes, _ = audio_facade.get_plugged_node_types() 733 if 'LINEOUT' in output_nodes: 734 return 'LINEOUT' 735 if 'HEADPHONE' in output_nodes: 736 return 'HEADPHONE' 737 raise error.TestFail('Can not detect line-out or headphone') 738
