Upload 4 files

consts.py

@@ -0,0 +1,3 @@
+port = 6006
+base_model_id = 'damo/speech_personal_sambert-hifigan_nsf_tts_zh-cn_pretrain_16k'
+base_model_version = 'v1.0.6'

utils_base.py

@@ -0,0 +1,56 @@
+# 基础方法封装
+import os
+import shutil
+import glob
+
+# 绝对路径获取方法
+curPath = os.path.dirname(os.path.abspath(__file__))
+def getAbsPath (relativePath):
+  joinPath = os.path.join(curPath, relativePath)
+  return os.path.normpath(
+    os.path.abspath(joinPath)
+  )
+
+# 数据集存放路径
+datasets_dir = getAbsPath('./datasets')
+if not shutil.os.path.exists(datasets_dir):
+  shutil.os.makedirs(datasets_dir)
+
+# 获取数据集列表 ----------------------------------------------------
+def get_dataset_list():
+  contents = os.listdir(datasets_dir)
+  sub_dirs = [
+    content
+    for content in contents
+    if os.path.isdir(os.path.join(datasets_dir, content))
+  ]
+  return sub_dirs
+
+# 小模型存放路径
+models_dir = getAbsPath('./models')
+if not shutil.os.path.exists(models_dir):
+  shutil.os.makedirs(models_dir)
+
+# 获取模型列表 ----------------------------------------------------
+def get_model_list():
+  contents = os.listdir(models_dir)
+  sub_dirs = [
+    content
+    for content in contents
+    if os.path.isdir(os.path.join(models_dir, content))
+  ]
+  return sub_dirs
+
+# 确保对应的空目录存在
+def ensure_empty_dir(dirpath):
+  if shutil.os.path.exists(dirpath):
+    shutil.rmtree(dirpath)
+  shutil.os.makedirs(dirpath)
+
+# 获取目录中的最后一个文件
+def get_last_file(dirpath):
+  files = glob.glob(os.path.join(dirpath, '*'))
+  sorted_files = sorted(files, key=os.path.basename)
+  if sorted_files:
+    return sorted_files[-1]
+  return False

utils_label.py

