diff --git a/tools/my_utils.py b/tools/my_utils.py new file mode 100644 index 0000000..776939d --- /dev/null +++ b/tools/my_utils.py @@ -0,0 +1,21 @@ +import ffmpeg +import numpy as np + + +def load_audio(file, sr): + try: + # https://github.com/openai/whisper/blob/main/whisper/audio.py#L26 + # This launches a subprocess to decode audio while down-mixing and resampling as necessary. + # Requires the ffmpeg CLI and `ffmpeg-python` package to be installed. + file = ( + file.strip(" ").strip('"').strip("\n").strip('"').strip(" ") + ) # 防止小白拷路径头尾带了空格和"和回车 + out, _ = ( + ffmpeg.input(file, threads=0) + .output("-", format="f32le", acodec="pcm_f32le", ac=1, ar=sr) + .run(cmd=["ffmpeg", "-nostdin"], capture_stdout=True, capture_stderr=True) + ) + except Exception as e: + raise RuntimeError(f"Failed to load audio: {e}") + + return np.frombuffer(out, np.float32).flatten() diff --git a/tools/slice_audio.py b/tools/slice_audio.py new file mode 100644 index 0000000..79592de --- /dev/null +++ b/tools/slice_audio.py @@ -0,0 +1,48 @@ +import os,sys,numpy as np +import traceback +from scipy.io import wavfile +# parent_directory = os.path.dirname(os.path.abspath(__file__)) +# sys.path.append(parent_directory) +from my_utils import load_audio +from slicer2 import Slicer + +def slice(inp,opt_root,threshold,min_length,min_interval,hop_size,max_sil_kept,_max,alpha,i_part,all_part): + os.makedirs(opt_root,exist_ok=True) + if os.path.isfile(inp): + input=[inp] + elif os.path.isdir(inp): + input=["%s/%s"%(inp,name)for name in sorted(list(os.listdir(inp)))] + else: + return "输入路径存在但既不是文件也不是文件夹" + slicer = Slicer( + sr=32000, # 长音频采样率 + threshold= int(threshold), # 音量小于这个值视作静音的备选切割点 + min_length= int(min_length), # 每段最小多长,如果第一段太短一直和后面段连起来直到超过这个值 + min_interval= int(min_interval), # 最短切割间隔 + hop_size= int(hop_size), # 怎么算音量曲线,越小精度越大计算量越高(不是精度越大效果越好) + max_sil_kept= int(max_sil_kept), # 切完后静音最多留多长 + ) + _max=float(_max) + alpha=float(alpha) + for inp_path in input[int(i_part)::int(all_part)]: + # print(inp_path) + try: + name = os.path.basename(inp_path) + audio = load_audio(inp_path, 32000) + # print(audio.shape) + for chunk, start, end in slicer.slice(audio): # start和end是帧数 + tmp_max = np.abs(chunk).max() + if(tmp_max>1):chunk/=tmp_max + chunk = (chunk / tmp_max * (_max * alpha)) + (1 - alpha) * chunk + wavfile.write( + "%s/%s_%s_%s.wav" % (opt_root, name, start, end), + 32000, + # chunk.astype(np.float32), + (chunk * 32767).astype(np.int16), + ) + except: + print(inp_path,"->fail->",traceback.format_exc()) + return "执行完毕,请检查输出文件" + +print(slice(*sys.argv[1:])) + diff --git a/tools/slicer2.py b/tools/slicer2.py new file mode 100644 index 0000000..ba6794b --- /dev/null +++ b/tools/slicer2.py @@ -0,0 +1,261 @@ +import numpy as np + + +# 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: + def __init__( + self, + sr: int, + threshold: float = -40.0, + min_length: int = 5000, + min_interval: int = 300, + hop_size: int = 20, + 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): + if len(waveform.shape) > 1: + return waveform[ + :, begin * self.hop_size : min(waveform.shape[1], end * self.hop_size) + ] + else: + return waveform[ + begin * self.hop_size : min(waveform.shape[0], end * self.hop_size) + ] + + # @timeit + def slice(self, waveform): + if len(waveform.shape) > 1: + 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,0,int(total_frames*self.hop_size)]] + else: + chunks = [] + if sil_tags[0][0] > 0: + chunks.append([self._apply_slice(waveform, 0, sil_tags[0][0]),0,int(sil_tags[0][0]*self.hop_size)]) + for i in range(len(sil_tags) - 1): + chunks.append( + [self._apply_slice(waveform, sil_tags[i][1], sil_tags[i + 1][0]),int(sil_tags[i][1]*self.hop_size),int(sil_tags[i + 1][0]*self.hop_size)] + ) + if sil_tags[-1][1] < total_frames: + chunks.append( + [self._apply_slice(waveform, sil_tags[-1][1], total_frames),int(sil_tags[-1][1]*self.hop_size),int(total_frames*self.hop_size)] + ) + return chunks + + +def main(): + import os.path + from argparse import ArgumentParser + + import librosa + import soundfile + + parser = ArgumentParser() + parser.add_argument("audio", type=str, help="The audio to be sliced") + parser.add_argument( + "--out", type=str, help="Output directory of the sliced audio clips" + ) + parser.add_argument( + "--db_thresh", + type=float, + required=False, + default=-40, + help="The dB threshold for silence detection", + ) + parser.add_argument( + "--min_length", + type=int, + required=False, + default=5000, + help="The minimum milliseconds required for each sliced audio clip", + ) + parser.add_argument( + "--min_interval", + type=int, + required=False, + default=300, + help="The minimum milliseconds for a silence part to be sliced", + ) + parser.add_argument( + "--hop_size", + type=int, + required=False, + default=10, + help="Frame length in milliseconds", + ) + parser.add_argument( + "--max_sil_kept", + type=int, + required=False, + default=500, + help="The maximum silence length kept around the sliced clip, presented in milliseconds", + ) + args = parser.parse_args() + out = args.out + if out is None: + out = os.path.dirname(os.path.abspath(args.audio)) + audio, sr = librosa.load(args.audio, sr=None, mono=False) + slicer = Slicer( + sr=sr, + threshold=args.db_thresh, + min_length=args.min_length, + min_interval=args.min_interval, + hop_size=args.hop_size, + max_sil_kept=args.max_sil_kept, + ) + chunks = slicer.slice(audio) + if not os.path.exists(out): + os.makedirs(out) + for i, chunk in enumerate(chunks): + if len(chunk.shape) > 1: + chunk = chunk.T + soundfile.write( + os.path.join( + out, + f"%s_%d.wav" + % (os.path.basename(args.audio).rsplit(".", maxsplit=1)[0], i), + ), + chunk, + sr, + ) + + +if __name__ == "__main__": + main() diff --git a/tools/subfix_webui.py b/tools/subfix_webui.py new file mode 100644 index 0000000..0e6585e --- /dev/null +++ b/tools/subfix_webui.py @@ -0,0 +1,492 @@ +import argparse,os +import copy +import json +import os +import uuid + +import librosa +import gradio as gr +import numpy as np +import soundfile + +g_json_key_text = "" +g_json_key_path = "" +g_load_file = "" +g_load_format = "" + +g_max_json_index = 0 +g_index = 0 +g_batch = 10 +g_text_list = [] +g_audio_list = [] +g_checkbox_list = [] +g_data_json = [] + + +def reload_data(index, batch): + global g_index + g_index = index + global g_batch + g_batch = batch + datas = g_data_json[index:index+batch] + output = [] + for d in datas: + output.append( + { + g_json_key_text: d[g_json_key_text], + g_json_key_path: d[g_json_key_path] + } + ) + return output + + +def b_change_index(index, batch): + global g_index, g_batch + g_index, g_batch = index, batch + datas = reload_data(index, batch) + output = [] + for i , _ in enumerate(datas): + output.append( + # gr.Textbox( + # label=f"Text {i+index}", + # value=_[g_json_key_text]#text + # ) + { + "__type__":"update", + "label":f"Text {i+index}", + "value":_[g_json_key_text] + } + ) + for _ in range(g_batch - len(datas)): + output.append( + # gr.Textbox( + # label=f"Text", + # value="" + # ) + { + "__type__": "update", + "label": f"Text", + "value": "" + } + ) + for _ in datas: + output.append(_[g_json_key_path]) + for _ in range(g_batch - len(datas)): + output.append(None) + for _ in range(g_batch): + output.append(False) + return output + + +def b_next_index(index, batch): + if (index + batch) <= g_max_json_index: + return index + batch , *b_change_index(index + batch, batch) + else: + return index, *b_change_index(index, batch) + + +def b_previous_index(index, batch): + if (index - batch) >= 0: + return index - batch , *b_change_index(index - batch, batch) + else: + return 0, *b_change_index(0, batch) + + +def b_submit_change(*text_list): + global g_data_json + change = False + for i, new_text in enumerate(text_list): + if g_index + i <= g_max_json_index: + new_text = new_text.strip()+' ' + if (g_data_json[g_index + i][g_json_key_text] != new_text): + g_data_json[g_index + i][g_json_key_text] = new_text + change = True + if change: + b_save_file() + return g_index, *b_change_index(g_index, g_batch) + + +def b_delete_audio(*checkbox_list): + global g_data_json, g_index, g_max_json_index + change = False + for i, checkbox in reversed(list(enumerate(checkbox_list))): + if g_index + i < len(g_data_json): + if (checkbox == True): + g_data_json.pop(g_index + i) + change = True + + g_max_json_index = len(g_data_json)-1 + if g_index > g_max_json_index: + g_index = g_max_json_index + g_index = g_index if g_index >= 0 else 0 + # if change: + # b_save_file() + # return gr.Slider(value=g_index, maximum=(g_max_json_index if g_max_json_index>=0 else 0)), *b_change_index(g_index, g_batch) + return {"value":g_index,"__type__":"update","maximum":(g_max_json_index if g_max_json_index>=0 else 0)},*b_change_index(g_index, g_batch) + + +def b_invert_selection(*checkbox_list): + new_list = [not item if item is True else True for item in checkbox_list] + return new_list + + +def get_next_path(filename): + base_dir = os.path.dirname(filename) + base_name = os.path.splitext(os.path.basename(filename))[0] + for i in range(100): + new_path = os.path.join(base_dir, f"{base_name}_{str(i).zfill(2)}.wav") + if not os.path.exists(new_path) : + return new_path + return os.path.join(base_dir, f'{str(uuid.uuid4())}.wav') + + +def b_audio_split(audio_breakpoint, *checkbox_list): + global g_data_json , g_max_json_index + checked_index = [] + for i, checkbox in enumerate(checkbox_list): + if (checkbox == True and g_index+i < len(g_data_json)): + checked_index.append(g_index + i) + if len(checked_index) == 1 : + index = checked_index[0] + audio_json = copy.deepcopy(g_data_json[index]) + path = audio_json[g_json_key_path] + data, sample_rate = librosa.load(path, sr=None, mono=True) + audio_maxframe = len(data) + break_frame = int(audio_breakpoint * sample_rate) + + if (break_frame >= 1 and break_frame < audio_maxframe): + audio_first = data[0:break_frame] + audio_second = data[break_frame:] + nextpath = get_next_path(path) + soundfile.write(nextpath, audio_second, sample_rate) + soundfile.write(path, audio_first, sample_rate) + g_data_json.insert(index + 1, audio_json) + g_data_json[index + 1][g_json_key_path] = nextpath + b_save_file() + + g_max_json_index = len(g_data_json) - 1 + # return gr.Slider(value=g_index, maximum=g_max_json_index), *b_change_index(g_index, g_batch) + return {"value":g_index,"maximum":g_max_json_index,"__type__":"update"}, *b_change_index(g_index, g_batch) + +def b_merge_audio(interval_r, *checkbox_list): + global g_data_json , g_max_json_index + checked_index = [] + audios_path = [] + audios_text = [] + for i, checkbox in enumerate(checkbox_list): + if (checkbox == True and g_index+i < len(g_data_json)): + checked_index.append(g_index + i) + + if (len(checked_index)>1): + for i in checked_index: + audios_path.append(g_data_json[i][g_json_key_path]) + audios_text.append(g_data_json[i][g_json_key_text]) + for i in reversed(checked_index[1:]): + g_data_json.pop(i) + + base_index = checked_index[0] + base_path = audios_path[0] + g_data_json[base_index][g_json_key_text] = "".join(audios_text) + + audio_list = [] + l_sample_rate = None + for i, path in enumerate(audios_path): + data, sample_rate = librosa.load(path, sr=l_sample_rate, mono=True) + l_sample_rate = sample_rate + if (i > 0): + silence = np.zeros(int(l_sample_rate * interval_r)) + audio_list.append(silence) + + audio_list.append(data) + + audio_concat = np.concatenate(audio_list) + + soundfile.write(base_path, audio_concat, l_sample_rate) + + b_save_file() + + g_max_json_index = len(g_data_json) - 1 + + # return gr.Slider(value=g_index, maximum=g_max_json_index), *b_change_index(g_index, g_batch) + return {"value":g_index,"maximum":g_max_json_index,"__type__":"update"}, *b_change_index(g_index, g_batch) + + +def b_save_json(): + with open(g_load_file,'w', encoding="utf-8") as file: + for data in g_data_json: + file.write(f'{json.dumps(data, ensure_ascii = False)}\n') + + +def b_save_list(): + with open(g_load_file,'w', encoding="utf-8") as file: + for data in g_data_json: + wav_path = data["wav_path"] + speaker_name = data["speaker_name"] + language = data["language"] + text = data["text"] + file.write(f"{wav_path}|{speaker_name}|{language}|{text}".strip()+'\n') + + +def b_load_json(): + global g_data_json, g_max_json_index + with open(g_load_file, 'r', encoding="utf-8") as file: + g_data_json = file.readlines() + g_data_json = [json.loads(line) for line in g_data_json] + g_max_json_index = len(g_data_json) - 1 + + +def b_load_list(): + global g_data_json, g_max_json_index + with open(g_load_file, 'r', encoding="utf-8") as source: + data_list = source.readlines() + for _ in data_list: + data = _.split('|') + if (len(data) == 4): + wav_path, speaker_name, language, text = data + g_data_json.append( + { + 'wav_path':wav_path, + 'speaker_name':speaker_name, + 'language':language, + 'text':text.strip() + } + ) + else: + print("error line:", data) + g_max_json_index = len(g_data_json) - 1 + + +def b_save_file(): + if g_load_format == "json": + b_save_json() + elif g_load_format == "list": + b_save_list() + + +def b_load_file(): + if g_load_format == "json": + b_load_json() + elif g_load_format == "list": + b_load_list() + + +def set_global(load_json, load_list, json_key_text, json_key_path, batch): + global g_json_key_text, g_json_key_path, g_load_file, g_load_format, g_batch + + g_batch = int(batch) + + if (load_json != "None"): + g_load_format = "json" + g_load_file = load_json + elif (load_list != "None"): + g_load_format = "list" + g_load_file = load_list + else: + g_load_format = "list" + g_load_file = "demo.list" + + g_json_key_text = json_key_text + g_json_key_path = json_key_path + + b_load_file() + + +if __name__ == "__main__": + parser = argparse.ArgumentParser(description='Process some integers.') + parser.add_argument('--load_json', default="None", help='source file, like demo.json') + parser.add_argument('--load_list', default="None", help='source file, like demo.list') + parser.add_argument('--webui_port_subfix', default=9871, help='source file, like demo.list') + parser.add_argument('--json_key_text', default="text", help='the text key name in json, Default: text') + parser.add_argument('--json_key_path', default="wav_path", help='the path key name in json, Default: wav_path') + parser.add_argument('--g_batch', default=10, help='max number g_batch wav to display, Default: 10') + + args = parser.parse_args() + + set_global(args.load_json, args.load_list, args.json_key_text, args.json_key_path, args.g_batch) + + with gr.Blocks() as demo: + + with gr.Row(): + btn_change_index = gr.Button("Change Index") + btn_submit_change = gr.Button("Submit Text") + btn_merge_audio = gr.Button("Merge Audio") + btn_delete_audio = gr.Button("Delete Audio") + btn_previous_index = gr.Button("Previous Index") + btn_next_index = gr.Button("Next Index") + + with gr.Row(): + index_slider = gr.Slider( + minimum=0, maximum=g_max_json_index, value=g_index, step=1, label="Index", scale=3 + ) + splitpoint_slider = gr.Slider( + minimum=0, maximum=120.0, value=0, step=0.1, label="Audio Split Point(s)", scale=3 + ) + btn_audio_split = gr.Button("Split Audio", scale=1) + btn_save_json = gr.Button("Save File", visible=True, scale=1) + btn_invert_selection = gr.Button("Invert Selection", scale=1) + + with gr.Row(): + with gr.Column(): + for _ in range(0,g_batch): + with gr.Row(): + text = gr.Textbox( + label = "Text", + visible = True, + scale=5 + ) + audio_output = gr.Audio( + label="Output Audio", + visible = True, + scale=5 + ) + audio_check = gr.Checkbox( + label="Yes", + show_label = True, + info = "Choose Audio", + scale=1 + ) + g_text_list.append(text) + g_audio_list.append(audio_output) + g_checkbox_list.append(audio_check) + + + + with gr.Row(): + batchsize_slider = gr.Slider( + minimum=1, maximum=g_batch, value=g_batch, step=1, label="Batch Size", scale=3, interactive=False + ) + interval_slider = gr.Slider( + minimum=0, maximum=2, value=0, step=0.01, label="Interval", scale=3 + ) + btn_theme_dark = gr.Button("Light Theme", link="?__theme=light", scale=1) + btn_theme_light = gr.Button("Dark Theme", link="?__theme=dark", scale=1) + + btn_change_index.click( + b_change_index, + inputs=[ + index_slider, + batchsize_slider, + ], + outputs=[ + *g_text_list, + *g_audio_list, + *g_checkbox_list + ], + ) + + + btn_submit_change.click( + b_submit_change, + inputs=[ + *g_text_list, + ], + outputs=[ + index_slider, + *g_text_list, + *g_audio_list, + *g_checkbox_list + ], + ) + + btn_previous_index.click( + b_previous_index, + inputs=[ + index_slider, + batchsize_slider, + ], + outputs=[ + index_slider, + *g_text_list, + *g_audio_list, + *g_checkbox_list + ], + ) + + btn_next_index.click( + b_next_index, + inputs=[ + index_slider, + batchsize_slider, + ], + outputs=[ + index_slider, + *g_text_list, + *g_audio_list, + *g_checkbox_list + ], + ) + + btn_delete_audio.click( + b_delete_audio, + inputs=[ + *g_checkbox_list + ], + outputs=[ + index_slider, + *g_text_list, + *g_audio_list, + *g_checkbox_list + ] + ) + + btn_merge_audio.click( + b_merge_audio, + inputs=[ + interval_slider, + *g_checkbox_list + ], + outputs=[ + index_slider, + *g_text_list, + *g_audio_list, + *g_checkbox_list + ] + ) + + btn_audio_split.click( + b_audio_split, + inputs=[ + splitpoint_slider, + *g_checkbox_list + ], + outputs=[ + index_slider, + *g_text_list, + *g_audio_list, + *g_checkbox_list + ] + ) + + btn_invert_selection.click( + b_invert_selection, + inputs=[ + *g_checkbox_list + ], + outputs=[ + *g_checkbox_list + ] + ) + + btn_save_json.click( + b_save_file + ) + + demo.load( + b_change_index, + inputs=[ + index_slider, + batchsize_slider, + ], + outputs=[ + *g_text_list, + *g_audio_list, + *g_checkbox_list + ], + ) + + demo.launch( + server_name="0.0.0.0", + inbrowser=True, + quiet=True, + server_port=int(args.webui_port_subfix) + ) \ No newline at end of file