光盘复制技术：一种便捷高效的数据复制方式

• Text: 全国一共有112所211高校
• Text Normalization: 全国一共有一百一十二所二一一高校
• Word Segmentation: 全国/一共/有/一百一十二/所/二一一/高校/
• G2P（注意此句中“一”的读音）:
    quan2 guo2 yi2 gong4 you3 yi4 bai3 yi1 shi2 er4 suo3 er4 yao1 yao1 gao1 xiao4
    （可以进一步把声母和韵母分开）
    q uan2 g uo2 y i2 g ong4 y ou3 y i4 b ai3 y i1 sh i2 er4 s uo3 er4 y ao1 y ao1 g ao1 x iao4
    （把音调和声韵母分开）
    q uan g uo y i g ong y ou y i b ai y i sh i er s uo er y ao y ao g ao x iao
    0 2 0 2 0 2 0 4 0 3 ...
• Prosody (prosodic words #1, prosodic phrases #2, intonation phrases #3, sentence #4):
    全国#2一共有#2一百#1一十二所#2二一一#1高校#4
    （分词的结果一般是固定的，但是不同人习惯不同，可能有不同的韵律）

# 卡尔普陪外孙玩滑梯
000001|baker_corpus|sil 20 k 12 a2 4 er2 10 p 12 u3 12 p 9 ei2 9 uai4 15 s 11 uen1 12 uan2 14 h 10 ua2 11 t 15 i1 16 sil 20

# pip install paddlespeech or
!git clone https://github.com/PaddlePaddle/PaddleSpeech.git
from IPython.display import clear_output
%cd PaddleSpeech
!pip install .
%cd -
# 清理很长的内容
clear_output()
# 本项目的依赖需要用到 nltk 包，但是有时会因为网络原因导致不好下载，此处手动下载一下放到百度服务器的包
!wget https://paddlespeech.bj.bcebos.com/Parakeet/tools/nltk_data.tar.gz
!tar zxvf nltk_data.tar.gz
clear_output()

!wget -P download https://paddlespeech.bj.bcebos.com/Parakeet/released_models/fastspeech2/fastspeech2_aishell3_ckpt_vc2_1.2.0.zip
!unzip -o -d download download/fastspeech2_aishell3_ckpt_vc2_1.2.0.zip
!wget -P download https://paddlespeech.bj.bcebos.com/Parakeet/released_models/pwgan/pwg_aishell3_ckpt_0.5.zip
!unzip -o -d download download/pwg_aishell3_ckpt_0.5.zip
clear_output()

!wget -P download https://paddlespeech.bj.bcebos.com/Parakeet/docs/demos/ref_audio.zip
!unzip -o -d download download/ref_audio.zip
clear_output()

# 设置 gpu 环境
%env CUDA_VISIBLE_DEVICES=0

import logging
import sys
import warnings
warnings.filterwarnings("ignore", category=DeprecationWarning)

import os
from pathlib import Path

import numpy as np
import paddle
import soundfile as sf
import yaml
from yacs.config import CfgNode

from paddlespeech.cli.vector import VectorExecutor
from paddlespeech.t2s.exps.syn_utils import get_am_inference
from paddlespeech.t2s.exps.syn_utils import get_voc_inference
from paddlespeech.t2s.frontend.zh_frontend import Frontend
clear_output()

am ='fastspeech2_aishell3'
am_config = 'download/fastspeech2_aishell3_ckpt_vc2_1.2.0/default.yaml'
am_ckpt = 'download/fastspeech2_aishell3_ckpt_vc2_1.2.0/snapshot_iter_96400.pdz'
am_stat = 'download/fastspeech2_aishell3_ckpt_vc2_1.2.0/speech_stats.npy'
phones_dict = "download/fastspeech2_aishell3_ckpt_vc2_1.2.0/phone_id_map.txt"

voc = 'pwgan_aishell3'
voc_config = 'download/pwg_aishell3_ckpt_0.5/default.yaml'
voc_ckpt = 'download/pwg_aishell3_ckpt_0.5/snapshot_iter_1000000.pdz'
voc_stat = 'download/pwg_aishell3_ckpt_0.5/feats_stats.npy'

# 读取 conf 配置文件并结构化
with open(am_config) as f:
    am_config = CfgNode(yaml.safe_load(f))
with open(voc_config) as f:
    voc_config = CfgNode(yaml.safe_load(f))
# 加载模型，第一次使用 vec_executor 时候会下载模型
# speaker encoder
vec_executor = VectorExecutor()

frontend = Frontend(phone_vocab_path=phones_dict)
print("frontend done!")

# acoustic model
am_inference = get_am_inference(
    am=am,
    am_config=am_config,
    am_ckpt=am_ckpt,
    am_stat=am_stat,
    phones_dict=phones_dict)

# vocoder
voc_inference = get_voc_inference(
    voc=voc,
    voc_config=voc_config,
    voc_ckpt=voc_ckpt,
    voc_stat=voc_stat)
ref_audio_path = 'download/ref_audio/yuantian01_02.wav'
sentence = '每当你觉得，想要批评什么人的时候，你切要记着，这个世界上的人，并非都具备你禀有的条件。'

input_ids = frontend.get_input_ids(sentence, merge_sentences=True)
phone_ids = input_ids["phone_ids"][0]

spk_emb = vec_executor(audio_file=ref_audio_path, force_yes=True)
spk_emb = paddle.to_tensor(spk_emb)

with paddle.no_grad():
    wav = voc_inference(am_inference(phone_ids, spk_emb=spk_emb))

sf.write(
    "output.wav",
    wav.numpy(),
    samplerate=am_config.fs)
print("Voice Cloning done!")
# 播放参考音频
import IPython.display as dp
dp.Audio("download/ref_audio/yuantian01_02.wav")
# 播放生成的音频
dp.Audio('output.wav', rate=am_config.fs)

# prepare aligner
!mkdir -p tools/aligner
%cd tools
# download MFA
# !wget https://github.com/MontrealCorpusTools/Montreal-Forced-Aligner/releases/download/v1.0.1/montreal-forced-aligner_linux.tar.gz
!wget https://paddlespeech.bj.bcebos.com/MFA/montreal-forced-aligner_linux.tar.gz
# extract MFA
!tar xvf montreal-forced-aligner_linux.tar.gz
# fix .so of MFA
%cd montreal-forced-aligner/lib
!ln -snf libpython3.6m.so.1.0 libpython3.6m.so
%cd -
# download align models and dicts
%cd aligner
!wget https://paddlespeech.bj.bcebos.com/MFA/ernie_sat/aishell3_model.zip
!wget https://paddlespeech.bj.bcebos.com/MFA/AISHELL-3/with_tone/simple.lexicon
!wget https://paddlespeech.bj.bcebos.com/MFA/ernie_sat/vctk_model.zip
!wget https://paddlespeech.bj.bcebos.com/MFA/LJSpeech-1.1/cmudict-0.7b
%cd ../../
clear_output()
# prepare pretrained FastSpeech2 models
!mkdir download
%cd download
!wget https://paddlespeech.bj.bcebos.com/Parakeet/released_models/fastspeech2/fastspeech2_conformer_baker_ckpt_0.5.zip
!wget https://paddlespeech.bj.bcebos.com/Parakeet/released_models/fastspeech2/fastspeech2_nosil_ljspeech_ckpt_0.5.zip
!unzip fastspeech2_conformer_baker_ckpt_0.5.zip
!unzip fastspeech2_nosil_ljspeech_ckpt_0.5.zip
%cd ../
clear_output()
# prepare source data
!mkdir source
%cd source
!wget https://paddlespeech.bj.bcebos.com/Parakeet/released_models/ernie_sat/source/SSB03540307.wav
!wget https://paddlespeech.bj.bcebos.com/Parakeet/released_models/ernie_sat/source/SSB03540428.wav
!wget https://paddlespeech.bj.bcebos.com/Parakeet/released_models/ernie_sat/source/LJ050-0278.wav
!wget https://paddlespeech.bj.bcebos.com/Parakeet/released_models/ernie_sat/source/p243_313.wav
!wget https://paddlespeech.bj.bcebos.com/Parakeet/released_models/ernie_sat/source/p299_096.wav
!wget https://paddlespeech.bj.bcebos.com/Parakeet/released_models/ernie_sat/source/this_was_not_the_show_for_me.wav
!wget https://paddlespeech.bj.bcebos.com/Parakeet/released_models/ernie_sat/source/README.md
%cd ../
clear_output()
# 下载预训练模型
!wget -P download https://paddlespeech.bj.bcebos.com/Parakeet/released_models/ernie_sat/erniesat_vctk_ckpt_1.2.0.zip
!unzip -o -d download download/erniesat_vctk_ckpt_1.2.0.zip
!wget -P download https://paddlespeech.bj.bcebos.com/Parakeet/released_models/hifigan/hifigan_vctk_ckpt_0.2.0.zip
!unzip -o -d download download/hifigan_vctk_ckpt_0.2.0.zip
clear_output()

import os
from pathlib import Path
from typing import List

import librosa
import numpy as np
import paddle
import pypinyin
import soundfile as sf
import yaml
from pypinyin_dict.phrase_pinyin_data import large_pinyin
from yacs.config import CfgNode

from paddlespeech.t2s.datasets.am_batch_fn import build_erniesat_collate_fn
from paddlespeech.t2s.datasets.get_feats import LogMelFBank
from paddlespeech.t2s.exps.ernie_sat.align import get_phns_spans
from paddlespeech.t2s.exps.ernie_sat.utils import eval_durs
from paddlespeech.t2s.exps.ernie_sat.utils import get_dur_adj_factor
from paddlespeech.t2s.exps.ernie_sat.utils import get_span_bdy
from paddlespeech.t2s.exps.ernie_sat.utils import get_tmp_name
from paddlespeech.t2s.exps.syn_utils import get_am_inference
from paddlespeech.t2s.exps.syn_utils import get_voc_inference
from paddlespeech.t2s.exps.syn_utils import norm

def _p2id(phonemes: List[str]) -> np.ndarray:
    # replace unk phone with sp
    phonemes = [phn if phn in vocab_phones else "sp" for phn in phonemes]
    phone_ids = [vocab_phones[item] for item in phonemes]
    return np.array(phone_ids, np.int64)

def prep_feats_with_dur(wav_path: str,
                        old_str: str='',
                        new_str: str='',
                        source_lang: str='en',
                        target_lang: str='en',
                        duration_adjust: bool=True,
                        fs: int=24000,
                        n_shift: int=300):
    '''
    Returns:
        np.ndarray: new wav, replace the part to be edited in original wav with 0
        List[str]: new phones
        List[float]: mfa start of new wav
        List[float]: mfa end of new wav
        List[int]: masked mel boundary of original wav
        List[int]: masked mel boundary of new wav
    '''
    wav_org, _ = librosa.load(wav_path, sr=fs)
    phns_spans_outs = get_phns_spans(
        wav_path=wav_path,
        old_str=old_str,
        new_str=new_str,
        source_lang=source_lang,
        target_lang=target_lang,
        fs=fs,
        n_shift=n_shift)

    mfa_start = phns_spans_outs['mfa_start']
    mfa_end = phns_spans_outs['mfa_end']
    old_phns = phns_spans_outs['old_phns']
    new_phns = phns_spans_outs['new_phns']
    span_to_repl = phns_spans_outs['span_to_repl']
    span_to_add = phns_spans_outs['span_to_add']

    # 中文的 phns 不一定都在 fastspeech2 的字典里, 用 sp 代替
    if target_lang in {'en', 'zh'}:
        old_durs = eval_durs(old_phns, target_lang=source_lang)
    else:
        assert target_lang in {'en', 'zh'}, \
            "calculate duration_predict is not support for this language..."

    orig_old_durs = [e - s for e, s in zip(mfa_end, mfa_start)]

    if duration_adjust:
        d_factor = get_dur_adj_factor(
            orig_dur=orig_old_durs, pred_dur=old_durs, phns=old_phns)
        d_factor = d_factor * 1.25
    else:
        d_factor = 1

    if target_lang in {'en', 'zh'}:
        new_durs = eval_durs(new_phns, target_lang=target_lang)
    else:
        assert target_lang == "zh" or target_lang == "en", \
            "calculate duration_predict is not support for this language..."

    # duration 要是整数
    new_durs_adjusted = [int(np.ceil(d_factor * i)) for i in new_durs]

    new_span_dur_sum = sum(new_durs_adjusted[span_to_add[0]:span_to_add[1]])
    old_span_dur_sum = sum(orig_old_durs[span_to_repl[0]:span_to_repl[1]])
    dur_offset = new_span_dur_sum - old_span_dur_sum
    new_mfa_start = mfa_start[:span_to_repl[0]]
    new_mfa_end = mfa_end[:span_to_repl[0]]

    for dur in new_durs_adjusted[span_to_add[0]:span_to_add[1]]:
        if len(new_mfa_end) == 0:
            new_mfa_start.append(0)
            new_mfa_end.append(dur)
        else:
            new_mfa_start.append(new_mfa_end[-1])
            new_mfa_end.append(new_mfa_end[-1] + dur)

    new_mfa_start += [i + dur_offset for i in mfa_start[span_to_repl[1]:]]
    new_mfa_end += [i + dur_offset for i in mfa_end[span_to_repl[1]:]]

    # 3. get new wav
    # 在原始句子后拼接
    if span_to_repl[0] >= len(mfa_start):
        wav_left_idx = len(wav_org)
        wav_right_idx = wav_left_idx
    # 在原始句子中间替换
    else:
        wav_left_idx = int(np.floor(mfa_start[span_to_repl[0]] * n_shift))
        wav_right_idx = int(np.ceil(mfa_end[span_to_repl[1] - 1] * n_shift))
    blank_wav = np.zeros(
        (int(np.ceil(new_span_dur_sum * n_shift)), ), dtype=wav_org.dtype)
    # 原始音频，需要编辑的部分替换成空音频，空音频的时间由 fs2 的 duration_predictor 决定
    new_wav = np.concatenate(
        [wav_org[:wav_left_idx], blank_wav, wav_org[wav_right_idx:]])

    # 4. get old and new mel span to be mask
    old_span_bdy = get_span_bdy(
        mfa_start=mfa_start, mfa_end=mfa_end, span_to_repl=span_to_repl)

    new_span_bdy = get_span_bdy(
        mfa_start=new_mfa_start, mfa_end=new_mfa_end, span_to_repl=span_to_add)

    # old_span_bdy, new_span_bdy 是帧级别的范围
    outs = {}
    outs['new_wav'] = new_wav
    outs['new_phns'] = new_phns
    outs['new_mfa_start'] = new_mfa_start
    outs['new_mfa_end'] = new_mfa_end
    outs['old_span_bdy'] = old_span_bdy
    outs['new_span_bdy'] = new_span_bdy
    return outs


def prep_feats(wav_path: str,
               old_str: str='',
               new_str: str='',
               source_lang: str='en',
               target_lang: str='en',
               duration_adjust: bool=True,
               fs: int=24000,
               n_shift: int=300):

    with_dur_outs = prep_feats_with_dur(
        wav_path=wav_path,
        old_str=old_str,
        new_str=new_str,
        source_lang=source_lang,
        target_lang=target_lang,
        duration_adjust=duration_adjust,
        fs=fs,
        n_shift=n_shift)

    wav_name = os.path.basename(wav_path)
    utt_id = wav_name.split('.')[0]

    wav = with_dur_outs['new_wav']
    phns = with_dur_outs['new_phns']
    mfa_start = with_dur_outs['new_mfa_start']
    mfa_end = with_dur_outs['new_mfa_end']
    old_span_bdy = with_dur_outs['old_span_bdy']
    new_span_bdy = with_dur_outs['new_span_bdy']
    span_bdy = np.array(new_span_bdy)

    mel = mel_extractor.get_log_mel_fbank(wav)
    erniesat_mean, erniesat_std = np.load(erniesat_stat)
    normed_mel = norm(mel, erniesat_mean, erniesat_std)
    tmp_name = get_tmp_name(text=old_str)
    tmpbase = './tmp_dir/' + tmp_name
    tmpbase = Path(tmpbase)
    tmpbase.mkdir(parents=True, exist_ok=True)

    mel_path = tmpbase / 'mel.npy'
    np.save(mel_path, normed_mel)
    durations = [e - s for e, s in zip(mfa_end, mfa_start)]
    text = _p2id(phns)

    datum = {
        "utt_id": utt_id,
        "spk_id": 0,
        "text": text,
        "text_lengths": len(text),
        "speech_lengths": len(normed_mel),
        "durations": durations,
        "speech": np.load(mel_path),
        "align_start": mfa_start,
        "align_end": mfa_end,
        "span_bdy": span_bdy
    }

    batch = collate_fn([datum])
    outs = dict()
    outs['batch'] = batch
    outs['old_span_bdy'] = old_span_bdy
    outs['new_span_bdy'] = new_span_bdy
    return outs


def get_mlm_output(wav_path: str,
                   old_str: str='',
                   new_str: str='',
                   source_lang: str='en',
                   target_lang: str='en',
                   duration_adjust: bool=True,
                   fs: int=24000,
                   n_shift: int=300):

    prep_feats_outs = prep_feats(
        wav_path=wav_path,
        old_str=old_str,
        new_str=new_str,
        source_lang=source_lang,
        target_lang=target_lang,
        duration_adjust=duration_adjust,
        fs=fs,
        n_shift=n_shift)

    batch = prep_feats_outs['batch']
    new_span_bdy = prep_feats_outs['new_span_bdy']
    old_span_bdy = prep_feats_outs['old_span_bdy']

    out_mels = erniesat_inference(
        speech=batch['speech'],
        text=batch['text'],
        masked_pos=batch['masked_pos'],
        speech_mask=batch['speech_mask'],
        text_mask=batch['text_mask'],
        speech_seg_pos=batch['speech_seg_pos'],
        text_seg_pos=batch['text_seg_pos'],
        span_bdy=new_span_bdy)

    # 拼接音频
    output_feat = paddle.concat(x=out_mels, axis=0)
    wav_org, _ = librosa.load(wav_path, sr=fs)
    outs = dict()
    outs['wav_org'] = wav_org
    outs['output_feat'] = output_feat
    outs['old_span_bdy'] = old_span_bdy
    outs['new_span_bdy'] = new_span_bdy

    return outs


def get_wav(wav_path: str,
            source_lang: str='en',
            target_lang: str='en',
            old_str: str='',
            new_str: str='',
            duration_adjust: bool=True,
            fs: int=24000,
            n_shift: int=300,
            task_name: str='synthesize'):

    outs = get_mlm_output(
        wav_path=wav_path,
        old_str=old_str,
        new_str=new_str,
        source_lang=source_lang,
        target_lang=target_lang,
        duration_adjust=duration_adjust,
        fs=fs,
        n_shift=n_shift)

    wav_org = outs['wav_org']
    output_feat = outs['output_feat']
    old_span_bdy = outs['old_span_bdy']
    new_span_bdy = outs['new_span_bdy']

    masked_feat = output_feat[new_span_bdy[0]:new_span_bdy[1]]

    with paddle.no_grad():
        alt_wav = voc_inference(masked_feat)
    alt_wav = np.squeeze(alt_wav)

    old_time_bdy = [n_shift * x for x in old_span_bdy]
    if task_name == 'edit':
        wav_replaced = np.concatenate(
            [wav_org[:old_time_bdy[0]], alt_wav, wav_org[old_time_bdy[1]:]])
    else:
        wav_replaced = alt_wav
    wav_dict = {"origin": wav_org, "output": wav_replaced}
    return wav_dict

erniesat_config = 'download/erniesat_vctk_ckpt_1.2.0/default.yaml'
erniesat_ckpt = 'download/erniesat_vctk_ckpt_1.2.0/snapshot_iter_199500.pdz'
erniesat_stat = 'download/erniesat_vctk_ckpt_1.2.0/speech_stats.npy'
phones_dict = 'download/erniesat_vctk_ckpt_1.2.0/phone_id_map.txt'

voc = 'hifigan_vctk'
voc_config = 'download/hifigan_vctk_ckpt_0.2.0/default.yaml'
voc_ckpt = 'download/hifigan_vctk_ckpt_0.2.0/snapshot_iter_2500000.pdz'
voc_stat = 'download/hifigan_vctk_ckpt_0.2.0/feats_stats.npy'

task_name = 'edit'
wav_path = 'source/p243_313.wav'
# should not -> is not impossible to
old_str = 'For that reason cover should not be given'
new_str = 'For that reason cover is not impossible to be given'
source_lang = 'en'
target_lang = 'en'
duration_adjust = True
output_name = 'sat_output.wav'

# evaluate(args)
with open(erniesat_config) as f:
    erniesat_config = CfgNode(yaml.safe_load(f))
with open(voc_config) as f:
    voc_config = CfgNode(yaml.safe_load(f))
# convert Chinese characters to pinyin
if source_lang == 'zh':
    old_str = pypinyin.lazy_pinyin(
        old_str,
        neutral_tone_with_five=True,
        style=pypinyin.Style.TONE3,
        tone_sandhi=True)
    old_str = ' '.join(old_str)
if target_lang == 'zh':
    new_str = pypinyin.lazy_pinyin(
        new_str,
        neutral_tone_with_five=True,
        style=pypinyin.Style.TONE3,
        tone_sandhi=True)
    new_str = ' '.join(new_str)

if task_name == 'edit':
    new_str = new_str
elif task_name == 'synthesize':
    new_str = old_str + ' ' + new_str
else:
    new_str = old_str + ' ' + new_str

# Extractor
mel_extractor = LogMelFBank(
    sr=erniesat_config.fs,
    n_fft=erniesat_config.n_fft,
    hop_length=erniesat_config.n_shift,
    win_length=erniesat_config.win_length,
    window=erniesat_config.window,
    n_mels=erniesat_config.n_mels,
    fmin=erniesat_config.fmin,
    fmax=erniesat_config.fmax)

collate_fn = build_erniesat_collate_fn(
    mlm_prob=erniesat_config.mlm_prob,
    mean_phn_span=erniesat_config.mean_phn_span,
    seg_emb=erniesat_config.model['enc_input_layer'] == 'sega_mlm',
    text_masking=False)

vocab_phones = {}

with open(phones_dict, 'rt') as f:
    phn_id = [line.strip().split() for line in f.readlines()]
for phn, id in phn_id:
    vocab_phones[phn] = int(id)

# ernie sat model
erniesat_inference = get_am_inference(
    am='erniesat_dataset',
    am_config=erniesat_config,
    am_ckpt=erniesat_ckpt,
    am_stat=erniesat_stat,
    phones_dict=phones_dict)

# vocoder
voc_inference = get_voc_inference(
    voc=voc,
    voc_config=voc_config,
    voc_ckpt=voc_ckpt,
    voc_stat=voc_stat)


wav_dict = get_wav(
    wav_path=wav_path,
    source_lang=source_lang,
    target_lang=target_lang,
    old_str=old_str,
    new_str=new_str,
    duration_adjust=duration_adjust,
    fs=erniesat_config.fs,
    n_shift=erniesat_config.n_shift,
    task_name=task_name)

sf.write(
    output_name, wav_dict['output'], samplerate=erniesat_config.fs)
print(
    f"\033[1;32;m Generated audio saved into {output_name} ! \033[0m")
# 播放输入音频
import IPython.display as dp
dp.Audio("source/p243_313.wav")
# 播放编辑好的音频
dp.Audio('sat_output.wav', rate=erniesat_config.fs)