feat: use new g2p tool to replace old pinyin conversion tool

app.py

@@ -7,11 +7,10 @@ import soundfile as sf
 import torch
 import torchcodec
 from cached_path import cached_path
+from formog2p.hakka import g2p
 from omegaconf import OmegaConf
 from torchcodec.decoders import AudioDecoder
 
-from pinyin.pinyin import get_pinyin, pinyin_configs
-
 try:
     import spaces
 
@@ -38,6 +37,45 @@ from f5_tts.infer.utils_infer import (
 from f5_tts.model import CFM, DiT
 from f5_tts.model.utils import get_tokenizer
 
+DIALECT_MAP = {
+    "四縣腔": "客語_四縣",
+    "海陸腔": "客語_海陸",
+    "南四縣腔": "客語_南四縣",
+    "大埔腔": "客語_大埔",
+    "饒平腔": "客語_饒平",
+    "詔安腔": "客語_詔安",
+}
+
+
+def to_pinyin(text, dialect):
+    # Split text by English parts, convert non-English to pinyin, keep English as-is
+    # Pattern matches English words (including spaces between them)
+    pattern = r"([a-zA-Z]+(?:\s+[a-zA-Z]+)*)"
+    parts = re.split(pattern, text)
+
+    result_parts = []
+    for part in parts:
+        if not part:
+            continue
+        # Check if part is English
+        if re.match(r"^[a-zA-Z]+(?:\s+[a-zA-Z]+)*$", part):
+            # Keep English as-is
+            result_parts.append(part)
+        else:
+            # Convert non-English to pinyin
+            result = g2p(part, dialect, pronunciation_type="pinyin")
+            if len(result.unknown_words) > 0:
+                raise gr.Error(
+                    f"參考句子中的[{','.join(result.unknown_words)}]目前無法轉成 pinyin。請嘗試其他句子。"
+                )
+            pinyin = " ".join(result.pronunciations)
+            result_parts.append(pinyin)
+
+    combined = " ".join(result_parts)
+    combined = re.sub(r"[！？]", "。", combined)
+    combined = re.sub(r"(\s?)([，。])(\s?)", r"\2", combined)
+    return combined
+
 
 def gpu_decorator(func):
     if USING_SPACES:
@@ -114,7 +152,6 @@ OmegaConf.register_new_resolver("load_f5tts", load_f5tts)
 models_config = OmegaConf.to_object(OmegaConf.load("configs/models.yaml"))
 
 DEFAULT_MODEL_ID = list(models_config.keys())[0]
-DEFAULT_DIALECT = list(pinyin_configs["lexicon"].keys())[0]
 
 
 @gpu_decorator
@@ -206,8 +243,8 @@ with gr.Blocks(
             )
 
             ref_dialect_radio = gr.Radio(
-                choices=pinyin_configs["lexicon"].keys(),
-                value=DEFAULT_DIALECT,
+                choices=DIALECT_MAP.items(),
+                value="客語_四縣",
                 label="ref 腔調",
             )
 
@@ -225,8 +262,8 @@ with gr.Blocks(
             )
 
             gen_dialect_radio = gr.Radio(
-                choices=pinyin_configs["lexicon"].keys(),
-                value=DEFAULT_DIALECT,
+                choices=DIALECT_MAP.items(),
+                value="客語_四縣",
                 label="gen 腔調",
             )
 
@@ -292,19 +329,12 @@ with gr.Blocks(
         if len(gen_text_input) == 0:
             raise gr.Error("請勿輸入空字串。")
 
-        words, pinyin, missing_words = get_pinyin(ref_text_input, ref_dialect_radio)
-        if len(missing_words) > 0:
-            raise gr.Error(
-                f"參考句子中的[{','.join(missing_words)}]目前無法轉成 pinyin。請嘗試其他句子。"
-            )
-        ref_text_input = pinyin
+        # let text
+        ref_text_input = to_pinyin(ref_text_input, ref_dialect_radio)
+        print(ref_text_input)
 
-        words, pinyin, missing_words = get_pinyin(gen_text_input, gen_dialect_radio)
-        if len(missing_words) > 0:
-            raise gr.Error(
-                f"生成句子中的[{','.join(missing_words)}]目前無法轉成 pinyin。請嘗試其他句子。"
-            )
-        gen_text_input = pinyin
+        gen_text_input = to_pinyin(gen_text_input, gen_dialect_radio)
+        print(gen_text_input)
 
         audio_out, spectrogram_path = infer(
             ref_audio_input,
@@ -340,23 +370,23 @@ with gr.Blocks(
             [
                 "./ref_wav/0000001_0.15-0.93.wav",
                 "恁早。",
-                "sixian",
+                "客語_四縣",
                 "食飯愛正經食，正毋會食到半出半入。",
-                "sixian",
+                "客語_四縣",
             ],
             [
                 "./ref_wav/0000002_0.15-2.73.wav",
                 "你今晡日著到恁派頭。",
-                "sixian",
+                "客語_四縣",
                 "食飯愛正經食，正毋會食到半出半入。",
-                "sixian",
+                "客語_四縣",
             ],
             [
                 "./ref_wav/0000002_0.15-2.73.wav",
                 "你今晡日著到恁派頭。",
-                "sixian",
+                "客語_四縣",
                 "歸條路吊等長長个花燈，祈求風調雨順，歸屋下人个心願，親像花燈下燒暖个光華。",
-                "sixian",
+                "客語_四縣",
             ],
         ],
         label="範例",

configs/pinyin.yaml

@@ -1,11 +0,0 @@
-delimiter_list: ${gh_download:FormoSpeech/FormoG2P, hakka/normalize/delimiters.json}
-replace_dict: ${gh_download:FormoSpeech/FormoG2P, hakka/normalize/replaced_words_htia.json}
-v2f_dict: ${gh_download:FormoSpeech/FormoG2P, [hakka/normalize/v2f_goyu.json, hakka/normalize/v2f_htia.json]}
-preserved_list: ${gh_download:FormoSpeech/FormoG2P, hakka/normalize/preserved_words_htia.json}
-lexicon:
-  sixian: ${gh_download:FormoSpeech/FormoG2P, hakka/sixian.json}
-  hailu: ${gh_download:FormoSpeech/FormoG2P, hakka/hailu.json}
-  dapu: ${gh_download:FormoSpeech/FormoG2P, hakka/dapu.json}
-  nansixian: ${gh_download:FormoSpeech/FormoG2P, hakka/nansixian.json}
-  raoping: ${gh_download:FormoSpeech/FormoG2P, hakka/raoping.json}
-  zhaoan: ${gh_download:FormoSpeech/FormoG2P, hakka/zhaoan.json}

pinyin/__init__.py

@@ -1,27 +0,0 @@
-import time
-
-import requests
-from omegaconf import OmegaConf
-
-
-def gh_download(repo, path):
-    paths = [path] if isinstance(path, str) else path
-    result = None
-    for path in paths:
-        url = f"https://raw.githubusercontent.com/{repo}/refs/heads/main/{path}"
-        response = requests.get(url)
-        if response.status_code != 200:
-            print(f"Status code: {response.status_code}")
-            raise Exception(f"Failed to download {path} from {repo}")
-
-        if result is None:
-            result = response.json()
-        elif isinstance(result, list):
-            result.extend(response.json())
-        elif isinstance(result, dict):
-            result.update(response.json())
-    time.sleep(0.5)
-    return result
-
-
-OmegaConf.register_new_resolver("gh_download", gh_download)

pinyin/convert_digits.py

@@ -1,180 +0,0 @@
-# Copyright 2024    Hung-Shin Lee (hungshinlee@gmail.com)
-# Apache 2.0
-
-import itertools
-import re
-
-c_basic = "零一二三四五六七八九"
-d2c = {str(d): c for d, c in enumerate(c_basic)}
-d2c["."] = "點"
-
-
-def num4year(matched):
-    def _num4year(num):
-        return "{}".format("".join([c_basic[int(i)] for i in num]))
-
-    matched_str = matched.group(0)
-    for m in matched.groups():
-        matched_str = matched_str.replace(m, _num4year(m))
-    return matched_str
-
-
-def num2chines_simple(matched):
-    return "{}".format("".join([d2c[i] for i in matched]))
-
-
-def num4percent(matched):
-    matched = matched.group(1)
-    return "百分之{}".format(num2chinese(matched[:-1]))
-
-
-def num4cellphone(matched):
-    matched = matched.group(1)
-    matched = matched.replace(" ", "").replace("-", "")
-    return "".join([c_basic[int(i)] for i in matched])
-
-
-def num4er(matched):  # 2 to 二
-    matched = matched.group(1)
-    return matched.replace("2", "二")
-
-
-def num4liang(matched):  # 2 to 兩
-    matched = matched.group(1)
-    return matched.replace("2", "兩")
-
-
-def num4general(matched):
-    num = matched.group(1)
-    if re.match(r"[A-Za-z-─]", num[0]):
-        if len(num[1:]) < 3:
-            # MP3 or F-16
-            return "{}{}".format(num[0], num2chinese(num[1:]))
-        else:
-            # AM104
-            return "{}{}".format(num[0], num2chines_simple(num[1:]))
-
-    else:
-        if re.match(r"[0-9]", num[0]):
-            return "{}".format(num2chinese(num))
-        else:
-            return "{}{}".format(num[0], num2chinese(num[1:]))
-
-
-def parse_num(text: str) -> str:
-    # year
-    text = re.sub(r"([0-9]{4})[到至]([0-9]{4})年", num4year, text)
-    text = re.sub(r"([0-9]{4})年", num4year, text)
-
-    # percentage
-    text = re.sub(r"([0-9]+\.?[0-9]?%)", num4percent, text)
-
-    # cellphone
-    text = re.sub(r"([0-9]{4}\s?-\s?[0-9]{6})", num4cellphone, text)
-
-    # single 2 to 二
-    text = re.sub(r"([^\d]2[診樓月號])", num4er, text)
-    text = re.sub(r"([初]2[^\d])", num4er, text)
-
-    # single 2 to 兩
-    text = re.sub(r"([^\d]2[^\d])", num4liang, text)
-
-    # general number
-    text = re.sub(r"([^0-9]?[0-9]+\.?[0-9]?)", num4general, text)
-
-    return text
-
-
-def num2chinese(num, big=False, simp=False, o=False, twoalt=True) -> str:
-    """
-    Converts numbers to Chinese representations.
-    https://gist.github.com/gumblex/0d65cad2ba607fd14de7
-    `big`   : use financial characters.
-    `simp`  : use simplified characters instead of traditional characters.
-    `o`     : use 〇 for zero.
-    `twoalt`: use 两/兩 for two when appropriate.
-    Note that `o` and `twoalt` is ignored when `big` is used,
-    and `twoalt` is ignored when `o` is used for formal representations.
-    """
-    # check num first
-    nd = str(num)
-    if abs(float(nd)) >= 1e48:
-        raise ValueError("number out of range")
-    elif "e" in nd:
-        raise ValueError("scientific notation is not supported")
-    c_symbol = "正负点" if simp else "正負點"
-    if o:  # formal
-        twoalt = False
-    if big:
-        c_basic = "零壹贰叁肆伍陆柒捌玖" if simp else "零壹貳參肆伍陸柒捌玖"
-        c_unit1 = "拾佰仟"
-        c_twoalt = "贰" if simp else "貳"
-    else:
-        c_basic = "〇一二三四五六七八九" if o else "零一二三四五六七八九"
-        c_unit1 = "十百千"
-        if twoalt:
-            c_twoalt = "两" if simp else "兩"
-        else:
-            c_twoalt = "二"
-    c_unit2 = "万亿兆京垓秭穰沟涧正载" if simp else "萬億兆京垓秭穰溝澗正載"
-
-    def revuniq(l):
-        return "".join(k for k, g in itertools.groupby(reversed(l)))
-
-    nd = str(num)
-    result = []
-    if nd[0] == "+":
-        result.append(c_symbol[0])
-    elif nd[0] == "-":
-        result.append(c_symbol[1])
-    if "." in nd:
-        integer, remainder = nd.lstrip("+-").split(".")
-    else:
-        integer, remainder = nd.lstrip("+-"), None
-    if int(integer):
-        splitted = [integer[max(i - 4, 0) : i] for i in range(len(integer), 0, -4)]
-        intresult = []
-        for nu, unit in enumerate(splitted):
-            # special cases
-            if int(unit) == 0:  # 0000
-                intresult.append(c_basic[0])
-                continue
-            elif nu > 0 and int(unit) == 2:  # 0002
-                intresult.append(c_twoalt + c_unit2[nu - 1])
-                continue
-            ulist = []
-            unit = unit.zfill(4)
-            for nc, ch in enumerate(reversed(unit)):
-                if ch == "0":
-                    if ulist:  # ???0
-                        ulist.append(c_basic[0])
-                elif nc == 0:
-                    ulist.append(c_basic[int(ch)])
-                elif nc == 1 and ch == "1" and all([i == "0" for i in unit[: nc + 1]]):
-                    # special case for tens
-                    # edit the 'elif' if you don't like
-                    # 十四, 三千零十四, 三千三百一十四
-                    ulist.append(c_unit1[0])
-                elif nc > 1 and ch == "2":
-                    ulist.append(c_twoalt + c_unit1[nc - 1])
-                else:
-                    ulist.append(c_basic[int(ch)] + c_unit1[nc - 1])
-            # print(ulist)
-            ustr = revuniq(ulist)
-            if nu == 0:
-                intresult.append(ustr)
-            else:
-                intresult.append(ustr + c_unit2[nu - 1])
-        result.append(revuniq(intresult).strip(c_basic[0]))
-    else:
-        result.append(c_basic[0])
-    if remainder:
-        result.append(c_symbol[2])
-        result.append("".join(c_basic[int(ch)] for ch in remainder))
-    return "".join(result)
-
-
-if __name__ == "__main__":
-    text = "若手機仔幾多號？吾手機仔係0964-498042。"
-
-    print(f"{text} -> {parse_num(text)}")

pinyin/pinyin.py

@@ -1,90 +0,0 @@
-import os
-import re
-from pathlib import Path
-
-import jieba
-from omegaconf import OmegaConf
-
-from pinyin.convert_digits import parse_num
-from pinyin.proc_text import (
-    apply_v2f,
-    normalize_text,
-    prep_regex,
-    run_jieba,
-    update_jieba_dict,
-)
-
-pinyin_configs = OmegaConf.to_object(OmegaConf.load("configs/pinyin.yaml"))
-for key in pinyin_configs["preserved_list"]:
-    pinyin_configs["v2f_dict"].pop(key, None)
-delimiter_regex, replace_regex, v2f_regex = prep_regex(
-    pinyin_configs["delimiter_list"],
-    pinyin_configs["replace_dict"],
-    pinyin_configs["v2f_dict"],
-)
-
-
-def get_pinyin(raw_text: str, dialect: str) -> tuple[str, str, str, list[str]]:
-    pinyin_split = re.split(r"([a-z]+\d+)", raw_text)
-
-    final_words = []
-    final_pinyin = []
-    final_missing_words = []
-    for hanzi_or_pinyin in pinyin_split:
-        if len(hanzi_or_pinyin.strip()) == 0:
-            continue
-
-        if re.search(r"[a-z]+\d+", hanzi_or_pinyin):
-            final_words.append(hanzi_or_pinyin)
-            final_pinyin.append(hanzi_or_pinyin)
-        else:
-            words, pinyin, missing_words = parse_hanzi_to_pinyin(
-                hanzi_or_pinyin, dialect
-            )
-            final_words.extend(words)
-            final_pinyin.extend(pinyin)
-            final_missing_words.extend(missing_words)
-
-    if len(final_pinyin) == 0 or len(final_missing_words) > 0:
-        return final_words, final_pinyin, final_missing_words
-
-    final_words = " ".join(final_words).replace(" ， ", "，") + "。"
-    final_pinyin = " ".join(final_pinyin).replace(" ， ", "，") + "。"
-
-    return final_words, final_pinyin, final_missing_words
-
-
-def parse_hanzi_to_pinyin(
-    hanzi: str, dialect: str
-) -> tuple[list[str], list[str], list[str], list[str]]:
-    lexicon = pinyin_configs["lexicon"][dialect]
-    update_jieba_dict(
-        list(lexicon.keys()), Path(os.path.dirname(jieba.__file__)) / "dict.txt"
-    )
-
-    text = normalize_text(hanzi, pinyin_configs["replace_dict"], replace_regex)
-    text = parse_num(text)
-    text_parts = [s.strip() for s in re.split(delimiter_regex, text) if s.strip()]
-    text = "，".join(text_parts)
-    word_list = run_jieba(text)
-    word_list = apply_v2f(word_list, pinyin_configs["v2f_dict"], v2f_regex)
-    word_list = run_jieba("".join(word_list))
-
-    final_words = []
-    final_pinyin = []
-    missing_words = []
-    for word in word_list:
-        if not bool(word.strip()):
-            continue
-        if word == "，":
-            final_words.append("，")
-            final_pinyin.append("，")
-        elif word not in lexicon:
-            final_words.append(word)
-            missing_words.append(word)
-        else:
-            final_words.append(f"{word}")
-            # NOTE 只有 lexicon[word] 中的第一個才被考慮
-            final_pinyin.append(lexicon[word]["pinyin"][0])
-
-    return final_words, final_pinyin, missing_words

pinyin/proc_text.py

@@ -1,84 +0,0 @@
-# Copyright 2024    Hung-Shin Lee (hungshinlee@gmail.com)
-# Apache 2.0
-
-import re
-from pathlib import Path
-from unicodedata import normalize
-
-import jieba
-import opencc
-
-jieba.setLogLevel(20)
-jieba.re_han_default = re.compile(r"([\u2e80-\U000e01efa-zA-Z0-9+#&\._%\-']+)", re.U)
-
-s2tw_converter = opencc.OpenCC("s2tw.json")
-
-
-def update_jieba_dict(
-    lexicon: list,
-    jieba_dict_path: Path,
-    high_freq_words: list = [],
-    high_freq_words_weight: int = 10,
-) -> list:
-    lexicon = sorted(set(lexicon))
-
-    jieba_dict_path.unlink(missing_ok=True)
-    Path("/tmp/jieba.cache").unlink(missing_ok=True)
-
-    with jieba_dict_path.open("w", encoding="utf-8") as file:
-        for word in lexicon:
-            if word in high_freq_words:
-                file.write(f"{word} {len(word) * high_freq_words_weight}\n")
-            else:
-                file.write(f"{word} {len(word)}\n")
-
-    jieba.dt.initialized = False
-
-    return lexicon
-
-
-def run_jieba(line: str) -> list:
-    # NOTE JIEBA 處理多行文本的結果會失去原本的行結構
-
-    seg_list = list(jieba.cut(line, cut_all=False, HMM=False))
-
-    return seg_list
-
-
-def normalize_text(text: str, replace_dict: dict, replace_regex: str) -> str:
-    def replace_match(match):
-        return replace_dict[match.group(0)]
-
-    text = re.sub(r"\x08", "", text)
-    text = re.sub(r"\ufeff", "", text)
-    text = re.sub(r"\u0010", "", text)
-    text = normalize("NFKC", text)
-    text = re.sub(replace_regex, replace_match, text)
-    text = " ".join(text.split()).upper()
-
-    return text
-
-
-def apply_v2f(word_list: list, v2f_dict: dict, v2f_regex: str) -> list:
-    result = []
-    for word in word_list:
-        result.append(re.sub(v2f_regex, lambda x: v2f_dict[x.group(0)], word))
-
-    return result
-
-
-def prep_regex(
-    delimiter_list: list, replace_dict: dict = {}, v2f_dict: dict = {}
-) -> tuple[str, str, str]:
-    delimiter_regex = "|".join(map(re.escape, delimiter_list))
-
-    replace_regex = ""
-    if len(replace_dict):
-        sorted_keys = sorted(replace_dict.keys(), key=len, reverse=True)
-        replace_regex = "|".join(map(re.escape, sorted_keys))
-
-    v2f_regex = ""
-    if len(v2f_dict):
-        v2f_regex = "|".join(map(re.escape, v2f_dict.keys()))
-
-    return delimiter_regex, replace_regex, v2f_regex

requirements.txt

@@ -1,4 +1,5 @@
 omegaconf
 opencc
 f5-tts
-jieba
+jieba
+formog2p