| import numpy as np |
| import onnxruntime as ort |
| from rknnlite.api.rknn_lite import RKNNLite |
| import numpy as np |
| import soundfile as sf |
| from transformers import AutoTokenizer |
| import time |
| import os |
| import re |
| import cn2an |
| from pypinyin import lazy_pinyin, Style |
| from typing import List |
| from typing import Tuple |
| import jieba |
| import jieba.posseg as psg |
|
|
| def convert_pad_shape(pad_shape): |
| layer = pad_shape[::-1] |
| pad_shape = [item for sublist in layer for item in sublist] |
| return pad_shape |
|
|
|
|
| def sequence_mask(length, max_length=None): |
| if max_length is None: |
| max_length = length.max() |
| x = np.arange(max_length, dtype=length.dtype) |
| return np.expand_dims(x, 0) < np.expand_dims(length, 1) |
|
|
|
|
| def generate_path(duration, mask): |
| """ |
| duration: [b, 1, t_x] |
| mask: [b, 1, t_y, t_x] |
| """ |
|
|
| b, _, t_y, t_x = mask.shape |
| cum_duration = np.cumsum(duration, -1) |
|
|
| cum_duration_flat = cum_duration.reshape(b * t_x) |
| path = sequence_mask(cum_duration_flat, t_y) |
| path = path.reshape(b, t_x, t_y) |
| path = path ^ np.pad(path, ((0, 0), (1, 0), (0, 0)))[:, :-1] |
| path = np.expand_dims(path, 1).transpose(0, 1, 3, 2) |
| return path |
|
|
|
|
| class InferenceSession: |
| def __init__(self, path, Providers=["CPUExecutionProvider"]): |
| ort_config = ort.SessionOptions() |
| ort_config.graph_optimization_level = ort.GraphOptimizationLevel.ORT_ENABLE_ALL |
| ort_config.intra_op_num_threads = 4 |
| ort_config.inter_op_num_threads = 4 |
| self.enc = ort.InferenceSession(path["enc"], providers=Providers, sess_options=ort_config) |
| self.emb_g = ort.InferenceSession(path["emb_g"], providers=Providers, sess_options=ort_config) |
| self.dp = ort.InferenceSession(path["dp"], providers=Providers, sess_options=ort_config) |
| self.sdp = ort.InferenceSession(path["sdp"], providers=Providers, sess_options=ort_config) |
| |
| |
| |
| |
| self.flow = ort.InferenceSession(path["flow"], providers=Providers, sess_options=ort_config) |
| self.dec = RKNNLite(verbose=False) |
| self.dec.load_rknn(path["dec"]) |
| self.dec.init_runtime() |
| |
|
|
| def __call__( |
| self, |
| seq, |
| tone, |
| language, |
| bert_zh, |
| bert_jp, |
| bert_en, |
| vqidx, |
| sid, |
| seed=114514, |
| seq_noise_scale=0.8, |
| sdp_noise_scale=0.6, |
| length_scale=1.0, |
| sdp_ratio=0.0, |
| rknn_pad_to = 1024 |
| ): |
| if seq.ndim == 1: |
| seq = np.expand_dims(seq, 0) |
| if tone.ndim == 1: |
| tone = np.expand_dims(tone, 0) |
| if language.ndim == 1: |
| language = np.expand_dims(language, 0) |
| assert (seq.ndim == 2, tone.ndim == 2, language.ndim == 2) |
|
|
| start_time = time.time() |
| g = self.emb_g.run( |
| None, |
| { |
| "sid": sid.astype(np.int64), |
| }, |
| )[0] |
| emb_g_time = time.time() - start_time |
| print(f"emb_g 运行时间: {emb_g_time:.4f} 秒") |
|
|
| g = np.expand_dims(g, -1) |
| start_time = time.time() |
| enc_rtn = self.enc.run( |
| None, |
| { |
| "x": seq.astype(np.int64), |
| "t": tone.astype(np.int64), |
| "language": language.astype(np.int64), |
| "bert_0": bert_zh.astype(np.float32), |
| "bert_1": bert_jp.astype(np.float32), |
| "bert_2": bert_en.astype(np.float32), |
| "g": g.astype(np.float32), |
| |
| "vqidx": vqidx.astype(np.int64), |
| "sid": sid.astype(np.int64), |
| }, |
| ) |
| enc_time = time.time() - start_time |
| print(f"enc 运行时间: {enc_time:.4f} 秒") |
|
|
| x, m_p, logs_p, x_mask = enc_rtn[0], enc_rtn[1], enc_rtn[2], enc_rtn[3] |
| np.random.seed(seed) |
| zinput = np.random.randn(x.shape[0], 2, x.shape[2]) * sdp_noise_scale |
|
|
| start_time = time.time() |
| sdp_output = self.sdp.run( |
| None, {"x": x, "x_mask": x_mask, "zin": zinput.astype(np.float32), "g": g} |
| )[0] |
| sdp_time = time.time() - start_time |
| print(f"sdp 运行时间: {sdp_time:.4f} 秒") |
|
|
| start_time = time.time() |
| dp_output = self.dp.run(None, {"x": x, "x_mask": x_mask, "g": g})[0] |
| dp_time = time.time() - start_time |
| print(f"dp 运行时间: {dp_time:.4f} 秒") |
|
|
| logw = sdp_output * (sdp_ratio) + dp_output * (1 - sdp_ratio) |
| w = np.exp(logw) * x_mask * length_scale |
| w_ceil = np.ceil(w) |
| y_lengths = np.clip(np.sum(w_ceil, (1, 2)), a_min=1.0, a_max=100000).astype( |
| np.int64 |
| ) |
| y_mask = np.expand_dims(sequence_mask(y_lengths, None), 1) |
| attn_mask = np.expand_dims(x_mask, 2) * np.expand_dims(y_mask, -1) |
| attn = generate_path(w_ceil, attn_mask) |
| m_p = np.matmul(attn.squeeze(1), m_p.transpose(0, 2, 1)).transpose( |
| 0, 2, 1 |
| ) |
| logs_p = np.matmul(attn.squeeze(1), logs_p.transpose(0, 2, 1)).transpose( |
| 0, 2, 1 |
| ) |
|
|
| z_p = ( |
| m_p |
| + np.random.randn(m_p.shape[0], m_p.shape[1], m_p.shape[2]) |
| * np.exp(logs_p) |
| * seq_noise_scale |
| ) |
| |
| actual_len = z_p.shape[2] |
| if actual_len > rknn_pad_to: |
| print("警告, 输入长度超过 rknn_pad_to, 将被截断") |
| z_p = z_p[:,:,:rknn_pad_to] |
| y_mask = y_mask[:,:,:rknn_pad_to] |
| else: |
| z_p = np.pad(z_p, ((0, 0), (0, 0), (0, rknn_pad_to - z_p.shape[2]))) |
| y_mask = np.pad(y_mask, ((0, 0), (0, 0), (0, rknn_pad_to - y_mask.shape[2]))) |
|
|
| start_time = time.time() |
| z = self.flow.run( |
| None, |
| { |
| "z_p": z_p.astype(np.float32), |
| "y_mask": y_mask.astype(np.float32), |
| "g": g, |
| }, |
| )[0] |
| flow_time = time.time() - start_time |
| print(f"flow 运行时间: {flow_time:.4f} 秒") |
|
|
| start_time = time.time() |
| dec_output = self.dec.inference([z.astype(np.float32), g])[0] |
| dec_time = time.time() - start_time |
| print(f"dec 运行时间: {dec_time:.4f} 秒") |
|
|
| |
| return dec_output[:,:,:actual_len*512] |
|
|
|
|
|
|
|
|
| class ToneSandhi: |
| def __init__(self): |
| self.must_neural_tone_words = { |
| "麻烦", |
| "麻利", |
| "鸳鸯", |
| "高粱", |
| "骨头", |
| "骆驼", |
| "马虎", |
| "首饰", |
| "馒头", |
| "馄饨", |
| "风筝", |
| "难为", |
| "队伍", |
| "阔气", |
| "闺女", |
| "门道", |
| "锄头", |
| "铺盖", |
| "铃铛", |
| "铁匠", |
| "钥匙", |
| "里脊", |
| "里头", |
| "部分", |
| "那么", |
| "道士", |
| "造化", |
| "迷糊", |
| "连累", |
| "这么", |
| "这个", |
| "运气", |
| "过去", |
| "软和", |
| "转悠", |
| "踏实", |
| "跳蚤", |
| "跟头", |
| "趔趄", |
| "财主", |
| "豆腐", |
| "讲究", |
| "记性", |
| "记号", |
| "认识", |
| "规矩", |
| "见识", |
| "裁缝", |
| "补丁", |
| "衣裳", |
| "衣服", |
| "衙门", |
| "街坊", |
| "行李", |
| "行当", |
| "蛤蟆", |
| "蘑菇", |
| "薄荷", |
| "葫芦", |
| "葡萄", |
| "萝卜", |
| "荸荠", |
| "苗条", |
| "苗头", |
| "苍蝇", |
| "芝麻", |
| "舒服", |
| "舒坦", |
| "舌头", |
| "自在", |
| "膏药", |
| "脾气", |
| "脑袋", |
| "脊梁", |
| "能耐", |
| "胳膊", |
| "胭脂", |
| "胡萝", |
| "胡琴", |
| "胡同", |
| "聪明", |
| "耽误", |
| "耽搁", |
| "耷拉", |
| "耳朵", |
| "老爷", |
| "老实", |
| "老婆", |
| "老头", |
| "老太", |
| "翻腾", |
| "罗嗦", |
| "罐头", |
| "编辑", |
| "结实", |
| "红火", |
| "累赘", |
| "糨糊", |
| "糊涂", |
| "精神", |
| "粮食", |
| "簸箕", |
| "篱笆", |
| "算计", |
| "算盘", |
| "答应", |
| "笤帚", |
| "笑语", |
| "笑话", |
| "窟窿", |
| "窝囊", |
| "窗户", |
| "稳当", |
| "稀罕", |
| "称呼", |
| "秧歌", |
| "秀气", |
| "秀才", |
| "福气", |
| "祖宗", |
| "砚台", |
| "码头", |
| "石榴", |
| "石头", |
| "石匠", |
| "知识", |
| "眼睛", |
| "眯缝", |
| "眨巴", |
| "眉毛", |
| "相声", |
| "盘算", |
| "白净", |
| "痢疾", |
| "痛快", |
| "疟疾", |
| "疙瘩", |
| "疏忽", |
| "畜生", |
| "生意", |
| "甘蔗", |
| "琵琶", |
| "琢磨", |
| "琉璃", |
| "玻璃", |
| "玫瑰", |
| "玄乎", |
| "狐狸", |
| "状元", |
| "特务", |
| "牲口", |
| "牙碜", |
| "牌楼", |
| "爽快", |
| "爱人", |
| "热闹", |
| "烧饼", |
| "烟筒", |
| "烂糊", |
| "点心", |
| "炊帚", |
| "灯笼", |
| "火候", |
| "漂亮", |
| "滑溜", |
| "溜达", |
| "温和", |
| "清楚", |
| "消息", |
| "浪头", |
| "活泼", |
| "比方", |
| "正经", |
| "欺负", |
| "模糊", |
| "槟榔", |
| "棺材", |
| "棒槌", |
| "棉花", |
| "核桃", |
| "栅栏", |
| "柴火", |
| "架势", |
| "枕头", |
| "枇杷", |
| "机灵", |
| "本事", |
| "木头", |
| "木匠", |
| "朋友", |
| "月饼", |
| "月亮", |
| "暖和", |
| "明白", |
| "时候", |
| "新鲜", |
| "故事", |
| "收拾", |
| "收成", |
| "提防", |
| "挖苦", |
| "挑剔", |
| "指甲", |
| "指头", |
| "拾掇", |
| "拳头", |
| "拨弄", |
| "招牌", |
| "招呼", |
| "抬举", |
| "护士", |
| "折腾", |
| "扫帚", |
| "打量", |
| "打算", |
| "打点", |
| "打扮", |
| "打听", |
| "打发", |
| "扎实", |
| "扁担", |
| "戒指", |
| "懒得", |
| "意识", |
| "意思", |
| "情形", |
| "悟性", |
| "怪物", |
| "思量", |
| "怎么", |
| "念头", |
| "念叨", |
| "快活", |
| "忙活", |
| "志气", |
| "心思", |
| "得罪", |
| "张罗", |
| "弟兄", |
| "开通", |
| "应酬", |
| "庄稼", |
| "干事", |
| "帮手", |
| "帐篷", |
| "希罕", |
| "师父", |
| "师傅", |
| "巴结", |
| "巴掌", |
| "差事", |
| "工夫", |
| "岁数", |
| "屁股", |
| "尾巴", |
| "少爷", |
| "小气", |
| "小伙", |
| "将就", |
| "对头", |
| "对付", |
| "寡妇", |
| "家伙", |
| "客气", |
| "实在", |
| "官司", |
| "学问", |
| "学生", |
| "字号", |
| "嫁妆", |
| "媳妇", |
| "媒人", |
| "婆家", |
| "娘家", |
| "委屈", |
| "姑娘", |
| "姐夫", |
| "妯娌", |
| "妥当", |
| "妖精", |
| "奴才", |
| "女婿", |
| "头发", |
| "太阳", |
| "大爷", |
| "大方", |
| "大意", |
| "大夫", |
| "多少", |
| "多么", |
| "外甥", |
| "壮实", |
| "地道", |
| "地方", |
| "在乎", |
| "困难", |
| "嘴巴", |
| "嘱咐", |
| "嘟囔", |
| "嘀咕", |
| "喜欢", |
| "喇嘛", |
| "喇叭", |
| "商量", |
| "唾沫", |
| "哑巴", |
| "哈欠", |
| "哆嗦", |
| "咳嗽", |
| "和尚", |
| "告诉", |
| "告示", |
| "含糊", |
| "吓唬", |
| "后头", |
| "名字", |
| "名堂", |
| "合同", |
| "吆喝", |
| "叫唤", |
| "口袋", |
| "厚道", |
| "厉害", |
| "千斤", |
| "包袱", |
| "包涵", |
| "匀称", |
| "勤快", |
| "动静", |
| "动弹", |
| "功夫", |
| "力气", |
| "前头", |
| "刺猬", |
| "刺激", |
| "别扭", |
| "利落", |
| "利索", |
| "利害", |
| "分析", |
| "出息", |
| "凑合", |
| "凉快", |
| "冷战", |
| "冤枉", |
| "冒失", |
| "养活", |
| "关系", |
| "先生", |
| "兄弟", |
| "便宜", |
| "使唤", |
| "佩服", |
| "作坊", |
| "体面", |
| "位置", |
| "似的", |
| "伙计", |
| "休息", |
| "什么", |
| "人家", |
| "亲戚", |
| "亲家", |
| "交情", |
| "云彩", |
| "事情", |
| "买卖", |
| "主意", |
| "丫头", |
| "丧气", |
| "两口", |
| "东西", |
| "东家", |
| "世故", |
| "不由", |
| "不在", |
| "下水", |
| "下巴", |
| "上头", |
| "上司", |
| "丈夫", |
| "丈人", |
| "一辈", |
| "那个", |
| "菩萨", |
| "父亲", |
| "母亲", |
| "咕噜", |
| "邋遢", |
| "费用", |
| "冤家", |
| "甜头", |
| "介绍", |
| "荒唐", |
| "大人", |
| "泥鳅", |
| "幸福", |
| "熟悉", |
| "计划", |
| "扑腾", |
| "蜡烛", |
| "姥爷", |
| "照顾", |
| "喉咙", |
| "吉他", |
| "弄堂", |
| "蚂蚱", |
| "凤凰", |
| "拖沓", |
| "寒碜", |
| "糟蹋", |
| "倒腾", |
| "报复", |
| "逻辑", |
| "盘缠", |
| "喽啰", |
| "牢骚", |
| "咖喱", |
| "扫把", |
| "惦记", |
| } |
| self.must_not_neural_tone_words = { |
| "男子", |
| "女子", |
| "分子", |
| "原子", |
| "量子", |
| "莲子", |
| "石子", |
| "瓜子", |
| "电子", |
| "人人", |
| "虎虎", |
| } |
| self.punc = "：，；。？！“”‘’':,;.?!" |
|
|
| |
| |
| |
| |
| |
| def _neural_sandhi(self, word: str, pos: str, finals: List[str]) -> List[str]: |
| |
| for j, item in enumerate(word): |
| if ( |
| j - 1 >= 0 |
| and item == word[j - 1] |
| and pos[0] in {"n", "v", "a"} |
| and word not in self.must_not_neural_tone_words |
| ): |
| finals[j] = finals[j][:-1] + "5" |
| ge_idx = word.find("个") |
| if len(word) >= 1 and word[-1] in "吧呢啊呐噻嘛吖嗨呐哦哒额滴哩哟喽啰耶喔诶": |
| finals[-1] = finals[-1][:-1] + "5" |
| elif len(word) >= 1 and word[-1] in "的地得": |
| finals[-1] = finals[-1][:-1] + "5" |
| |
| |
| |
| elif ( |
| len(word) > 1 |
| and word[-1] in "们子" |
| and pos in {"r", "n"} |
| and word not in self.must_not_neural_tone_words |
| ): |
| finals[-1] = finals[-1][:-1] + "5" |
| |
| elif len(word) > 1 and word[-1] in "上下里" and pos in {"s", "l", "f"}: |
| finals[-1] = finals[-1][:-1] + "5" |
| |
| elif len(word) > 1 and word[-1] in "来去" and word[-2] in "上下进出回过起开": |
| finals[-1] = finals[-1][:-1] + "5" |
| |
| elif ( |
| ge_idx >= 1 |
| and ( |
| word[ge_idx - 1].isnumeric() |
| or word[ge_idx - 1] in "几有两半多各整每做是" |
| ) |
| ) or word == "个": |
| finals[ge_idx] = finals[ge_idx][:-1] + "5" |
| else: |
| if ( |
| word in self.must_neural_tone_words |
| or word[-2:] in self.must_neural_tone_words |
| ): |
| finals[-1] = finals[-1][:-1] + "5" |
|
|
| word_list = self._split_word(word) |
| finals_list = [finals[: len(word_list[0])], finals[len(word_list[0]) :]] |
| for i, word in enumerate(word_list): |
| |
| if ( |
| word in self.must_neural_tone_words |
| or word[-2:] in self.must_neural_tone_words |
| ): |
| finals_list[i][-1] = finals_list[i][-1][:-1] + "5" |
| finals = sum(finals_list, []) |
| return finals |
|
|
| def _bu_sandhi(self, word: str, finals: List[str]) -> List[str]: |
| |
| if len(word) == 3 and word[1] == "不": |
| finals[1] = finals[1][:-1] + "5" |
| else: |
| for i, char in enumerate(word): |
| |
| if char == "不" and i + 1 < len(word) and finals[i + 1][-1] == "4": |
| finals[i] = finals[i][:-1] + "2" |
| return finals |
|
|
| def _yi_sandhi(self, word: str, finals: List[str]) -> List[str]: |
| |
| if word.find("一") != -1 and all( |
| [item.isnumeric() for item in word if item != "一"] |
| ): |
| return finals |
| |
| elif len(word) == 3 and word[1] == "一" and word[0] == word[-1]: |
| finals[1] = finals[1][:-1] + "5" |
| |
| elif word.startswith("第一"): |
| finals[1] = finals[1][:-1] + "1" |
| else: |
| for i, char in enumerate(word): |
| if char == "一" and i + 1 < len(word): |
| |
| if finals[i + 1][-1] == "4": |
| finals[i] = finals[i][:-1] + "2" |
| |
| else: |
| |
| if word[i + 1] not in self.punc: |
| finals[i] = finals[i][:-1] + "4" |
| return finals |
|
|
| def _split_word(self, word: str) -> List[str]: |
| word_list = jieba.cut_for_search(word) |
| word_list = sorted(word_list, key=lambda i: len(i), reverse=False) |
| first_subword = word_list[0] |
| first_begin_idx = word.find(first_subword) |
| if first_begin_idx == 0: |
| second_subword = word[len(first_subword) :] |
| new_word_list = [first_subword, second_subword] |
| else: |
| second_subword = word[: -len(first_subword)] |
| new_word_list = [second_subword, first_subword] |
| return new_word_list |
|
|
| def _three_sandhi(self, word: str, finals: List[str]) -> List[str]: |
| if len(word) == 2 and self._all_tone_three(finals): |
| finals[0] = finals[0][:-1] + "2" |
| elif len(word) == 3: |
| word_list = self._split_word(word) |
| if self._all_tone_three(finals): |
| |
| if len(word_list[0]) == 2: |
| finals[0] = finals[0][:-1] + "2" |
| finals[1] = finals[1][:-1] + "2" |
| |
| elif len(word_list[0]) == 1: |
| finals[1] = finals[1][:-1] + "2" |
| else: |
| finals_list = [finals[: len(word_list[0])], finals[len(word_list[0]) :]] |
| if len(finals_list) == 2: |
| for i, sub in enumerate(finals_list): |
| |
| if self._all_tone_three(sub) and len(sub) == 2: |
| finals_list[i][0] = finals_list[i][0][:-1] + "2" |
| |
| elif ( |
| i == 1 |
| and not self._all_tone_three(sub) |
| and finals_list[i][0][-1] == "3" |
| and finals_list[0][-1][-1] == "3" |
| ): |
| finals_list[0][-1] = finals_list[0][-1][:-1] + "2" |
| finals = sum(finals_list, []) |
| |
| elif len(word) == 4: |
| finals_list = [finals[:2], finals[2:]] |
| finals = [] |
| for sub in finals_list: |
| if self._all_tone_three(sub): |
| sub[0] = sub[0][:-1] + "2" |
| finals += sub |
|
|
| return finals |
|
|
| def _all_tone_three(self, finals: List[str]) -> bool: |
| return all(x[-1] == "3" for x in finals) |
|
|
| |
| |
| def _merge_bu(self, seg: List[Tuple[str, str]]) -> List[Tuple[str, str]]: |
| new_seg = [] |
| last_word = "" |
| for word, pos in seg: |
| if last_word == "不": |
| word = last_word + word |
| if word != "不": |
| new_seg.append((word, pos)) |
| last_word = word[:] |
| if last_word == "不": |
| new_seg.append((last_word, "d")) |
| last_word = "" |
| return new_seg |
|
|
| |
| |
| |
| |
| |
| |
| def _merge_yi(self, seg: List[Tuple[str, str]]) -> List[Tuple[str, str]]: |
| new_seg = [] |
| |
| for i, (word, pos) in enumerate(seg): |
| if ( |
| i - 1 >= 0 |
| and word == "一" |
| and i + 1 < len(seg) |
| and seg[i - 1][0] == seg[i + 1][0] |
| and seg[i - 1][1] == "v" |
| ): |
| new_seg[i - 1][0] = new_seg[i - 1][0] + "一" + new_seg[i - 1][0] |
| else: |
| if ( |
| i - 2 >= 0 |
| and seg[i - 1][0] == "一" |
| and seg[i - 2][0] == word |
| and pos == "v" |
| ): |
| continue |
| else: |
| new_seg.append([word, pos]) |
| seg = new_seg |
| new_seg = [] |
| |
| for i, (word, pos) in enumerate(seg): |
| if new_seg and new_seg[-1][0] == "一": |
| new_seg[-1][0] = new_seg[-1][0] + word |
| else: |
| new_seg.append([word, pos]) |
| return new_seg |
|
|
| |
| def _merge_continuous_three_tones( |
| self, seg: List[Tuple[str, str]] |
| ) -> List[Tuple[str, str]]: |
| new_seg = [] |
| sub_finals_list = [ |
| lazy_pinyin(word, neutral_tone_with_five=True, style=Style.FINALS_TONE3) |
| for (word, pos) in seg |
| ] |
| assert len(sub_finals_list) == len(seg) |
| merge_last = [False] * len(seg) |
| for i, (word, pos) in enumerate(seg): |
| if ( |
| i - 1 >= 0 |
| and self._all_tone_three(sub_finals_list[i - 1]) |
| and self._all_tone_three(sub_finals_list[i]) |
| and not merge_last[i - 1] |
| ): |
| |
| if ( |
| not self._is_reduplication(seg[i - 1][0]) |
| and len(seg[i - 1][0]) + len(seg[i][0]) <= 3 |
| ): |
| new_seg[-1][0] = new_seg[-1][0] + seg[i][0] |
| merge_last[i] = True |
| else: |
| new_seg.append([word, pos]) |
| else: |
| new_seg.append([word, pos]) |
|
|
| return new_seg |
|
|
| def _is_reduplication(self, word: str) -> bool: |
| return len(word) == 2 and word[0] == word[1] |
|
|
| |
| def _merge_continuous_three_tones_2( |
| self, seg: List[Tuple[str, str]] |
| ) -> List[Tuple[str, str]]: |
| new_seg = [] |
| sub_finals_list = [ |
| lazy_pinyin(word, neutral_tone_with_five=True, style=Style.FINALS_TONE3) |
| for (word, pos) in seg |
| ] |
| assert len(sub_finals_list) == len(seg) |
| merge_last = [False] * len(seg) |
| for i, (word, pos) in enumerate(seg): |
| if ( |
| i - 1 >= 0 |
| and sub_finals_list[i - 1][-1][-1] == "3" |
| and sub_finals_list[i][0][-1] == "3" |
| and not merge_last[i - 1] |
| ): |
| |
| if ( |
| not self._is_reduplication(seg[i - 1][0]) |
| and len(seg[i - 1][0]) + len(seg[i][0]) <= 3 |
| ): |
| new_seg[-1][0] = new_seg[-1][0] + seg[i][0] |
| merge_last[i] = True |
| else: |
| new_seg.append([word, pos]) |
| else: |
| new_seg.append([word, pos]) |
| return new_seg |
|
|
| def _merge_er(self, seg: List[Tuple[str, str]]) -> List[Tuple[str, str]]: |
| new_seg = [] |
| for i, (word, pos) in enumerate(seg): |
| if i - 1 >= 0 and word == "儿" and seg[i - 1][0] != "#": |
| new_seg[-1][0] = new_seg[-1][0] + seg[i][0] |
| else: |
| new_seg.append([word, pos]) |
| return new_seg |
|
|
| def _merge_reduplication(self, seg: List[Tuple[str, str]]) -> List[Tuple[str, str]]: |
| new_seg = [] |
| for i, (word, pos) in enumerate(seg): |
| if new_seg and word == new_seg[-1][0]: |
| new_seg[-1][0] = new_seg[-1][0] + seg[i][0] |
| else: |
| new_seg.append([word, pos]) |
| return new_seg |
|
|
| def pre_merge_for_modify(self, seg: List[Tuple[str, str]]) -> List[Tuple[str, str]]: |
| seg = self._merge_bu(seg) |
| try: |
| seg = self._merge_yi(seg) |
| except: |
| print("_merge_yi failed") |
| seg = self._merge_reduplication(seg) |
| seg = self._merge_continuous_three_tones(seg) |
| seg = self._merge_continuous_three_tones_2(seg) |
| seg = self._merge_er(seg) |
| return seg |
|
|
| def modified_tone(self, word: str, pos: str, finals: List[str]) -> List[str]: |
| finals = self._bu_sandhi(word, finals) |
| finals = self._yi_sandhi(word, finals) |
| finals = self._neural_sandhi(word, pos, finals) |
| finals = self._three_sandhi(word, finals) |
| return finals |
|
|
|
|
| punctuation = ["!", "?", "…", ",", ".", "'", "-"] |
| pu_symbols = punctuation + ["SP", "UNK"] |
| pad = "_" |
|
|
| |
| zh_symbols = [ |
| "E", |
| "En", |
| "a", |
| "ai", |
| "an", |
| "ang", |
| "ao", |
| "b", |
| "c", |
| "ch", |
| "d", |
| "e", |
| "ei", |
| "en", |
| "eng", |
| "er", |
| "f", |
| "g", |
| "h", |
| "i", |
| "i0", |
| "ia", |
| "ian", |
| "iang", |
| "iao", |
| "ie", |
| "in", |
| "ing", |
| "iong", |
| "ir", |
| "iu", |
| "j", |
| "k", |
| "l", |
| "m", |
| "n", |
| "o", |
| "ong", |
| "ou", |
| "p", |
| "q", |
| "r", |
| "s", |
| "sh", |
| "t", |
| "u", |
| "ua", |
| "uai", |
| "uan", |
| "uang", |
| "ui", |
| "un", |
| "uo", |
| "v", |
| "van", |
| "ve", |
| "vn", |
| "w", |
| "x", |
| "y", |
| "z", |
| "zh", |
| "AA", |
| "EE", |
| "OO", |
| ] |
| num_zh_tones = 6 |
|
|
| |
| ja_symbols = [ |
| "N", |
| "a", |
| "a:", |
| "b", |
| "by", |
| "ch", |
| "d", |
| "dy", |
| "e", |
| "e:", |
| "f", |
| "g", |
| "gy", |
| "h", |
| "hy", |
| "i", |
| "i:", |
| "j", |
| "k", |
| "ky", |
| "m", |
| "my", |
| "n", |
| "ny", |
| "o", |
| "o:", |
| "p", |
| "py", |
| "q", |
| "r", |
| "ry", |
| "s", |
| "sh", |
| "t", |
| "ts", |
| "ty", |
| "u", |
| "u:", |
| "w", |
| "y", |
| "z", |
| "zy", |
| ] |
| num_ja_tones = 2 |
|
|
| |
| en_symbols = [ |
| "aa", |
| "ae", |
| "ah", |
| "ao", |
| "aw", |
| "ay", |
| "b", |
| "ch", |
| "d", |
| "dh", |
| "eh", |
| "er", |
| "ey", |
| "f", |
| "g", |
| "hh", |
| "ih", |
| "iy", |
| "jh", |
| "k", |
| "l", |
| "m", |
| "n", |
| "ng", |
| "ow", |
| "oy", |
| "p", |
| "r", |
| "s", |
| "sh", |
| "t", |
| "th", |
| "uh", |
| "uw", |
| "V", |
| "w", |
| "y", |
| "z", |
| "zh", |
| ] |
| num_en_tones = 4 |
|
|
| |
| normal_symbols = sorted(set(zh_symbols + ja_symbols + en_symbols)) |
| symbols = [pad] + normal_symbols + pu_symbols |
| sil_phonemes_ids = [symbols.index(i) for i in pu_symbols] |
|
|
| |
| num_tones = num_zh_tones + num_ja_tones + num_en_tones |
|
|
| |
| language_id_map = {"ZH": 0, "JP": 1, "EN": 2} |
| num_languages = len(language_id_map.keys()) |
|
|
| language_tone_start_map = { |
| "ZH": 0, |
| "JP": num_zh_tones, |
| "EN": num_zh_tones + num_ja_tones, |
| } |
|
|
| current_file_path = os.path.dirname(__file__) |
| pinyin_to_symbol_map = { |
| line.split("\t")[0]: line.strip().split("\t")[1] |
| for line in open(os.path.join(current_file_path, "opencpop-strict.txt")).readlines() |
| } |
|
|
|
|
|
|
|
|
| rep_map = { |
| "：": ",", |
| "；": ",", |
| "，": ",", |
| "。": ".", |
| "！": "!", |
| "？": "?", |
| "\n": ".", |
| "·": ",", |
| "、": ",", |
| "...": "…", |
| "$": ".", |
| "“": "'", |
| "”": "'", |
| '"': "'", |
| "‘": "'", |
| "’": "'", |
| "（": "'", |
| "）": "'", |
| "(": "'", |
| ")": "'", |
| "《": "'", |
| "》": "'", |
| "【": "'", |
| "】": "'", |
| "[": "'", |
| "]": "'", |
| "—": "-", |
| "～": "-", |
| "~": "-", |
| "「": "'", |
| "」": "'", |
| } |
|
|
| tone_modifier = ToneSandhi() |
|
|
|
|
| def replace_punctuation(text): |
| text = text.replace("嗯", "恩").replace("呣", "母") |
| pattern = re.compile("|".join(re.escape(p) for p in rep_map.keys())) |
|
|
| replaced_text = pattern.sub(lambda x: rep_map[x.group()], text) |
|
|
| replaced_text = re.sub( |
| r"[^\u4e00-\u9fa5" + "".join(punctuation) + r"]+", "", replaced_text |
| ) |
|
|
| return replaced_text |
|
|
|
|
| def g2p(text): |
| pattern = r"(?<=[{0}])\s*".format("".join(punctuation)) |
| sentences = [i for i in re.split(pattern, text) if i.strip() != ""] |
| phones, tones, word2ph = _g2p(sentences) |
| assert sum(word2ph) == len(phones) |
| assert len(word2ph) == len(text) |
| phones = ["_"] + phones + ["_"] |
| tones = [0] + tones + [0] |
| word2ph = [1] + word2ph + [1] |
| return phones, tones, word2ph |
|
|
|
|
| def _get_initials_finals(word): |
| initials = [] |
| finals = [] |
| orig_initials = lazy_pinyin(word, neutral_tone_with_five=True, style=Style.INITIALS) |
| orig_finals = lazy_pinyin( |
| word, neutral_tone_with_five=True, style=Style.FINALS_TONE3 |
| ) |
| for c, v in zip(orig_initials, orig_finals): |
| initials.append(c) |
| finals.append(v) |
| return initials, finals |
|
|
|
|
| def _g2p(segments): |
| phones_list = [] |
| tones_list = [] |
| word2ph = [] |
| for seg in segments: |
| |
| seg = re.sub("[a-zA-Z]+", "", seg) |
| seg_cut = psg.lcut(seg) |
| initials = [] |
| finals = [] |
| seg_cut = tone_modifier.pre_merge_for_modify(seg_cut) |
| for word, pos in seg_cut: |
| if pos == "eng": |
| continue |
| sub_initials, sub_finals = _get_initials_finals(word) |
| sub_finals = tone_modifier.modified_tone(word, pos, sub_finals) |
| initials.append(sub_initials) |
| finals.append(sub_finals) |
|
|
| |
| initials = sum(initials, []) |
| finals = sum(finals, []) |
| |
| for c, v in zip(initials, finals): |
| raw_pinyin = c + v |
| |
| |
| if c == v: |
| assert c in punctuation |
| phone = [c] |
| tone = "0" |
| word2ph.append(1) |
| else: |
| v_without_tone = v[:-1] |
| tone = v[-1] |
|
|
| pinyin = c + v_without_tone |
| assert tone in "12345" |
|
|
| if c: |
| |
| v_rep_map = { |
| "uei": "ui", |
| "iou": "iu", |
| "uen": "un", |
| } |
| if v_without_tone in v_rep_map.keys(): |
| pinyin = c + v_rep_map[v_without_tone] |
| else: |
| |
| pinyin_rep_map = { |
| "ing": "ying", |
| "i": "yi", |
| "in": "yin", |
| "u": "wu", |
| } |
| if pinyin in pinyin_rep_map.keys(): |
| pinyin = pinyin_rep_map[pinyin] |
| else: |
| single_rep_map = { |
| "v": "yu", |
| "e": "e", |
| "i": "y", |
| "u": "w", |
| } |
| if pinyin[0] in single_rep_map.keys(): |
| pinyin = single_rep_map[pinyin[0]] + pinyin[1:] |
|
|
| assert pinyin in pinyin_to_symbol_map.keys(), (pinyin, seg, raw_pinyin) |
| phone = pinyin_to_symbol_map[pinyin].split(" ") |
| word2ph.append(len(phone)) |
|
|
| phones_list += phone |
| tones_list += [int(tone)] * len(phone) |
| return phones_list, tones_list, word2ph |
|
|
|
|
| def text_normalize(text): |
| numbers = re.findall(r"\d+(?:\.?\d+)?", text) |
| for number in numbers: |
| text = text.replace(number, cn2an.an2cn(number), 1) |
| text = replace_punctuation(text) |
| return text |
|
|
| def get_bert_feature( |
| text, |
| word2ph, |
| style_text=None, |
| style_weight=0.7, |
| ): |
| global bert_model |
|
|
| |
| inputs = tokenizer(text, return_tensors="np",padding="max_length",truncation=True,max_length=256) |
| |
| |
| start_time = time.time() |
| res = bert_model.inference([inputs["input_ids"], inputs["attention_mask"], inputs["token_type_ids"]]) |
| flow_time = time.time() - start_time |
| print(f"bert 运行时间: {flow_time:.4f} 秒") |
| |
| |
| res = res[0][0] |
| |
| if style_text: |
| assert False |
| |
| |
| |
| |
| |
| |
| |
| assert len(word2ph) == len(text) + 2 |
| word2phone = word2ph |
| phone_level_feature = [] |
| for i in range(len(word2phone)): |
| if style_text: |
| repeat_feature = ( |
| res[i].repeat(word2phone[i], 1) * (1 - style_weight) |
| |
| ) |
| else: |
| repeat_feature = np.tile(res[i], (word2phone[i], 1)) |
| phone_level_feature.append(repeat_feature) |
|
|
| phone_level_feature = np.concatenate(phone_level_feature, axis=0) |
|
|
| return phone_level_feature.T |
|
|
| def clean_text(text, language): |
| norm_text = text_normalize(text) |
| phones, tones, word2ph = g2p(norm_text) |
| return norm_text, phones, tones, word2ph |
|
|
|
|
| def clean_text_bert(text, language): |
| norm_text = text_normalize(text) |
| phones, tones, word2ph = g2p(norm_text) |
| bert = get_bert_feature(norm_text, word2ph) |
| return phones, tones, bert |
|
|
| _symbol_to_id = {s: i for i, s in enumerate(symbols)} |
|
|
| def cleaned_text_to_sequence(cleaned_text, tones, language): |
| """Converts a string of text to a sequence of IDs corresponding to the symbols in the text. |
| Args: |
| text: string to convert to a sequence |
| Returns: |
| List of integers corresponding to the symbols in the text |
| """ |
| phones = [_symbol_to_id[symbol] for symbol in cleaned_text] |
| tone_start = language_tone_start_map[language] |
| tones = [i + tone_start for i in tones] |
| lang_id = language_id_map[language] |
| lang_ids = [lang_id for i in phones] |
| return phones, tones, lang_ids |
|
|
| def text_to_sequence(text, language): |
| norm_text, phones, tones, word2ph = clean_text(text, language) |
| return cleaned_text_to_sequence(phones, tones, language) |
|
|
| def intersperse(lst, item): |
| result = [item] * (len(lst) * 2 + 1) |
| result[1::2] = lst |
| return result |
|
|
| def get_text(text, language_str, style_text=None, style_weight=0.7, add_blank=False): |
| |
| norm_text, phone, tone, word2ph = clean_text(text, language_str) |
| phone, tone, language = cleaned_text_to_sequence(phone, tone, language_str) |
|
|
| if add_blank: |
| phone = intersperse(phone, 0) |
| tone = intersperse(tone, 0) |
| language = intersperse(language, 0) |
| for i in range(len(word2ph)): |
| word2ph[i] = word2ph[i] * 2 |
| word2ph[0] += 1 |
| bert_ori = get_bert_feature( |
| norm_text, word2ph, style_text, style_weight |
| ) |
| del word2ph |
| assert bert_ori.shape[-1] == len(phone), phone |
|
|
| if language_str == "ZH": |
| bert = bert_ori |
| ja_bert = np.zeros((1024, len(phone))) |
| en_bert = np.zeros((1024, len(phone))) |
| elif language_str == "JP": |
| bert = np.zeros((1024, len(phone))) |
| ja_bert = bert_ori |
| en_bert = np.zeros((1024, len(phone))) |
| elif language_str == "EN": |
| bert = np.zeros((1024, len(phone))) |
| ja_bert = np.zeros((1024, len(phone))) |
| en_bert = bert_ori |
| else: |
| raise ValueError("language_str should be ZH, JP or EN") |
|
|
| assert bert.shape[-1] == len( |
| phone |
| ), f"Bert seq len {bert.shape[-1]} != {len(phone)}" |
| phone = np.array(phone) |
| tone = np.array(tone) |
| language = np.array(language) |
| return bert, ja_bert, en_bert, phone, tone, language |
|
|
| if __name__ == "__main__": |
| name = "lx" |
| model_prefix = f"onnx/{name}/{name}_" |
| bert_path = "./bert/chinese-roberta-wwm-ext-large" |
| flow_dec_input_len = 1024 |
| model_sample_rate = 44100 |
| |
| text = "我个人认为，这个意大利面就应该拌42号混凝土，因为这个螺丝钉的长度，它很容易会直接影响到挖掘机的扭矩你知道吧。你往里砸的时候，一瞬间它就会产生大量的高能蛋白，俗称ufo，会严重影响经济的发展，甚至对整个太平洋以及充电器都会造成一定的核污染。你知道啊？再者说，根据这个勾股定理，你可以很容易地推断出人工饲养的东条英机，它是可以捕获野生的三角函数的。所以说这个秦始皇的切面是否具有放射性啊，特朗普的N次方是否含有沉淀物，都不影响这个沃尔玛跟维尔康在南极会合。" |
|
|
| global bert_model,tokenizer |
| tokenizer = AutoTokenizer.from_pretrained(bert_path) |
| bert_model = RKNNLite(verbose=False) |
| bert_model.load_rknn(bert_path + "/model.rknn") |
| bert_model.init_runtime() |
| model = InferenceSession({ |
| "enc": model_prefix + "enc_p.onnx", |
| "emb_g": model_prefix + "emb.onnx", |
| "dp": model_prefix + "dp.onnx", |
| "sdp": model_prefix + "sdp.onnx", |
| "flow": model_prefix + "flow.onnx", |
| "dec": model_prefix + "dec.rknn", |
| }) |
|
|
| |
| text_seg = re.split(r'(?<=[。！？；])', text) |
| output_acc = np.array([0.0]) |
|
|
| for text in text_seg: |
| bert, ja_bert, en_bert, phone, tone, language = get_text(text, "ZH", add_blank=True) |
| bert = np.transpose(bert) |
| ja_bert = np.transpose(ja_bert) |
| en_bert = np.transpose(en_bert) |
|
|
| sid = np.array([0]) |
| vqidx = np.array([0]) |
|
|
| output = model(phone, tone, language, bert, ja_bert, en_bert, vqidx, sid , |
| rknn_pad_to=flow_dec_input_len, |
| seed=114514, |
| seq_noise_scale=0.8, |
| sdp_noise_scale=0.6, |
| length_scale=1, |
| sdp_ratio=0, |
| )[0,0] |
| output_acc = np.concatenate([output_acc, output]) |
| print(f"已生成长度: {len(output_acc) / model_sample_rate:.2f} 秒") |
| |
| sf.write('output.wav', output_acc, model_sample_rate) |
| print("已生成output.wav") |
