Spaces:
Sleeping
Sleeping
| # Copyright (c) 2023 Amphion. | |
| # | |
| # This source code is licensed under the MIT license found in the | |
| # LICENSE file in the root directory of this source tree. | |
| import os | |
| import json | |
| import numpy as np | |
| from tqdm import tqdm | |
| import torch | |
| import torchaudio | |
| from utils.io import save_audio | |
| from utils.audio import load_audio_torch | |
| # This function is obtained from librosa. | |
| def get_rms( | |
| y, | |
| *, | |
| frame_length=2048, | |
| hop_length=512, | |
| pad_mode="constant", | |
| ): | |
| padding = (int(frame_length // 2), int(frame_length // 2)) | |
| y = np.pad(y, padding, mode=pad_mode) | |
| axis = -1 | |
| # put our new within-frame axis at the end for now | |
| out_strides = y.strides + tuple([y.strides[axis]]) | |
| # Reduce the shape on the framing axis | |
| x_shape_trimmed = list(y.shape) | |
| x_shape_trimmed[axis] -= frame_length - 1 | |
| out_shape = tuple(x_shape_trimmed) + tuple([frame_length]) | |
| xw = np.lib.stride_tricks.as_strided(y, shape=out_shape, strides=out_strides) | |
| if axis < 0: | |
| target_axis = axis - 1 | |
| else: | |
| target_axis = axis + 1 | |
| xw = np.moveaxis(xw, -1, target_axis) | |
| # Downsample along the target axis | |
| slices = [slice(None)] * xw.ndim | |
| slices[axis] = slice(0, None, hop_length) | |
| x = xw[tuple(slices)] | |
| # Calculate power | |
| power = np.mean(np.abs(x) ** 2, axis=-2, keepdims=True) | |
| return np.sqrt(power) | |
| class Slicer: | |
| """ | |
| Copy from: https://github.com/openvpi/audio-slicer/blob/main/slicer2.py | |
| """ | |
| def __init__( | |
| self, | |
| sr: int, | |
| threshold: float = -40.0, | |
| min_length: int = 5000, | |
| min_interval: int = 300, | |
| hop_size: int = 10, | |
| max_sil_kept: int = 5000, | |
| ): | |
| if not min_length >= min_interval >= hop_size: | |
| raise ValueError( | |
| "The following condition must be satisfied: min_length >= min_interval >= hop_size" | |
| ) | |
| if not max_sil_kept >= hop_size: | |
| raise ValueError( | |
| "The following condition must be satisfied: max_sil_kept >= hop_size" | |
| ) | |
| min_interval = sr * min_interval / 1000 | |
| self.threshold = 10 ** (threshold / 20.0) | |
| self.hop_size = round(sr * hop_size / 1000) | |
| self.win_size = min(round(min_interval), 4 * self.hop_size) | |
| self.min_length = round(sr * min_length / 1000 / self.hop_size) | |
| self.min_interval = round(min_interval / self.hop_size) | |
| self.max_sil_kept = round(sr * max_sil_kept / 1000 / self.hop_size) | |
| def _apply_slice(self, waveform, begin, end): | |
| begin = begin * self.hop_size | |
| if len(waveform.shape) > 1: | |
| end = min(waveform.shape[1], end * self.hop_size) | |
| return waveform[:, begin:end], begin, end | |
| else: | |
| end = min(waveform.shape[0], end * self.hop_size) | |
| return waveform[begin:end], begin, end | |
| # @timeit | |
| def slice(self, waveform, return_chunks_positions=False): | |
| if len(waveform.shape) > 1: | |
| # (#channle, wave_len) -> (wave_len) | |
| samples = waveform.mean(axis=0) | |
| else: | |
| samples = waveform | |
| if samples.shape[0] <= self.min_length: | |
| return [waveform] | |
| rms_list = get_rms( | |
| y=samples, frame_length=self.win_size, hop_length=self.hop_size | |
| ).squeeze(0) | |
| sil_tags = [] | |
| silence_start = None | |
| clip_start = 0 | |
| for i, rms in enumerate(rms_list): | |
| # Keep looping while frame is silent. | |
| if rms < self.threshold: | |
| # Record start of silent frames. | |
| if silence_start is None: | |
| silence_start = i | |
| continue | |
| # Keep looping while frame is not silent and silence start has not been recorded. | |
| if silence_start is None: | |
| continue | |
| # Clear recorded silence start if interval is not enough or clip is too short | |
| is_leading_silence = silence_start == 0 and i > self.max_sil_kept | |
| need_slice_middle = ( | |
| i - silence_start >= self.min_interval | |
| and i - clip_start >= self.min_length | |
| ) | |
| if not is_leading_silence and not need_slice_middle: | |
| silence_start = None | |
| continue | |
| # Need slicing. Record the range of silent frames to be removed. | |
| if i - silence_start <= self.max_sil_kept: | |
| pos = rms_list[silence_start : i + 1].argmin() + silence_start | |
| if silence_start == 0: | |
| sil_tags.append((0, pos)) | |
| else: | |
| sil_tags.append((pos, pos)) | |
| clip_start = pos | |
| elif i - silence_start <= self.max_sil_kept * 2: | |
| pos = rms_list[ | |
| i - self.max_sil_kept : silence_start + self.max_sil_kept + 1 | |
| ].argmin() | |
| pos += i - self.max_sil_kept | |
| pos_l = ( | |
| rms_list[ | |
| silence_start : silence_start + self.max_sil_kept + 1 | |
| ].argmin() | |
| + silence_start | |
| ) | |
| pos_r = ( | |
| rms_list[i - self.max_sil_kept : i + 1].argmin() | |
| + i | |
| - self.max_sil_kept | |
| ) | |
| if silence_start == 0: | |
| sil_tags.append((0, pos_r)) | |
| clip_start = pos_r | |
| else: | |
| sil_tags.append((min(pos_l, pos), max(pos_r, pos))) | |
| clip_start = max(pos_r, pos) | |
| else: | |
| pos_l = ( | |
| rms_list[ | |
| silence_start : silence_start + self.max_sil_kept + 1 | |
| ].argmin() | |
| + silence_start | |
| ) | |
| pos_r = ( | |
| rms_list[i - self.max_sil_kept : i + 1].argmin() | |
| + i | |
| - self.max_sil_kept | |
| ) | |
| if silence_start == 0: | |
| sil_tags.append((0, pos_r)) | |
| else: | |
| sil_tags.append((pos_l, pos_r)) | |
| clip_start = pos_r | |
| silence_start = None | |
| # Deal with trailing silence. | |
| total_frames = rms_list.shape[0] | |
| if ( | |
| silence_start is not None | |
| and total_frames - silence_start >= self.min_interval | |
| ): | |
| silence_end = min(total_frames, silence_start + self.max_sil_kept) | |
| pos = rms_list[silence_start : silence_end + 1].argmin() + silence_start | |
| sil_tags.append((pos, total_frames + 1)) | |
| # Apply and return slices. | |
| if len(sil_tags) == 0: | |
| return [waveform] | |
| else: | |
| chunks = [] | |
| chunks_pos_of_waveform = [] | |
| if sil_tags[0][0] > 0: | |
| chunk, begin, end = self._apply_slice(waveform, 0, sil_tags[0][0]) | |
| chunks.append(chunk) | |
| chunks_pos_of_waveform.append((begin, end)) | |
| for i in range(len(sil_tags) - 1): | |
| chunk, begin, end = self._apply_slice( | |
| waveform, sil_tags[i][1], sil_tags[i + 1][0] | |
| ) | |
| chunks.append(chunk) | |
| chunks_pos_of_waveform.append((begin, end)) | |
| if sil_tags[-1][1] < total_frames: | |
| chunk, begin, end = self._apply_slice( | |
| waveform, sil_tags[-1][1], total_frames | |
| ) | |
| chunks.append(chunk) | |
| chunks_pos_of_waveform.append((begin, end)) | |
| return ( | |
| chunks | |
| if not return_chunks_positions | |
| else ( | |
| chunks, | |
| chunks_pos_of_waveform, | |
| ) | |
| ) | |
| def split_utterances_from_audio( | |
| wav_file, | |
| output_dir, | |
| max_duration_of_utterance=10.0, | |
| min_interval=300, | |
| db_threshold=-40, | |
| ): | |
| """ | |
| Split a long audio into utterances accoring to the silence (VAD). | |
| max_duration_of_utterance (second): | |
| The maximum duration of every utterance (seconds) | |
| min_interval (millisecond): | |
| The smaller min_interval is, the more sliced audio clips this script is likely to generate. | |
| """ | |
| print("File:", wav_file.split("/")[-1]) | |
| waveform, fs = torchaudio.load(wav_file) | |
| slicer = Slicer(sr=fs, min_interval=min_interval, threshold=db_threshold) | |
| chunks, positions = slicer.slice(waveform, return_chunks_positions=True) | |
| durations = [(end - begin) / fs for begin, end in positions] | |
| print( | |
| "Slicer's min silence part is {}ms, min and max duration of sliced utterances is {}s and {}s".format( | |
| min_interval, min(durations), max(durations) | |
| ) | |
| ) | |
| res_chunks, res_positions = [], [] | |
| for i, chunk in enumerate(chunks): | |
| if len(chunk.shape) == 1: | |
| chunk = chunk[None, :] | |
| begin, end = positions[i] | |
| assert end - begin == chunk.shape[-1] | |
| max_wav_len = max_duration_of_utterance * fs | |
| if chunk.shape[-1] <= max_wav_len: | |
| res_chunks.append(chunk) | |
| res_positions.append(positions[i]) | |
| else: | |
| # TODO: to reserve overlapping and conduct fade-in, fade-out | |
| # Get segments number | |
| number = 2 | |
| while chunk.shape[-1] // number >= max_wav_len: | |
| number += 1 | |
| seg_len = chunk.shape[-1] // number | |
| # Split | |
| for num in range(number): | |
| s = seg_len * num | |
| t = min(s + seg_len, chunk.shape[-1]) | |
| seg_begin = begin + s | |
| seg_end = begin + t | |
| res_chunks.append(chunk[:, s:t]) | |
| res_positions.append((seg_begin, seg_end)) | |
| # Save utterances | |
| os.makedirs(output_dir, exist_ok=True) | |
| res = {"fs": int(fs)} | |
| for i, chunk in enumerate(res_chunks): | |
| filename = "{:04d}.wav".format(i) | |
| res[filename] = [int(p) for p in res_positions[i]] | |
| save_audio(os.path.join(output_dir, filename), chunk, fs) | |
| # Save positions | |
| with open(os.path.join(output_dir, "positions.json"), "w") as f: | |
| json.dump(res, f, indent=4, ensure_ascii=False) | |
| return res | |
| def is_silence( | |
| wavform, | |
| fs, | |
| threshold=-40.0, | |
| min_interval=300, | |
| hop_size=10, | |
| min_length=5000, | |
| ): | |
| """ | |
| Detect whether the given wavform is a silence | |
| wavform: (T, ) | |
| """ | |
| threshold = 10 ** (threshold / 20.0) | |
| hop_size = round(fs * hop_size / 1000) | |
| win_size = min(round(min_interval), 4 * hop_size) | |
| min_length = round(fs * min_length / 1000 / hop_size) | |
| if wavform.shape[0] <= min_length: | |
| return True | |
| # (#Frame,) | |
| rms_array = get_rms(y=wavform, frame_length=win_size, hop_length=hop_size).squeeze( | |
| 0 | |
| ) | |
| return (rms_array < threshold).all() | |
| def split_audio( | |
| wav_file, target_sr, output_dir, max_duration_of_segment=10.0, overlap_duration=1.0 | |
| ): | |
| """ | |
| Split a long audio into segments. | |
| target_sr: | |
| The target sampling rate to save the segments. | |
| max_duration_of_utterance (second): | |
| The maximum duration of every utterance (second) | |
| overlap_duraion: | |
| Each segment has "overlap duration" (second) overlap with its previous and next segment | |
| """ | |
| # (#channel, T) -> (T,) | |
| waveform, fs = torchaudio.load(wav_file) | |
| waveform = torchaudio.functional.resample( | |
| waveform, orig_freq=fs, new_freq=target_sr | |
| ) | |
| waveform = torch.mean(waveform, dim=0) | |
| # waveform, _ = load_audio_torch(wav_file, target_sr) | |
| assert len(waveform.shape) == 1 | |
| assert overlap_duration < max_duration_of_segment | |
| length = int(max_duration_of_segment * target_sr) | |
| stride = int((max_duration_of_segment - overlap_duration) * target_sr) | |
| chunks = [] | |
| for i in range(0, len(waveform), stride): | |
| # (length,) | |
| chunks.append(waveform[i : i + length]) | |
| if i + length >= len(waveform): | |
| break | |
| # Save segments | |
| os.makedirs(output_dir, exist_ok=True) | |
| results = [] | |
| for i, chunk in enumerate(chunks): | |
| uid = "{:04d}".format(i) | |
| filename = os.path.join(output_dir, "{}.wav".format(uid)) | |
| results.append( | |
| {"Uid": uid, "Path": filename, "Duration": len(chunk) / target_sr} | |
| ) | |
| save_audio( | |
| filename, | |
| chunk, | |
| target_sr, | |
| turn_up=not is_silence(chunk, target_sr), | |
| add_silence=False, | |
| ) | |
| return results | |
| def merge_segments_torchaudio(wav_files, fs, output_path, overlap_duration=1.0): | |
| """Merge the given wav_files (may have overlaps) into a long audio | |
| fs: | |
| The sampling rate of the wav files. | |
| output_path: | |
| The output path to save the merged audio. | |
| overlap_duration (float, optional): | |
| Each segment has "overlap duration" (second) overlap with its previous and next segment. Defaults to 1.0. | |
| """ | |
| waveforms = [] | |
| for file in wav_files: | |
| # (T,) | |
| waveform, _ = load_audio_torch(file, fs) | |
| waveforms.append(waveform) | |
| if len(waveforms) == 1: | |
| save_audio(output_path, waveforms[0], fs, add_silence=False, turn_up=False) | |
| return | |
| overlap_len = int(overlap_duration * fs) | |
| fade_out = torchaudio.transforms.Fade(fade_out_len=overlap_len) | |
| fade_in = torchaudio.transforms.Fade(fade_in_len=overlap_len) | |
| fade_in_and_out = torchaudio.transforms.Fade(fade_out_len=overlap_len) | |
| segments_lens = [len(wav) for wav in waveforms] | |
| merged_waveform_len = sum(segments_lens) - overlap_len * (len(waveforms) - 1) | |
| merged_waveform = torch.zeros(merged_waveform_len) | |
| start = 0 | |
| for index, wav in enumerate( | |
| tqdm(waveforms, desc="Merge for {}".format(output_path)) | |
| ): | |
| wav_len = len(wav) | |
| if index == 0: | |
| wav = fade_out(wav) | |
| elif index == len(waveforms) - 1: | |
| wav = fade_in(wav) | |
| else: | |
| wav = fade_in_and_out(wav) | |
| merged_waveform[start : start + wav_len] = wav | |
| start += wav_len - overlap_len | |
| save_audio(output_path, merged_waveform, fs, add_silence=False, turn_up=True) | |
| def merge_segments_encodec(wav_files, fs, output_path, overlap_duration=1.0): | |
| """Merge the given wav_files (may have overlaps) into a long audio | |
| fs: | |
| The sampling rate of the wav files. | |
| output_path: | |
| The output path to save the merged audio. | |
| overlap_duration (float, optional): | |
| Each segment has "overlap duration" (second) overlap with its previous and next segment. Defaults to 1.0. | |
| """ | |
| waveforms = [] | |
| for file in wav_files: | |
| # (T,) | |
| waveform, _ = load_audio_torch(file, fs) | |
| waveforms.append(waveform) | |
| if len(waveforms) == 1: | |
| save_audio(output_path, waveforms[0], fs, add_silence=False, turn_up=False) | |
| return | |
| device = waveforms[0].device | |
| dtype = waveforms[0].dtype | |
| shape = waveforms[0].shape[:-1] | |
| overlap_len = int(overlap_duration * fs) | |
| segments_lens = [len(wav) for wav in waveforms] | |
| merged_waveform_len = sum(segments_lens) - overlap_len * (len(waveforms) - 1) | |
| sum_weight = torch.zeros(merged_waveform_len, device=device, dtype=dtype) | |
| out = torch.zeros(*shape, merged_waveform_len, device=device, dtype=dtype) | |
| offset = 0 | |
| for frame in waveforms: | |
| frame_length = frame.size(-1) | |
| t = torch.linspace(0, 1, frame_length + 2, device=device, dtype=torch.float32)[ | |
| 1:-1 | |
| ] | |
| weight = 0.5 - (t - 0.5).abs() | |
| weighted_frame = frame * weight | |
| cur = out[..., offset : offset + frame_length] | |
| cur += weighted_frame[..., : cur.size(-1)] | |
| out[..., offset : offset + frame_length] = cur | |
| cur = sum_weight[offset : offset + frame_length] | |
| cur += weight[..., : cur.size(-1)] | |
| sum_weight[offset : offset + frame_length] = cur | |
| offset += frame_length - overlap_len | |
| assert sum_weight.min() > 0 | |
| merged_waveform = out / sum_weight | |
| save_audio(output_path, merged_waveform, fs, add_silence=False, turn_up=True) | |