# 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. """Signal processing utility module. """ import array import logging import os import sys import enum try: import numpy as np except ImportError: logging.critical('Cannot import the third-party Python package numpy') sys.exit(1) try: import pydub import pydub.generators except ImportError: logging.critical('Cannot import the third-party Python package pydub') sys.exit(1) try: import scipy.signal import scipy.fftpack except ImportError: logging.critical('Cannot import the third-party Python package scipy') sys.exit(1) from . import exceptions class SignalProcessingUtils(object): """Collection of signal processing utilities. """ @enum.unique class MixPadding(enum.Enum): NO_PADDING = 0 ZERO_PADDING = 1 LOOP = 2 def __init__(self): pass @classmethod def LoadWav(cls, filepath, channels=1): """Loads wav file. Args: filepath: path to the wav audio track file to load. channels: number of channels (downmixing to mono by default). Returns: AudioSegment instance. """ if not os.path.exists(filepath): logging.error('cannot find the <%s> audio track file', filepath) raise exceptions.FileNotFoundError() return pydub.AudioSegment.from_file(filepath, format='wav', channels=channels) @classmethod def SaveWav(cls, output_filepath, signal): """Saves wav file. Args: output_filepath: path to the wav audio track file to save. signal: AudioSegment instance. """ return signal.export(output_filepath, format='wav') @classmethod def CountSamples(cls, signal): """Number of samples per channel. Args: signal: AudioSegment instance. Returns: An integer. """ number_of_samples = len(signal.get_array_of_samples()) assert signal.channels > 0 assert number_of_samples % signal.channels == 0 return number_of_samples / signal.channels @classmethod def GenerateSilence(cls, duration=1000, sample_rate=48000): """Generates silence. This method can also be used to create a template AudioSegment instance. A template can then be used with other Generate*() methods accepting an AudioSegment instance as argument. Args: duration: duration in ms. sample_rate: sample rate. Returns: AudioSegment instance. """ return pydub.AudioSegment.silent(duration, sample_rate) @classmethod def GeneratePureTone(cls, template, frequency=440.0): """Generates a pure tone. The pure tone is generated with the same duration and in the same format of the given template signal. Args: template: AudioSegment instance. frequency: Frequency of the pure tone in Hz. Return: AudioSegment instance. """ if frequency > template.frame_rate >> 1: raise exceptions.SignalProcessingException('Invalid frequency') generator = pydub.generators.Sine(sample_rate=template.frame_rate, bit_depth=template.sample_width * 8, freq=frequency) return generator.to_audio_segment(duration=len(template), volume=0.0) @classmethod def GenerateWhiteNoise(cls, template): """Generates white noise. The white noise is generated with the same duration and in the same format of the given template signal. Args: template: AudioSegment instance. Return: AudioSegment instance. """ generator = pydub.generators.WhiteNoise( sample_rate=template.frame_rate, bit_depth=template.sample_width * 8) return generator.to_audio_segment(duration=len(template), volume=0.0) @classmethod def AudioSegmentToRawData(cls, signal): samples = signal.get_array_of_samples() if samples.typecode != 'h': raise exceptions.SignalProcessingException( 'Unsupported samples type') return np.array(signal.get_array_of_samples(), np.int16) @classmethod def Fft(cls, signal, normalize=True): if signal.channels != 1: raise NotImplementedError('multiple-channel FFT not implemented') x = cls.AudioSegmentToRawData(signal).astype(np.float32) if normalize: x /= max(abs(np.max(x)), 1.0) y = scipy.fftpack.fft(x) return y[:len(y) / 2] @classmethod def DetectHardClipping(cls, signal, threshold=2): """Detects hard clipping. Hard clipping is simply detected by counting samples that touch either the lower or upper bound too many times in a row (according to `threshold`). The presence of a single sequence of samples meeting such property is enough to label the signal as hard clipped. Args: signal: AudioSegment instance. threshold: minimum number of samples at full-scale in a row. Returns: True if hard clipping is detect, False otherwise. """ if signal.channels != 1: raise NotImplementedError( 'multiple-channel clipping not implemented') if signal.sample_width != 2: # Note that signal.sample_width is in bytes. raise exceptions.SignalProcessingException( 'hard-clipping detection only supported for 16 bit samples') samples = cls.AudioSegmentToRawData(signal) # Detect adjacent clipped samples. samples_type_info = np.iinfo(samples.dtype) mask_min = samples == samples_type_info.min mask_max = samples == samples_type_info.max def HasLongSequence(vector, min_legth=threshold): """Returns True if there are one or more long sequences of True flags.""" seq_length = 0 for b in vector: seq_length = seq_length + 1 if b else 0 if seq_length >= min_legth: return True return False return HasLongSequence(mask_min) or HasLongSequence(mask_max) @classmethod def ApplyImpulseResponse(cls, signal, impulse_response): """Applies an impulse response to a signal. Args: signal: AudioSegment instance. impulse_response: list or numpy vector of float values. Returns: AudioSegment instance. """ # Get samples. assert signal.channels == 1, ( 'multiple-channel recordings not supported') samples = signal.get_array_of_samples() # Convolve. logging.info( 'applying %d order impulse response to a signal lasting %d ms', len(impulse_response), len(signal)) convolved_samples = scipy.signal.fftconvolve(in1=samples, in2=impulse_response, mode='full').astype( np.int16) logging.info('convolution computed') # Cast. convolved_samples = array.array(signal.array_type, convolved_samples) # Verify. logging.debug('signal length: %d samples', len(samples)) logging.debug('convolved signal length: %d samples', len(convolved_samples)) assert len(convolved_samples) > len(samples) # Generate convolved signal AudioSegment instance. convolved_signal = pydub.AudioSegment(data=convolved_samples, metadata={ 'sample_width': signal.sample_width, 'frame_rate': signal.frame_rate, 'frame_width': signal.frame_width, 'channels': signal.channels, }) assert len(convolved_signal) > len(signal) return convolved_signal @classmethod def Normalize(cls, signal): """Normalizes a signal. Args: signal: AudioSegment instance. Returns: An AudioSegment instance. """ return signal.apply_gain(-signal.max_dBFS) @classmethod def Copy(cls, signal): """Makes a copy os a signal. Args: signal: AudioSegment instance. Returns: An AudioSegment instance. """ return pydub.AudioSegment(data=signal.get_array_of_samples(), metadata={ 'sample_width': signal.sample_width, 'frame_rate': signal.frame_rate, 'frame_width': signal.frame_width, 'channels': signal.channels, }) @classmethod def MixSignals(cls, signal, noise, target_snr=0.0, pad_noise=MixPadding.NO_PADDING): """Mixes `signal` and `noise` with a target SNR. Mix `signal` and `noise` with a desired SNR by scaling `noise`. If the target SNR is +/- infinite, a copy of signal/noise is returned. If `signal` is shorter than `noise`, the length of the mix equals that of `signal`. Otherwise, the mix length depends on whether padding is applied. When padding is not applied, that is `pad_noise` is set to NO_PADDING (default), the mix length equals that of `noise` - i.e., `signal` is truncated. Otherwise, `noise` is extended and the resulting mix has the same length of `signal`. Args: signal: AudioSegment instance (signal). noise: AudioSegment instance (noise). target_snr: float, numpy.Inf or -numpy.Inf (dB). pad_noise: SignalProcessingUtils.MixPadding, default: NO_PADDING. Returns: An AudioSegment instance. """ # Handle infinite target SNR. if target_snr == -np.Inf: # Return a copy of noise. logging.warning('SNR = -Inf, returning noise') return cls.Copy(noise) elif target_snr == np.Inf: # Return a copy of signal. logging.warning('SNR = +Inf, returning signal') return cls.Copy(signal) # Check signal and noise power. signal_power = float(signal.dBFS) noise_power = float(noise.dBFS) if signal_power == -np.Inf: logging.error('signal has -Inf power, cannot mix') raise exceptions.SignalProcessingException( 'cannot mix a signal with -Inf power') if noise_power == -np.Inf: logging.error('noise has -Inf power, cannot mix') raise exceptions.SignalProcessingException( 'cannot mix a signal with -Inf power') # Mix. gain_db = signal_power - noise_power - target_snr signal_duration = len(signal) noise_duration = len(noise) if signal_duration <= noise_duration: # Ignore `pad_noise`, `noise` is truncated if longer that `signal`, the # mix will have the same length of `signal`. return signal.overlay(noise.apply_gain(gain_db)) elif pad_noise == cls.MixPadding.NO_PADDING: # `signal` is longer than `noise`, but no padding is applied to `noise`. # Truncate `signal`. return noise.overlay(signal, gain_during_overlay=gain_db) elif pad_noise == cls.MixPadding.ZERO_PADDING: # TODO(alessiob): Check that this works as expected. return signal.overlay(noise.apply_gain(gain_db)) elif pad_noise == cls.MixPadding.LOOP: # `signal` is longer than `noise`, extend `noise` by looping. return signal.overlay(noise.apply_gain(gain_db), loop=True) else: raise exceptions.SignalProcessingException('invalid padding type')