import os import gc import torchaudio import pandas from faster_whisper import WhisperModel from glob import glob from tqdm import tqdm from TTS.tts.layers.xtts.tokenizer import multilingual_cleaners # Add support for JA train # from utils.tokenizer import multilingual_cleaners import torch import torchaudio # torch.set_num_threads(1) torch.set_num_threads(16) import os audio_types = (".wav", ".mp3", ".flac") def find_latest_best_model(folder_path): search_path = os.path.join(folder_path, '**', 'best_model.pth') files = glob(search_path, recursive=True) latest_file = max(files, key=os.path.getctime, default=None) return latest_file def list_audios(basePath, contains=None): # return the set of files that are valid return list_files(basePath, validExts=audio_types, contains=contains) def list_files(basePath, validExts=None, contains=None): # loop over the directory structure for (rootDir, dirNames, filenames) in os.walk(basePath): # loop over the filenames in the current directory for filename in filenames: # if the contains string is not none and the filename does not contain # the supplied string, then ignore the file if contains is not None and filename.find(contains) == -1: continue # determine the file extension of the current file ext = filename[filename.rfind("."):].lower() # check to see if the file is an audio and should be processed if validExts is None or ext.endswith(validExts): # construct the path to the audio and yield it audioPath = os.path.join(rootDir, filename) yield audioPath def format_audio_list(audio_files, asr_model, target_language="en", out_path=None, buffer=0.2, eval_percentage=0.15, speaker_name="coqui", gradio_progress=None): audio_total_size = 0 os.makedirs(out_path, exist_ok=True) lang_file_path = os.path.join(out_path, "lang.txt") current_language = None if os.path.exists(lang_file_path): with open(lang_file_path, 'r', encoding='utf-8') as existing_lang_file: current_language = existing_lang_file.read().strip() if current_language != target_language: with open(lang_file_path, 'w', encoding='utf-8') as lang_file: lang_file.write(target_language + '\n') print("Warning, existing language does not match target language. Updated lang.txt with target language.") else: print("Existing language matches target language") metadata = {"audio_file": [], "text": [], "speaker_name": []} train_metadata_path = os.path.join(out_path, "metadata_train.csv") eval_metadata_path = os.path.join(out_path, "metadata_eval.csv") existing_metadata = {'train': None, 'eval': None} if os.path.exists(train_metadata_path): existing_metadata['train'] = pandas.read_csv(train_metadata_path, sep="|") print("Existing training metadata found and loaded.") if os.path.exists(eval_metadata_path): existing_metadata['eval'] = pandas.read_csv(eval_metadata_path, sep="|") print("Existing evaluation metadata found and loaded.") if gradio_progress is not None: tqdm_object = gradio_progress.tqdm(audio_files, desc="Formatting...") else: tqdm_object = tqdm(audio_files) for audio_path in tqdm_object: audio_file_name_without_ext, _= os.path.splitext(os.path.basename(audio_path)) prefix_check = f"wavs/{audio_file_name_without_ext}_" skip_processing = False for key in ['train', 'eval']: if existing_metadata[key] is not None: mask = existing_metadata[key]['audio_file'].str.startswith(prefix_check) if mask.any(): print(f"Segments from {audio_file_name_without_ext} have been previously processed; skipping...") skip_processing = True break if skip_processing: continue wav, sr = torchaudio.load(audio_path) if wav.size(0) != 1: wav = torch.mean(wav, dim=0, keepdim=True) wav = wav.squeeze() audio_total_size += (wav.size(-1) / sr) segments, _= asr_model.transcribe(audio_path, vad_filter=True, word_timestamps=True, language=target_language) segments = list(segments) i = 0 sentence = "" sentence_start = None first_word = True words_list = [] for _, segment in enumerate(segments): words = list(segment.words) words_list.extend(words) for word_idx, word in enumerate(words_list): if first_word: sentence_start = word.start if word_idx == 0: sentence_start = max(sentence_start - buffer, 0) else: previous_word_end = words_list[word_idx - 1].end sentence_start = max(sentence_start - buffer, (previous_word_end + sentence_start) / 2) sentence = word.word first_word = False else: sentence += word.word if word.word[-1] in ["!", "。", ".", "?"]: sentence = sentence[1:] sentence = multilingual_cleaners(sentence, target_language) audio_file_name, _= os.path.splitext(os.path.basename(audio_path)) audio_file = f"wavs/{audio_file_name}_{str(i).zfill(8)}.wav" if word_idx + 1 < len(words_list): next_word_start = words_list[word_idx + 1].start else: next_word_start = (wav.shape[0] - 1) / sr word_end = min((word.end + next_word_start) / 2, word.end + buffer) absolute_path = os.path.join(out_path, audio_file) os.makedirs(os.path.dirname(absolute_path), exist_ok=True) i += 1 first_word = True audio = wav[int(sr*sentence_start):int(sr *word_end)].unsqueeze(0) if audio.size(-1) >= sr / 3: torchaudio.save(absolute_path, audio, sr) else: continue metadata["audio_file"].append(audio_file) metadata["text"].append(sentence) metadata["speaker_name"].append(speaker_name) df = pandas.DataFrame(metadata) mode = 'w' if not os.path.exists(train_metadata_path) else 'a' header = not os.path.exists(train_metadata_path) df.to_csv(train_metadata_path, sep="|", index=False, mode=mode, header=header) mode = 'w' if not os.path.exists(eval_metadata_path) else 'a' header = not os.path.exists(eval_metadata_path) df.to_csv(eval_metadata_path, sep="|", index=False, mode=mode, header=header) metadata = {"audio_file": [], "text": [], "speaker_name": []} if os.path.exists(train_metadata_path) and os.path.exists(eval_metadata_path): existing_train_df = existing_metadata['train'] existing_eval_df = existing_metadata['eval'] else: existing_train_df = pandas.DataFrame(columns=["audio_file", "text", "speaker_name"]) existing_eval_df = pandas.DataFrame(columns=["audio_file", "text", "speaker_name"]) new_data_df = pandas.read_csv(train_metadata_path, sep="|") combined_train_df = pandas.concat([existing_train_df, new_data_df], ignore_index=True).drop_duplicates().reset_index(drop=True) combined_eval_df = pandas.concat([existing_eval_df, new_data_df], ignore_index=True).drop_duplicates().reset_index(drop=True) combined_train_df_shuffled = combined_train_df.sample(frac=1) num_val_samples = int(len(combined_train_df_shuffled)* eval_percentage) final_eval_set = combined_train_df_shuffled[:num_val_samples] final_training_set = combined_train_df_shuffled[num_val_samples:] final_training_set.sort_values('audio_file').to_csv(train_metadata_path, sep='|', index=False) final_eval_set.sort_values('audio_file').to_csv(eval_metadata_path, sep='|', index=False) return train_metadata_path, eval_metadata_path, audio_total_size