@@ -0,0 +1,99 @@
+# 对 sambert 训练的数据标注处理
+import os
+import shutil
+import uuid
+import librosa
+import gradio as gr
+from scipy.io import wavfile
+import numpy as np
+import whisper
+from modelscope.tools import run_auto_label
+from utils_base import ensure_empty_dir, datasets_dir, get_dataset_list
+
+# 绝对路径获取方法
+curPath = os.path.dirname(os.path.abspath(__file__))
+def getAbsPath (relativePath):
+  joinPath = os.path.join(curPath, relativePath)
+  return os.path.normpath(
+    os.path.abspath(joinPath)
+  )
+
+# 初始化 whisper 模型的加载
+model_path = getAbsPath('../../models/whisper/medium.pt')
+whisper_model = None
+if shutil.os.path.exists(model_path):
+  whisper_model = whisper.load_model(model_path)
+else:
+  whisper_model = whisper.load_model('medium')
+
+# whisper 音频分割方法 ----------------------------------------------
+def split_long_audio(model, filepaths, save_path, out_sr=44100):
+  # 格式化输入的音频路径(兼容单个音频和多个音频)
+  if isinstance(filepaths, str):
+    filepaths = [filepaths]
+
+  # 对音频依次做拆分并存放到临时路径
+  for file_idx, filepath in enumerate(filepaths):
+    print(f"Transcribing file {file_idx}: '{filepath}' to segments...")
+    result = model.transcribe(filepath, word_timestamps=True, task="transcribe", beam_size=5, best_of=5)
+    segments = result['segments']
+
+    # 采用 librosa 配合 scipy 做音频数据分割
+    wav, sr = librosa.load(filepath, sr=None, offset=0, duration=None, mono=True)
+    wav, _ = librosa.effects.trim(wav, top_db=20)
+    peak = np.abs(wav).max()
+    if peak > 1.0:
+      wav = 0.98 * wav / peak
+    wav2 = librosa.resample(wav, orig_sr=sr, target_sr=out_sr)
+    wav2 /= max(wav2.max(), -wav2.min())
+
+    # 将长音频文件分割成一条条的短音频并放入指定的目录
+    for i, seg in enumerate(segments):
+      start_time = seg['start']
+      end_time = seg['end']
+      wav_seg = wav2[int(start_time * out_sr):int(end_time * out_sr)]
+      wav_seg_name = f"{file_idx}_{i}.wav"
+      out_fpath = os.path.join(save_path, wav_seg_name)
+      wavfile.write(out_fpath, rate=out_sr, data=(wav_seg * np.iinfo(np.int16).max).astype(np.int16))
+
+# 自动标注与标注后的文件打包 --------------------------------------------
+def auto_label(audio, name):
+  if not audio or not name:
+    return '', gr.update(choices=get_dataset_list())
+
+  # 创建临时目录用于存放分割后的音频与再次标注的信息
+  input_wav = getAbsPath(f'./temp/input-{ uuid.uuid4() }')
+  ensure_empty_dir(input_wav)
+
+  work_dir = os.path.join(datasets_dir, name)
+  ensure_empty_dir(work_dir)
+
+  # 音频分割
+  split_long_audio(whisper_model, audio, input_wav)
+
+  # 音频自动标注
+  # 第一次会自动下载对应的模型
+  run_auto_label(
+    input_wav=input_wav,
+    work_dir=work_dir,
+    resource_revision='v1.0.7'
+  )
+
+  # 移除目录
+  shutil.rmtree(input_wav)
+
+  # 返回结果
+  return '打标成功', gr.update(choices=get_dataset_list())
+  
+# 删除数据集 ----------------------------------------------------
+# name - 删除的数据集名称
+def delete_dataset(name):
+  try:
+    if not name:
+      return gr.update(choices=get_dataset_list())
+
+    target_dir = os.path.join(datasets_dir, name)
+    shutil.rmtree(target_dir)
+    return gr.update(choices=get_dataset_list(), value=None)
+  except Exception:
+    return gr.update(choices=get_dataset_list(), value=None)

utils_sambert.py

