blob: 8f9e5422a70cfd63262283de1cd7bcb584b7a93d [file] [log] [blame]
# Copyright (c) 2017 The WebRTC project authors. All Rights Reserved.
# Use of this source code is governed by a BSD-style license
# that can be found in the LICENSE file in the root of the source
# tree. An additional intellectual property rights grant can be found
# in the file PATENTS. All contributing project authors may
# be found in the AUTHORS file in the root of the source tree.
"""Input mixer module.
import logging
import os
from . import exceptions
from . import signal_processing
class ApmInputMixer(object):
"""Class to mix a set of audio segments down to the APM input."""
_HARD_CLIPPING_LOG_MSG = 'hard clipping detected in the mixed signal'
def __init__(self):
def HardClippingLogMessage(cls):
"""Returns the log message used when hard clipping is detected in the mix.
This method is mainly intended to be used by the unit tests.
def Mix(cls, output_path, capture_input_filepath, echo_filepath):
"""Mixes capture and echo.
Creates the overall capture input for APM by mixing the "echo-free" capture
signal with the echo signal (e.g., echo simulated via the
echo_path_simulation module).
The echo signal cannot be shorter than the capture signal and the generated
mix will have the same duration of the capture signal. The latter property
is enforced in order to let the input of APM and the reference signal
created by TestDataGenerator have the same length (required for the
evaluation step).
Hard-clipping may occur in the mix; a warning is raised when this happens.
If |echo_filepath| is None, nothing is done and |capture_input_filepath| is
speech: AudioSegment instance.
echo_path: AudioSegment instance or None.
Path to the mix audio track file.
if echo_filepath is None:
return capture_input_filepath
# Build the mix output file name as a function of the echo file name.
# This ensures that if the internal parameters of the echo path simulator
# change, no erroneous cache hit occurs.
echo_file_name, _ = os.path.splitext(os.path.split(echo_filepath)[1])
mix_filepath = os.path.join(output_path, 'mix_capture_{}.wav'.format(
# Create the mix if not done yet.
mix = None
if not os.path.exists(mix_filepath):
echo_free_capture = signal_processing.SignalProcessingUtils.LoadWav(
echo = signal_processing.SignalProcessingUtils.LoadWav(echo_filepath)
if signal_processing.SignalProcessingUtils.CountSamples(echo) < (
raise exceptions.InputMixerException(
'echo cannot be shorter than capture')
mix = echo_free_capture.overlay(echo)
signal_processing.SignalProcessingUtils.SaveWav(mix_filepath, mix)
# Check if hard clipping occurs.
if mix is None:
mix = signal_processing.SignalProcessingUtils.LoadWav(mix_filepath)
if signal_processing.SignalProcessingUtils.DetectHardClipping(mix):
return mix_filepath