| # 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. |
| """Extraction of annotations from audio files. |
| """ |
| |
| from __future__ import division |
| import logging |
| import os |
| import shutil |
| import struct |
| import subprocess |
| import sys |
| import tempfile |
| |
| try: |
| import numpy as np |
| except ImportError: |
| logging.critical('Cannot import the third-party Python package numpy') |
| sys.exit(1) |
| |
| from . import external_vad |
| from . import exceptions |
| from . import signal_processing |
| |
| |
| class AudioAnnotationsExtractor(object): |
| """Extracts annotations from audio files. |
| """ |
| |
| class VadType(object): |
| ENERGY_THRESHOLD = 1 # TODO(alessiob): Consider switching to P56 standard. |
| WEBRTC_COMMON_AUDIO = 2 # common_audio/vad/include/vad.h |
| WEBRTC_APM = 4 # modules/audio_processing/vad/vad.h |
| |
| def __init__(self, value): |
| if (not isinstance(value, int)) or not 0 <= value <= 7: |
| raise exceptions.InitializationException('Invalid vad type: ' + |
| value) |
| self._value = value |
| |
| def Contains(self, vad_type): |
| return self._value | vad_type == self._value |
| |
| def __str__(self): |
| vads = [] |
| if self.Contains(self.ENERGY_THRESHOLD): |
| vads.append("energy") |
| if self.Contains(self.WEBRTC_COMMON_AUDIO): |
| vads.append("common_audio") |
| if self.Contains(self.WEBRTC_APM): |
| vads.append("apm") |
| return "VadType({})".format(", ".join(vads)) |
| |
| _OUTPUT_FILENAME_TEMPLATE = '{}annotations.npz' |
| |
| # Level estimation params. |
| _ONE_DB_REDUCTION = np.power(10.0, -1.0 / 20.0) |
| _LEVEL_FRAME_SIZE_MS = 1.0 |
| # The time constants in ms indicate the time it takes for the level estimate |
| # to go down/up by 1 db if the signal is zero. |
| _LEVEL_ATTACK_MS = 5.0 |
| _LEVEL_DECAY_MS = 20.0 |
| |
| # VAD params. |
| _VAD_THRESHOLD = 1 |
| _VAD_WEBRTC_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), |
| os.pardir, os.pardir) |
| _VAD_WEBRTC_COMMON_AUDIO_PATH = os.path.join(_VAD_WEBRTC_PATH, 'vad') |
| |
| _VAD_WEBRTC_APM_PATH = os.path.join(_VAD_WEBRTC_PATH, 'apm_vad') |
| |
| def __init__(self, vad_type, external_vads=None): |
| self._signal = None |
| self._level = None |
| self._level_frame_size = None |
| self._common_audio_vad = None |
| self._energy_vad = None |
| self._apm_vad_probs = None |
| self._apm_vad_rms = None |
| self._vad_frame_size = None |
| self._vad_frame_size_ms = None |
| self._c_attack = None |
| self._c_decay = None |
| |
| self._vad_type = self.VadType(vad_type) |
| logging.info('VADs used for annotations: ' + str(self._vad_type)) |
| |
| if external_vads is None: |
| external_vads = {} |
| self._external_vads = external_vads |
| |
| assert len(self._external_vads) == len(external_vads), ( |
| 'The external VAD names must be unique.') |
| for vad in external_vads.values(): |
| if not isinstance(vad, external_vad.ExternalVad): |
| raise exceptions.InitializationException('Invalid vad type: ' + |
| str(type(vad))) |
| logging.info('External VAD used for annotation: ' + str(vad.name)) |
| |
| assert os.path.exists(self._VAD_WEBRTC_COMMON_AUDIO_PATH), \ |
| self._VAD_WEBRTC_COMMON_AUDIO_PATH |
| assert os.path.exists(self._VAD_WEBRTC_APM_PATH), \ |
| self._VAD_WEBRTC_APM_PATH |
| |
| @classmethod |
| def GetOutputFileNameTemplate(cls): |
| return cls._OUTPUT_FILENAME_TEMPLATE |
| |
| def GetLevel(self): |
| return self._level |
| |
| def GetLevelFrameSize(self): |
| return self._level_frame_size |
| |
| @classmethod |
| def GetLevelFrameSizeMs(cls): |
| return cls._LEVEL_FRAME_SIZE_MS |
| |
| def GetVadOutput(self, vad_type): |
| if vad_type == self.VadType.ENERGY_THRESHOLD: |
| return self._energy_vad |
| elif vad_type == self.VadType.WEBRTC_COMMON_AUDIO: |
| return self._common_audio_vad |
| elif vad_type == self.VadType.WEBRTC_APM: |
| return (self._apm_vad_probs, self._apm_vad_rms) |
| else: |
| raise exceptions.InitializationException('Invalid vad type: ' + |
| vad_type) |
| |
| def GetVadFrameSize(self): |
| return self._vad_frame_size |
| |
| def GetVadFrameSizeMs(self): |
| return self._vad_frame_size_ms |
| |
| def Extract(self, filepath): |
| # Load signal. |
| self._signal = signal_processing.SignalProcessingUtils.LoadWav( |
| filepath) |
| if self._signal.channels != 1: |
| raise NotImplementedError( |
| 'Multiple-channel annotations not implemented') |
| |
| # Level estimation params. |
| self._level_frame_size = int(self._signal.frame_rate / 1000 * |
| (self._LEVEL_FRAME_SIZE_MS)) |
| self._c_attack = 0.0 if self._LEVEL_ATTACK_MS == 0 else ( |
| self._ONE_DB_REDUCTION**(self._LEVEL_FRAME_SIZE_MS / |
| self._LEVEL_ATTACK_MS)) |
| self._c_decay = 0.0 if self._LEVEL_DECAY_MS == 0 else ( |
| self._ONE_DB_REDUCTION**(self._LEVEL_FRAME_SIZE_MS / |
| self._LEVEL_DECAY_MS)) |
| |
| # Compute level. |
| self._LevelEstimation() |
| |
| # Ideal VAD output, it requires clean speech with high SNR as input. |
| if self._vad_type.Contains(self.VadType.ENERGY_THRESHOLD): |
| # Naive VAD based on level thresholding. |
| vad_threshold = np.percentile(self._level, self._VAD_THRESHOLD) |
| self._energy_vad = np.uint8(self._level > vad_threshold) |
| self._vad_frame_size = self._level_frame_size |
| self._vad_frame_size_ms = self._LEVEL_FRAME_SIZE_MS |
| if self._vad_type.Contains(self.VadType.WEBRTC_COMMON_AUDIO): |
| # WebRTC common_audio/ VAD. |
| self._RunWebRtcCommonAudioVad(filepath, self._signal.frame_rate) |
| if self._vad_type.Contains(self.VadType.WEBRTC_APM): |
| # WebRTC modules/audio_processing/ VAD. |
| self._RunWebRtcApmVad(filepath) |
| for extvad_name in self._external_vads: |
| self._external_vads[extvad_name].Run(filepath) |
| |
| def Save(self, output_path, annotation_name=""): |
| ext_kwargs = { |
| 'extvad_conf-' + ext_vad: |
| self._external_vads[ext_vad].GetVadOutput() |
| for ext_vad in self._external_vads |
| } |
| np.savez_compressed(file=os.path.join( |
| output_path, |
| self.GetOutputFileNameTemplate().format(annotation_name)), |
| level=self._level, |
| level_frame_size=self._level_frame_size, |
| level_frame_size_ms=self._LEVEL_FRAME_SIZE_MS, |
| vad_output=self._common_audio_vad, |
| vad_energy_output=self._energy_vad, |
| vad_frame_size=self._vad_frame_size, |
| vad_frame_size_ms=self._vad_frame_size_ms, |
| vad_probs=self._apm_vad_probs, |
| vad_rms=self._apm_vad_rms, |
| **ext_kwargs) |
| |
| def _LevelEstimation(self): |
| # Read samples. |
| samples = signal_processing.SignalProcessingUtils.AudioSegmentToRawData( |
| self._signal).astype(np.float32) / 32768.0 |
| num_frames = len(samples) // self._level_frame_size |
| num_samples = num_frames * self._level_frame_size |
| |
| # Envelope. |
| self._level = np.max(np.reshape(np.abs(samples[:num_samples]), |
| (num_frames, self._level_frame_size)), |
| axis=1) |
| assert len(self._level) == num_frames |
| |
| # Envelope smoothing. |
| smooth = lambda curr, prev, k: (1 - k) * curr + k * prev |
| self._level[0] = smooth(self._level[0], 0.0, self._c_attack) |
| for i in range(1, num_frames): |
| self._level[i] = smooth( |
| self._level[i], self._level[i - 1], self._c_attack if |
| (self._level[i] > self._level[i - 1]) else self._c_decay) |
| |
| def _RunWebRtcCommonAudioVad(self, wav_file_path, sample_rate): |
| self._common_audio_vad = None |
| self._vad_frame_size = None |
| |
| # Create temporary output path. |
| tmp_path = tempfile.mkdtemp() |
| output_file_path = os.path.join( |
| tmp_path, |
| os.path.split(wav_file_path)[1] + '_vad.tmp') |
| |
| # Call WebRTC VAD. |
| try: |
| subprocess.call([ |
| self._VAD_WEBRTC_COMMON_AUDIO_PATH, '-i', wav_file_path, '-o', |
| output_file_path |
| ], |
| cwd=self._VAD_WEBRTC_PATH) |
| |
| # Read bytes. |
| with open(output_file_path, 'rb') as f: |
| raw_data = f.read() |
| |
| # Parse side information. |
| self._vad_frame_size_ms = struct.unpack('B', raw_data[0])[0] |
| self._vad_frame_size = self._vad_frame_size_ms * sample_rate / 1000 |
| assert self._vad_frame_size_ms in [10, 20, 30] |
| extra_bits = struct.unpack('B', raw_data[-1])[0] |
| assert 0 <= extra_bits <= 8 |
| |
| # Init VAD vector. |
| num_bytes = len(raw_data) |
| num_frames = 8 * (num_bytes - |
| 2) - extra_bits # 8 frames for each byte. |
| self._common_audio_vad = np.zeros(num_frames, np.uint8) |
| |
| # Read VAD decisions. |
| for i, byte in enumerate(raw_data[1:-1]): |
| byte = struct.unpack('B', byte)[0] |
| for j in range(8 if i < num_bytes - 3 else (8 - extra_bits)): |
| self._common_audio_vad[i * 8 + j] = int(byte & 1) |
| byte = byte >> 1 |
| except Exception as e: |
| logging.error('Error while running the WebRTC VAD (' + e.message + |
| ')') |
| finally: |
| if os.path.exists(tmp_path): |
| shutil.rmtree(tmp_path) |
| |
| def _RunWebRtcApmVad(self, wav_file_path): |
| # Create temporary output path. |
| tmp_path = tempfile.mkdtemp() |
| output_file_path_probs = os.path.join( |
| tmp_path, |
| os.path.split(wav_file_path)[1] + '_vad_probs.tmp') |
| output_file_path_rms = os.path.join( |
| tmp_path, |
| os.path.split(wav_file_path)[1] + '_vad_rms.tmp') |
| |
| # Call WebRTC VAD. |
| try: |
| subprocess.call([ |
| self._VAD_WEBRTC_APM_PATH, '-i', wav_file_path, '-o_probs', |
| output_file_path_probs, '-o_rms', output_file_path_rms |
| ], |
| cwd=self._VAD_WEBRTC_PATH) |
| |
| # Parse annotations. |
| self._apm_vad_probs = np.fromfile(output_file_path_probs, |
| np.double) |
| self._apm_vad_rms = np.fromfile(output_file_path_rms, np.double) |
| assert len(self._apm_vad_rms) == len(self._apm_vad_probs) |
| |
| except Exception as e: |
| logging.error('Error while running the WebRTC APM VAD (' + |
| e.message + ')') |
| finally: |
| if os.path.exists(tmp_path): |
| shutil.rmtree(tmp_path) |