@@ -0,0 +1,141 @@
+# 训练部分实现
+import os
+import shutil
+import uuid
+import gradio as gr
+from modelscope.models.audio.tts import SambertHifigan
+from modelscope.pipelines import pipeline
+from modelscope.utils.constant import Tasks
+
+from modelscope.metainfo import Trainers
+from modelscope.trainers import build_trainer
+from modelscope.utils.audio.audio_utils import TtsTrainType
+from modelscope.hub.utils.utils import get_cache_dir
+
+from utils_base import ensure_empty_dir, get_last_file, models_dir, get_model_list
+import consts
+
+# 绝对路径获取方法
+curPath = os.path.dirname(os.path.abspath(__file__))
+def getAbsPath (relativePath):
+  joinPath = os.path.join(curPath, relativePath)
+  return os.path.normpath(
+    os.path.abspath(joinPath)
+  )
+
+# 模型训练 ---------------------------------------------------------
+# name      - 训练结果(小模型)命名
+# steps     - 训练步数
+# train_dataset_zip - 数据集zip包路径
+def train(name, steps, train_dataset_name):
+  # 创建临时目录用于放置 训练结果
+  work_dir = getAbsPath(f'./temp/work-{ uuid.uuid4() }')
+  ensure_empty_dir(work_dir)
+
+  # 数据集目录
+  train_dataset = getAbsPath(f'./datasets/{ train_dataset_name }')
+
+  # 进行训练
+  trainer = build_trainer(
+    Trainers.speech_kantts_trainer,
+    default_args=dict(
+      # 指定要finetune的 模型/版本
+      model = consts.base_model_id,
+      model_revision = consts.base_model_version,
+
+      work_dir = work_dir,            # 指定临时工作目录
+      train_dataset = train_dataset,  # 数据集目录
+
+      # 训练参数
+      train_type = {
+        TtsTrainType.TRAIN_TYPE_SAMBERT: {  # 配置训练AM（sambert）模型
+          'train_steps': steps + 1,        # 训练多少个step
+          'save_interval_steps': 20,       # 每训练多少个step保存一次checkpoint
+          'log_interval': 10               # 每训练多少个step打印一次训练日志
+        }
+      }
+    )
+  )
+  trainer.train()
+
+  # 挑选需要的文件到结果目录
+  target_dir = os.path.join(models_dir, name)
+  ensure_empty_dir(target_dir)
+  shutil.os.makedirs(os.path.join(target_dir, 'tmp_am', 'ckpt'))
+  shutil.os.makedirs(os.path.join(target_dir, 'data', 'se'))
+
+  shutil.copy(
+    get_last_file(os.path.join(work_dir, 'tmp_am', 'ckpt')),
+    os.path.join(target_dir, 'tmp_am', 'ckpt')
+  )
+  shutil.copy(
+    os.path.join(work_dir, 'tmp_am', 'config.yaml'),
+    os.path.join(target_dir, 'tmp_am'),
+  )
+  shutil.copy(
+    os.path.join(work_dir, 'data', 'audio_config.yaml'),
+    os.path.join(target_dir, 'data'),
+  )
+  shutil.copy(
+    os.path.join(work_dir, 'data', 'se', 'se.npy'),
+    os.path.join(target_dir, 'data', 'se'),
+  )
+
+  # 清理文件
+  shutil.rmtree(work_dir)
+  shutil.rmtree(train_dataset)
+
+  # 返回结果
+  return '训练完成', gr.update(choices=get_model_list())
+
+# 模型推理 ---------------------------------------------------------
+# name - 使用的小模型名称
+# txt - 需要合成音频的文字
+def infer(name, txt):
+  try:
+    base_model_path = os.path.join(get_cache_dir(), consts.base_model_id)
+    model_path = os.path.join(models_dir, name)
+    custom_infer_abs = {
+      'voice_name': 'F7',
+
+      # 小模型部分
+      'am_ckpt': os.path.join(model_path, 'tmp_am', 'ckpt'),
+      'am_config': os.path.join(model_path, 'tmp_am', 'config.yaml'),
+      'audio_config': os.path.join(model_path, 'data', 'audio_config.yaml'),
+      'se_file': os.path.join(model_path, 'data', 'se', 'se.npy'),
+
+      # 基础模型部分
+      'voc_ckpt': os.path.join(
+        base_model_path, 'basemodel_16k', 'hifigan', 'ckpt'
+      ),
+      'voc_config': os.path.join(
+        base_model_path, 'basemodel_16k', 'hifigan', 'config.yaml'
+      )
+    }
+
+    model = SambertHifigan(
+      base_model_path,
+      **{ 'custom_ckpt': custom_infer_abs }
+    )
+    inference = pipeline(task=Tasks.text_to_speech, model=model)
+    output = inference(input=txt)
+
+    output_path = f'/tmp/{ uuid.uuid4() }.wav'
+    with open(output_path, mode='bx') as f:
+      f.write(output['output_wav'])
+    return output_path
+  except Exception:
+    return False
+
+# 删除模型 ---------------------------------------------------------
+# name - 删除的小模型名称
+def delete_model(name):
+  try:
+    if not name:
+      return gr.update(choices=get_model_list())
+
+    target_dir = os.path.join(models_dir, name)
+    shutil.rmtree(target_dir)
+    return gr.update(choices=get_model_list(), value=None)
+  except Exception:
+    return gr.update(choices=get_model_list(), value=None)