server/site_tests/audio_AudioNoiseCancellation/audio_AudioNoiseCancellation.py

# Copyright 2021 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.
"""This is a server side noise cancellation test using the Chameleon board."""

import logging
import os
import time

from autotest_lib.client.common_lib import error
from autotest_lib.client.cros.audio import audio_test_data
from autotest_lib.client.cros.audio import visqol_utils
from autotest_lib.client.cros.bluetooth.bluetooth_audio_test_data import (
        download_file_from_bucket, get_visqol_binary)
from autotest_lib.client.cros.chameleon import audio_test_utils
from autotest_lib.client.cros.chameleon import chameleon_audio_ids
from autotest_lib.client.cros.chameleon import chameleon_audio_helper
from autotest_lib.server.cros.audio import audio_test

DIST_FILES = 'gs://chromeos-localmirror/distfiles'
DATA_DIR = '/tmp'


class audio_AudioNoiseCancellation(audio_test.AudioTest):
    """Server side input audio noise cancellation test.

    This test talks to a Chameleon board and a Cros device to verify
    input audio noise cancellation function of the Cros device.

    """
    version = 1
    DELAY_BEFORE_RECORD_SECONDS = 0.5
    RECORD_SECONDS = 10
    DELAY_AFTER_BINDING = 0.5

    # This is a 11-second, 2-channel raw audio which will be played by Chameleon
    # speaker on test. It is mixed with the speech reference and the simulated
    # office ambient noise.
    SPEECH_WITH_NOISE_FILE = 'speech_with_noise.raw'

    # This is a 9-second, 1-channel wav file which will be the reference file on
    # VISQOL score calculation. It is the pure speech reference without noise.
    SPEECH_REFERENCE_FILE = 'speech_ref.wav'

    # VISQOL score is ranged from 1.0 to 5.0; the larger score means the better
    # speech quality.
    MIN_VISQOL_SCORE = 3.5

    def cleanup(self):
        # Restore the default state of bypass blocking mechanism in Cras.
        self.facade.set_bypass_block_noise_cancellation(bypass=False)

        # Remove downloaded speech file and reference file.
        if os.path.exists(os.path.join(DATA_DIR, self.SPEECH_WITH_NOISE_FILE)):
            os.remove(os.path.join(DATA_DIR, self.SPEECH_WITH_NOISE_FILE))
        if os.path.exists(os.path.join(DATA_DIR, self.SPEECH_REFERENCE_FILE)):
            os.remove(os.path.join(DATA_DIR, self.SPEECH_REFERENCE_FILE))

    def run_once(self):
        """Runs Audio Noise Cancellation test."""
        if not self.facade.get_noise_cancellation_supported():
            logging.warning('Noise Cancellation is not supported.')
            raise error.TestWarn('Noise Cancellation is not supported.')

        # Download the speech with noise file from bucket.
        remote_path = os.path.join(DIST_FILES, self.SPEECH_WITH_NOISE_FILE)
        if not download_file_from_bucket(
                DATA_DIR, remote_path, lambda _, __, p: p.returncode == 0):
            logging.error('Failed to download %s to %s', remote_path, DATA_DIR)
            raise error.TestError(
                    'Failed to download speech file from bucket.')

        speech_file = audio_test_data.AudioTestData(
                path=os.path.join(DATA_DIR, self.SPEECH_WITH_NOISE_FILE),
                data_format=dict(file_type='raw',
                                 sample_format='S16_LE',
                                 channel=2,
                                 rate=48000),
                duration_secs=11.0)

        # Get and set VISQOL working environment.
        get_visqol_binary()

        # Download the speech reference file from bucket.
        remote_path = os.path.join(DIST_FILES, self.SPEECH_REFERENCE_FILE)
        if not download_file_from_bucket(
                DATA_DIR, remote_path, lambda _, __, p: p.returncode == 0):
            logging.error('Failed to download %s to %s', remote_path, DATA_DIR)
            raise error.TestError('Failed to download ref file from bucket.')

        # Bypass blocking mechanism in Cras to make sure Noise Cancellation is
        # enabled.
        self.facade.set_bypass_block_noise_cancellation(bypass=True)

        source = self.widget_factory.create_widget(
                chameleon_audio_ids.ChameleonIds.LINEOUT)
        sink = self.widget_factory.create_widget(
                chameleon_audio_ids.PeripheralIds.SPEAKER)
        binder = self.widget_factory.create_binder(source, sink)

        recorder = self.widget_factory.create_widget(
                chameleon_audio_ids.CrosIds.INTERNAL_MIC)

        with chameleon_audio_helper.bind_widgets(binder):
            time.sleep(self.DELAY_AFTER_BINDING)

            audio_test_utils.dump_cros_audio_logs(self.host, self.facade,
                                                  self.resultsdir,
                                                  'after_binding')

            # Selects and checks the node selected by cras is correct.
            audio_test_utils.check_and_set_chrome_active_node_types(
                    self.facade, None,
                    audio_test_utils.get_internal_mic_node(self.host))

            logging.info('Setting playback data on Chameleon')
            source.set_playback_data(speech_file)

            # Starts playing, waits for some time, and then starts recording.
            # This is to avoid artifact caused by chameleon codec initialization
            # in the beginning of playback.
            logging.info('Start playing %s from Chameleon', speech_file.path)
            source.start_playback()

            time.sleep(self.DELAY_BEFORE_RECORD_SECONDS)
            logging.info('Start recording from Cros device.')
            recorder.start_recording()

            time.sleep(self.RECORD_SECONDS)

            recorder.stop_recording()
            logging.info('Stopped recording from Cros device.')

            audio_test_utils.dump_cros_audio_logs(self.host, self.facade,
                                                  self.resultsdir,
                                                  'after_recording')

            recorder.read_recorded_binary()
            logging.info('Read recorded binary from Cros device.')

        # Removes the beginning of recorded data. This is to avoid artifact
        # caused by Cros device codec initialization in the beginning of
        # recording.
        recorder.remove_head(1.0)

        recorded_file = os.path.join(self.resultsdir, "recorded.raw")
        logging.info('Saving recorded data to %s', recorded_file)
        recorder.save_file(recorded_file)

        # WAV file is also saved by recorder.save_file().
        recorded_wav_path = recorded_file + '.wav'
        if not os.path.isfile(recorded_wav_path):
            logging.error('WAV file %s does not exist.', recorded_wav_path)
            raise error.TestError('Failed to find recorded wav file.')

        # Get VISQOL score. The score should be high because we expect the
        # recorded data is already noise-cancelled. Set speech_mode to False
        # because the recording rate is 48k, while speech_mode only accepts 16k
        # rate.
        ref_wav_path = os.path.join(DATA_DIR, self.SPEECH_REFERENCE_FILE)
        score = visqol_utils.get_visqol_score(ref_file=ref_wav_path,
                                              deg_file=recorded_wav_path,
                                              log_dir=self.resultsdir,
                                              speech_mode=False)

        logging.info('Got score %f, min passing score: %f', score,
                     self.MIN_VISQOL_SCORE)

        # Track VISQOL performance score
        test_desc = 'internal_mic_noise_cancellation'
        self.write_perf_keyval({test_desc: score})

        if score < self.MIN_VISQOL_SCORE:
            raise error.TestError(
                    'Failed to pass visqol score; got: %f, min: %f' %
                    (score, self.MIN_VISQOL_SCORE))