tuna2134/sbv2-v2
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

Refactor: rename text_processing to nlp

Browse Source 
"text_processing" is clearer, but the import statement is longer.
"nlp" is shorter and makes it clear that it is natural language processing.
This commit is contained in:
tsukumi
2024-03-08 06:20:44 +00:00
parent 8add1b4202
commit fac4f9a8ab
38 changed files with 54 additions and 54 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
style_bert_vits2/nlp/japanese/__init__.py Normal file
View File

@@ -0,0 +1,2 @@
from style_bert_vits2.nlp.japanese.g2p import g2p # noqa: F401
from style_bert_vits2.nlp.japanese.normalizer import normalize_text # noqa: F401

87
style_bert_vits2/nlp/japanese/bert_feature.py Normal file
View File

@@ -0,0 +1,87 @@
import sys
from typing import Optional
import torch
from transformers import PreTrainedModel
from style_bert_vits2.constants import Languages
from style_bert_vits2.nlp import bert_models
from style_bert_vits2.nlp.japanese.g2p import text_to_sep_kata
__models: dict[torch.device | str, PreTrainedModel] = {}
def extract_bert_feature(
text: str,
word2ph: list[int],
device: torch.device | str,
assist_text: Optional[str] = None,
assist_text_weight: float = 0.7,
) -> torch.Tensor:
"""
日本語のテキストから BERT の特徴量を抽出する
Args:
text (str): 日本語のテキスト
word2ph (list[int]): 元のテキストの各文字に音素が何個割り当てられるかを表すリスト
device (torch.device | str): 推論に利用するデバイス
assist_text (Optional[str], optional): 補助テキスト (デフォルト: None)
assist_text_weight (float, optional): 補助テキストの重み (デフォルト: 0.7)
Returns:
torch.Tensor: BERT の特徴量
"""
# 各単語が何文字かを作る `word2ph` を使う必要があるので、読めない文字は必ず無視する
# でないと `word2ph` の結果とテキストの文字数結果が整合性が取れない
text = "".join(text_to_sep_kata(text, raise_yomi_error=False)[0])
if assist_text:
assist_text = "".join(text_to_sep_kata(assist_text, raise_yomi_error=False)[0])
if (
sys.platform == "darwin"
and torch.backends.mps.is_available()
and device == "cpu"
):
device = "mps"
if not device:
device = "cuda"
if device == "cuda" and not torch.cuda.is_available():
device = "cpu"
if device not in __models.keys():
__models[device] = bert_models.load_model(Languages.JP).to(device) # type: ignore
style_res_mean = None
with torch.no_grad():
tokenizer = bert_models.load_tokenizer(Languages.JP)
inputs = tokenizer(text, return_tensors="pt")
for i in inputs:
inputs[i] = inputs[i].to(device) # type: ignore
res = __models[device](**inputs, output_hidden_states=True)
res = torch.cat(res["hidden_states"][-3:-2], -1)[0].cpu()
if assist_text:
style_inputs = tokenizer(assist_text, return_tensors="pt")
for i in style_inputs:
style_inputs[i] = style_inputs[i].to(device) # type: ignore
style_res = __models[device](**style_inputs, output_hidden_states=True)
style_res = torch.cat(style_res["hidden_states"][-3:-2], -1)[0].cpu()
style_res_mean = style_res.mean(0)
assert len(word2ph) == len(text) + 2, text
word2phone = word2ph
phone_level_feature = []
for i in range(len(word2phone)):
if assist_text:
assert style_res_mean is not None
repeat_feature = (
res[i].repeat(word2phone[i], 1) * (1 - assist_text_weight)
+ style_res_mean.repeat(word2phone[i], 1) * assist_text_weight
)
else:
repeat_feature = res[i].repeat(word2phone[i], 1)
phone_level_feature.append(repeat_feature)
phone_level_feature = torch.cat(phone_level_feature, dim=0)
return phone_level_feature.T

493
style_bert_vits2/nlp/japanese/g2p.py Normal file
View File

@@ -0,0 +1,493 @@
import re
from style_bert_vits2.constants import Languages
from style_bert_vits2.logging import logger
from style_bert_vits2.nlp import bert_models
from style_bert_vits2.nlp.japanese import pyopenjtalk_worker as pyopenjtalk
from style_bert_vits2.nlp.japanese.mora_list import MORA_KATA_TO_MORA_PHONEMES
from style_bert_vits2.nlp.japanese.normalizer import replace_punctuation
from style_bert_vits2.nlp.symbols import PUNCTUATIONS
pyopenjtalk.initialize()
def g2p(
norm_text: str,
use_jp_extra: bool = True,
raise_yomi_error: bool = False
) -> tuple[list[str], list[int], list[int]]:
"""
他で使われるメインの関数。`normalize_text()` で正規化された `norm_text` を受け取り、
- phones: 音素のリスト（ただし `!` や `,` や `.` など punctuation が含まれうる）
- tones: アクセントのリスト、0（低）と1（高）からなり、phones と同じ長さ
- word2ph: 元のテキストの各文字に音素が何個割り当てられるかを表すリスト
のタプルを返す。
ただし `phones` と `tones` の最初と終わりに `_` が入り、応じて `word2ph` の最初と最後に 1 が追加される。
Args:
norm_text (str): 正規化されたテキスト
use_jp_extra (bool, optional): False の場合、「ん」の音素を「N」ではなく「n」とする。Defaults to True.
raise_yomi_error (bool, optional): False の場合、読めない文字が消えたような扱いとして処理される。Defaults to False.
Returns:
tuple[list[str], list[int], list[int]]: 音素のリスト、アクセントのリスト、word2ph のリスト
"""
# pyopenjtalk のフルコンテキストラベルを使ってアクセントを取り出すと、punctuation の位置が消えてしまい情報が失われてしまう：
# 「こんにちは、世界。」と「こんにちは！世界。」と「こんにちは！！！？？？世界……。」は全て同じになる。
# よって、まず punctuation 無しの音素とアクセントのリストを作り、
# それとは別に pyopenjtalk.run_frontend() で得られる音素リスト（こちらは punctuation が保持される）を使い、
# アクセント割当をしなおすことによって punctuation を含めた音素とアクセントのリストを作る。
# punctuation がすべて消えた、音素とアクセントのタプルのリスト（「ん」は「N」）
phone_tone_list_wo_punct = __g2phone_tone_wo_punct(norm_text)
# sep_text: 単語単位の単語のリスト、読めない文字があったら raise_yomi_error なら例外、そうでないなら読めない文字が消えて返ってくる
# sep_kata: 単語単位の単語のカタカナ読みのリスト
sep_text, sep_kata = text_to_sep_kata(norm_text, raise_yomi_error=raise_yomi_error)
# sep_phonemes: 各単語ごとの音素のリストのリスト
sep_phonemes = __handle_long([__kata_to_phoneme_list(i) for i in sep_kata])
# phone_w_punct: sep_phonemes を結合した、punctuation を元のまま保持した音素列
phone_w_punct: list[str] = []
for i in sep_phonemes:
phone_w_punct += i
# punctuation 無しのアクセント情報を使って、punctuation を含めたアクセント情報を作る
phone_tone_list = __align_tones(phone_w_punct, phone_tone_list_wo_punct)
# logger.debug(f"phone_tone_list:\n{phone_tone_list}")
# word2ph は厳密な解答は不可能なので（「今日」「眼鏡」等の熟字訓が存在）、
# Bert-VITS2 では、単語単位の分割を使って、単語の文字ごとにだいたい均等に音素を分配する
# sep_text から、各単語を1文字1文字分割して、文字のリスト（のリスト）を作る
sep_tokenized: list[list[str]] = []
for i in sep_text:
if i not in PUNCTUATIONS:
sep_tokenized.append(
bert_models.load_tokenizer(Languages.JP).tokenize(i)
) # ここでおそらく`i`が文字単位に分割される
else:
sep_tokenized.append([i])
# 各単語について、音素の数と文字の数を比較して、均等っぽく分配する
word2ph = []
for token, phoneme in zip(sep_tokenized, sep_phonemes):
phone_len = len(phoneme)
word_len = len(token)
word2ph += __distribute_phone(phone_len, word_len)
# 最初と最後に `_` 記号を追加、アクセントは 0（低）、word2ph もそれに合わせて追加
phone_tone_list = [("_", 0)] + phone_tone_list + [("_", 0)]
word2ph = [1] + word2ph + [1]
phones = [phone for phone, _ in phone_tone_list]
tones = [tone for _, tone in phone_tone_list]
assert len(phones) == sum(word2ph), f"{len(phones)} != {sum(word2ph)}"
# use_jp_extra でない場合は「N」を「n」に変換
if not use_jp_extra:
phones = [phone if phone != "N" else "n" for phone in phones]
return phones, tones, word2ph
def text_to_sep_kata(
norm_text: str,
raise_yomi_error: bool = False
) -> tuple[list[str], list[str]]:
"""
`normalize_text` で正規化済みの `norm_text` を受け取り、それを単語分割し、
分割された単語リストとその読み（カタカナ or 記号1文字）のリストのタプルを返す。
単語分割結果は、`g2p()` の `word2ph` で1文字あたりに割り振る音素記号の数を決めるために使う。
例:
`私はそう思う!って感じ?` →
["私", "は", "そう", "思う", "!", "って", "感じ", "?"], ["ワタシ", "ワ", "ソー", "オモウ", "!", "ッテ", "カンジ", "?"]
Args:
norm_text (str): 正規化されたテキスト
raise_yomi_error (bool, optional): False の場合、読めない文字が消えたような扱いとして処理される。Defaults to False.
Returns:
tuple[list[str], list[str]]: 分割された単語リストと、その読み（カタカナ or 記号1文字）のリスト
"""
# parsed: OpenJTalkの解析結果
parsed = pyopenjtalk.run_frontend(norm_text)
sep_text: list[str] = []
sep_kata: list[str] = []
for parts in parsed:
# word: 実際の単語の文字列
# yomi: その読み、但し無声化サインの`’`は除去
word, yomi = replace_punctuation(parts["string"]), parts["pron"].replace(
"’", ""
)
"""
ここで `yomi` の取りうる値は以下の通りのはず。
- `word` が通常単語 → 通常の読み（カタカナ）
（カタカナからなり、長音記号も含みうる、`アー` 等）
- `word` が `ー` から始まる → `ーラー` や `ーーー` など
- `word` が句読点や空白等 → `、`
- `word` が punctuation の繰り返し → 全角にしたもの
基本的に punctuation は1文字ずつ分かれるが、何故かある程度連続すると1つにまとまる。
他にも `word` が読めないキリル文字アラビア文字等が来ると `、` になるが、正規化でこの場合は起きないはず。
また元のコードでは `yomi` が空白の場合の処理があったが、これは起きないはず。
処理すべきは `yomi` が `、` の場合のみのはず。
"""
assert yomi != "", f"Empty yomi: {word}"
if yomi == "、":
# word は正規化されているので、`.`, `,`, `!`, `'`, `-`, `--` のいずれか
if not set(word).issubset(set(PUNCTUATIONS)): # 記号繰り返しか判定
# ここは pyopenjtalk が読めない文字等のときに起こる
if raise_yomi_error:
raise YomiError(f"Cannot read: {word} in:\n{norm_text}")
logger.warning(f"Ignoring unknown: {word} in:\n{norm_text}")
continue
# yomi は元の記号のままに変更
yomi = word
elif yomi == "？":
assert word == "?", f"yomi `？` comes from: {word}"
yomi = "?"
sep_text.append(word)
sep_kata.append(yomi)
return sep_text, sep_kata
def __g2phone_tone_wo_punct(text: str) -> list[tuple[str, int]]:
"""
テキストに対して、音素とアクセント（0か1）のペアのリストを返す。
ただし「!」「.」「?」等の非音素記号 (punctuation) は全て消える（ポーズ記号も残さない）。
非音素記号を含める処理は `align_tones()` で行われる。
また「っ」は「q」に、「ん」は「N」に変換される。
例: "こんにちは、世界ー。。元気？！" →
[('k', 0), ('o', 0), ('N', 1), ('n', 1), ('i', 1), ('ch', 1), ('i', 1), ('w', 1), ('a', 1), ('s', 1), ('e', 1), ('k', 0), ('a', 0), ('i', 0), ('i', 0), ('g', 1), ('e', 1), ('N', 0), ('k', 0), ('i', 0)]
Args:
text (str): テキスト
Returns:
list[tuple[str, int]]: 音素とアクセントのペアのリスト
"""
prosodies = __pyopenjtalk_g2p_prosody(text, drop_unvoiced_vowels=True)
# logger.debug(f"prosodies: {prosodies}")
result: list[tuple[str, int]] = []
current_phrase: list[tuple[str, int]] = []
current_tone = 0
for i, letter in enumerate(prosodies):
# 特殊記号の処理
# 文頭記号、無視する
if letter == "^":
assert i == 0, "Unexpected ^"
# アクセント句の終わりに来る記号
elif letter in ("$", "?", "_", "#"):
# 保持しているフレーズを、アクセント数値を 0-1 に修正し結果に追加
result.extend(__fix_phone_tone(current_phrase))
# 末尾に来る終了記号、無視（文中の疑問文は `_` になる）
if letter in ("$", "?"):
assert i == len(prosodies) - 1, f"Unexpected {letter}"
# あとは "_"（ポーズ）と "#"（アクセント句の境界）のみ
# これらは残さず、次のアクセント句に備える。
current_phrase = []
# 0 を基準点にしてそこから上昇・下降する（負の場合は上の `fix_phone_tone` で直る）
current_tone = 0
# アクセント上昇記号
elif letter == "[":
current_tone = current_tone + 1
# アクセント下降記号
elif letter == "]":
current_tone = current_tone - 1
# それ以外は通常の音素
else:
if letter == "cl": # 「っ」の処理
letter = "q"
# elif letter == "N": # 「ん」の処理
# letter = "n"
current_phrase.append((letter, current_tone))
return result
def __pyopenjtalk_g2p_prosody(text: str, drop_unvoiced_vowels: bool = True) -> list[str]:
"""
ESPnet の実装から引用、変更点無し。「ん」は「N」なことに注意。
ref: https://github.com/espnet/espnet/blob/master/espnet2/text/phoneme_tokenizer.py
------------------------------------------------------------------------------------------
Extract phoneme + prosoody symbol sequence from input full-context labels.
The algorithm is based on `Prosodic features control by symbols as input of
sequence-to-sequence acoustic modeling for neural TTS`_ with some r9y9's tweaks.
Args:
text (str): Input text.
drop_unvoiced_vowels (bool): whether to drop unvoiced vowels.
Returns:
List[str]: List of phoneme + prosody symbols.
Examples:
>>> from espnet2.text.phoneme_tokenizer import pyopenjtalk_g2p_prosody
>>> pyopenjtalk_g2p_prosody("こんにちは。")
['^', 'k', 'o', '[', 'N', 'n', 'i', 'ch', 'i', 'w', 'a', '$']
.. _`Prosodic features control by symbols as input of sequence-to-sequence acoustic
modeling for neural TTS`: https://doi.org/10.1587/transinf.2020EDP7104
"""
def _numeric_feature_by_regex(regex: str, s: str) -> int:
match = re.search(regex, s)
if match is None:
return -50
return int(match.group(1))
labels = pyopenjtalk.make_label(pyopenjtalk.run_frontend(text))
N = len(labels)
phones = []
for n in range(N):
lab_curr = labels[n]
# current phoneme
p3 = re.search(r"\-(.*?)\+", lab_curr).group(1) # type: ignore
# deal unvoiced vowels as normal vowels
if drop_unvoiced_vowels and p3 in "AEIOU":
p3 = p3.lower()
# deal with sil at the beginning and the end of text
if p3 == "sil":
assert n == 0 or n == N - 1
if n == 0:
phones.append("^")
elif n == N - 1:
# check question form or not
e3 = _numeric_feature_by_regex(r"!(\d+)_", lab_curr)
if e3 == 0:
phones.append("$")
elif e3 == 1:
phones.append("?")
continue
elif p3 == "pau":
phones.append("_")
continue
else:
phones.append(p3)
# accent type and position info (forward or backward)
a1 = _numeric_feature_by_regex(r"/A:([0-9\-]+)\+", lab_curr)
a2 = _numeric_feature_by_regex(r"\+(\d+)\+", lab_curr)
a3 = _numeric_feature_by_regex(r"\+(\d+)/", lab_curr)
# number of mora in accent phrase
f1 = _numeric_feature_by_regex(r"/F:(\d+)_", lab_curr)
a2_next = _numeric_feature_by_regex(r"\+(\d+)\+", labels[n + 1])
# accent phrase border
if a3 == 1 and a2_next == 1 and p3 in "aeiouAEIOUNcl":
phones.append("#")
# pitch falling
elif a1 == 0 and a2_next == a2 + 1 and a2 != f1:
phones.append("]")
# pitch rising
elif a2 == 1 and a2_next == 2:
phones.append("[")
return phones
def __fix_phone_tone(phone_tone_list: list[tuple[str, int]]) -> list[tuple[str, int]]:
"""
`phone_tone_list` の tone（アクセントの値）を 0 か 1 の範囲に修正する。
例: [(a, 0), (i, -1), (u, -1)] → [(a, 1), (i, 0), (u, 0)]
Args:
phone_tone_list (list[tuple[str, int]]): 音素とアクセントのペアのリスト
Returns:
list[tuple[str, int]]: 修正された音素とアクセントのペアのリスト
"""
tone_values = set(tone for _, tone in phone_tone_list)
if len(tone_values) == 1:
assert tone_values == {0}, tone_values
return phone_tone_list
elif len(tone_values) == 2:
if tone_values == {0, 1}:
return phone_tone_list
elif tone_values == {-1, 0}:
return [
(letter, 0 if tone == -1 else 1) for letter, tone in phone_tone_list
]
else:
raise ValueError(f"Unexpected tone values: {tone_values}")
else:
raise ValueError(f"Unexpected tone values: {tone_values}")
def __handle_long(sep_phonemes: list[list[str]]) -> list[list[str]]:
"""
フレーズごとに分かれた音素（長音記号がそのまま）のリストのリスト `sep_phonemes` を受け取り、
その長音記号を処理して、音素のリストのリストを返す。
基本的には直前の音素を伸ばすが、直前の音素が母音でない場合もしくは冒頭の場合は、
おそらく長音記号とダッシュを勘違いしていると思われるので、ダッシュに対応する音素 `-` に変換する。
Args:
sep_phonemes (list[list[str]]): フレーズごとに分かれた音素のリストのリスト
Returns:
list[list[str]]: 長音記号を処理した音素のリストのリスト
"""
# 母音の集合 (便宜上「ん」を含める)
VOWELS = {"a", "i", "u", "e", "o", "N"}
for i in range(len(sep_phonemes)):
if len(sep_phonemes[i]) == 0:
# 空白文字等でリストが空の場合
continue
if sep_phonemes[i][0] == "ー":
if i != 0:
prev_phoneme = sep_phonemes[i - 1][-1]
if prev_phoneme in VOWELS:
# 母音と「ん」のあとの伸ばし棒なので、その母音に変換
sep_phonemes[i][0] = sep_phonemes[i - 1][-1]
else:
# 「。ーー」等おそらく予期しない長音記号
# ダッシュの勘違いだと思われる
sep_phonemes[i][0] = "-"
else:
# 冒頭に長音記号が来ていおり、これはダッシュの勘違いと思われる
sep_phonemes[i][0] = "-"
if "ー" in sep_phonemes[i]:
for j in range(len(sep_phonemes[i])):
if sep_phonemes[i][j] == "ー":
sep_phonemes[i][j] = sep_phonemes[i][j - 1][-1]
return sep_phonemes
def __kata_to_phoneme_list(text: str) -> list[str]:
"""
原則カタカナの `text` を受け取り、それをそのままいじらずに音素記号のリストに変換。
注意点：
- punctuation かその繰り返しが来た場合、punctuation たちをそのままリストにして返す。
- 冒頭に続く「ー」はそのまま「ー」のままにする（`handle_long()` で処理される）
- 文中の「ー」は前の音素記号の最後の音素記号に変換される。
例：
`ーーソーナノカーー` → ["ー", "ー", "s", "o", "o", "n", "a", "n", "o", "k", "a", "a", "a"]
`?` → ["?"]
`!?!?!?!?!` → ["!", "?", "!", "?", "!", "?", "!", "?", "!"]
Args:
text (str): カタカナのテキスト
Returns:
list[str]: 音素記号のリスト
"""
if set(text).issubset(set(PUNCTUATIONS)):
return list(text)
# `text` がカタカナ（`ー`含む）のみからなるかどうかをチェック
if re.fullmatch(r"[\u30A0-\u30FF]+", text) is None:
raise ValueError(f"Input must be katakana only: {text}")
sorted_keys = sorted(MORA_KATA_TO_MORA_PHONEMES.keys(), key=len, reverse=True)
pattern = "|".join(map(re.escape, sorted_keys))
def mora2phonemes(mora: str) -> str:
cosonant, vowel = MORA_KATA_TO_MORA_PHONEMES[mora]
if cosonant is None:
return f" {vowel}"
return f" {cosonant} {vowel}"
spaced_phonemes = re.sub(pattern, lambda m: mora2phonemes(m.group()), text)
# 長音記号「ー」の処理
long_pattern = r"(\w)(ー*)"
long_replacement = lambda m: m.group(1) + (" " + m.group(1)) * len(m.group(2)) # type: ignore
spaced_phonemes = re.sub(long_pattern, long_replacement, spaced_phonemes)
return spaced_phonemes.strip().split(" ")
def __align_tones(
phones_with_punct: list[str],
phone_tone_list: list[tuple[str, int]]
) -> list[tuple[str, int]]:
"""
例: …私は、、そう思う。
phones_with_punct:
[".", ".", ".", "w", "a", "t", "a", "sh", "i", "w", "a", ",", ",", "s", "o", "o", "o", "m", "o", "u", "."]
phone_tone_list:
[("w", 0), ("a", 0), ("t", 1), ("a", 1), ("sh", 1), ("i", 1), ("w", 1), ("a", 1), ("_", 0), ("s", 0), ("o", 0), ("o", 1), ("o", 1), ("m", 1), ("o", 1), ("u", 0))]
Return:
[(".", 0), (".", 0), (".", 0), ("w", 0), ("a", 0), ("t", 1), ("a", 1), ("sh", 1), ("i", 1), ("w", 1), ("a", 1), (",", 0), (",", 0), ("s", 0), ("o", 0), ("o", 1), ("o", 1), ("m", 1), ("o", 1), ("u", 0), (".", 0)]
Args:
phones_with_punct (list[str]): punctuation を含む音素のリスト
phone_tone_list (list[tuple[str, int]]): punctuation を含まない音素とアクセントのペアのリスト
Returns:
list[tuple[str, int]]: punctuation を含む音素とアクセントのペアのリスト
"""
result: list[tuple[str, int]] = []
tone_index = 0
for phone in phones_with_punct:
if tone_index >= len(phone_tone_list):
# 余った punctuation がある場合 → (punctuation, 0) を追加
result.append((phone, 0))
elif phone == phone_tone_list[tone_index][0]:
# phone_tone_list の現在の音素と一致する場合 → tone をそこから取得、(phone, tone) を追加
result.append((phone, phone_tone_list[tone_index][1]))
# 探す index を1つ進める
tone_index += 1
elif phone in PUNCTUATIONS:
# phone が punctuation の場合 → (phone, 0) を追加
result.append((phone, 0))
else:
logger.debug(f"phones: {phones_with_punct}")
logger.debug(f"phone_tone_list: {phone_tone_list}")
logger.debug(f"result: {result}")
logger.debug(f"tone_index: {tone_index}")
logger.debug(f"phone: {phone}")
raise ValueError(f"Unexpected phone: {phone}")
return result
def __distribute_phone(n_phone: int, n_word: int) -> list[int]:
"""
左から右に 1 ずつ振り分け、次にまた左から右に1ずつ増やし、というふうに、
音素の数 `n_phone` を単語の数 `n_word` に分配する。
Args:
n_phone (int): 音素の数
n_word (int): 単語の数
Returns:
list[int]: 単語ごとの音素の数のリスト
"""
phones_per_word = [0] * n_word
for _ in range(n_phone):
min_tasks = min(phones_per_word)
min_index = phones_per_word.index(min_tasks)
phones_per_word[min_index] += 1
return phones_per_word
class YomiError(Exception):
"""
OpenJTalk で、読みが正しく取得できない箇所があるときに発生する例外。
基本的に「学習の前処理のテキスト処理時」には発生させ、そうでない場合は、
ignore_yomi_error=True にしておいて、この例外を発生させないようにする。
"""
pass

90
style_bert_vits2/nlp/japanese/g2p_utils.py Normal file
View File

@@ -0,0 +1,90 @@
from style_bert_vits2.nlp.japanese.g2p import g2p
from style_bert_vits2.nlp.japanese.mora_list import (
MORA_KATA_TO_MORA_PHONEMES,
MORA_PHONEMES_TO_MORA_KATA,
)
from style_bert_vits2.nlp.symbols import PUNCTUATIONS
def g2kata_tone(norm_text: str) -> list[tuple[str, int]]:
"""
テキストからカタカナとアクセントのペアのリストを返す。
推論時のみに使われる関数のため、常に `raise_yomi_error=False` を指定して g2p() を呼ぶ仕様になっている。
Args:
norm_text: 正規化されたテキスト。
Returns:
カタカナと音高のリスト。
"""
phones, tones, _ = g2p(norm_text, use_jp_extra=True, raise_yomi_error=False)
return phone_tone2kata_tone(list(zip(phones, tones)))
def phone_tone2kata_tone(phone_tone: list[tuple[str, int]]) -> list[tuple[str, int]]:
"""
phone_tone の phone 部分をカタカナに変換する。ただし最初と最後の ("_", 0) は無視する。
Args:
phone_tone: 音素と音高のリスト。
Returns:
カタカナと音高のリスト。
"""
# 子音の集合
CONSONANTS = set([
consonant
for consonant, _ in MORA_KATA_TO_MORA_PHONEMES.values()
if consonant is not None
])
phone_tone = phone_tone[1:] # 最初の("_", 0)を無視
phones = [phone for phone, _ in phone_tone]
tones = [tone for _, tone in phone_tone]
result: list[tuple[str, int]] = []
current_mora = ""
for phone, next_phone, tone, next_tone in zip(phones, phones[1:], tones, tones[1:]):
# zip の関係で最後の ("_", 0) は無視されている
if phone in PUNCTUATIONS:
result.append((phone, tone))
continue
if phone in CONSONANTS: # n以外の子音の場合
assert current_mora == "", f"Unexpected {phone} after {current_mora}"
assert tone == next_tone, f"Unexpected {phone} tone {tone} != {next_tone}"
current_mora = phone
else:
# phoneが母音もしくは「N」
current_mora += phone
result.append((MORA_PHONEMES_TO_MORA_KATA[current_mora], tone))
current_mora = ""
return result
def kata_tone2phone_tone(kata_tone: list[tuple[str, int]]) -> list[tuple[str, int]]:
"""
`phone_tone2kata_tone()` の逆の変換を行う。
Args:
kata_tone: カタカナと音高のリスト。
Returns:
音素と音高のリスト。
"""
result: list[tuple[str, int]] = [("_", 0)]
for mora, tone in kata_tone:
if mora in PUNCTUATIONS:
result.append((mora, tone))
else:
consonant, vowel = MORA_KATA_TO_MORA_PHONEMES[mora]
if consonant is None:
result.append((vowel, tone))
else:
result.append((consonant, tone))
result.append((vowel, tone))
result.append(("_", 0))
return result

236
style_bert_vits2/nlp/japanese/mora_list.py Normal file
View File

@@ -0,0 +1,236 @@
"""
以下のコードは VOICEVOX のソースコードからお借りし最低限の改造を行ったもの。
https://github.com/VOICEVOX/voicevox_engine/blob/master/voicevox_engine/tts_pipeline/mora_list.py
"""
"""
以下のモーラ対応表は OpenJTalk のソースコードから取得し、
カタカナ表記とモーラが一対一対応するように改造した。
ライセンス表記：
-----------------------------------------------------------------
The Japanese TTS System "Open JTalk"
developed by HTS Working Group
http://open-jtalk.sourceforge.net/
-----------------------------------------------------------------
Copyright (c) 2008-2014 Nagoya Institute of Technology
Department of Computer Science
All rights reserved.
Redistribution and use in source and binary forms, with or
without modification, are permitted provided that the following
conditions are met:
- Redistributions of source code must retain the above copyright
notice, this list of conditions and the following disclaimer.
- Redistributions in binary form must reproduce the above
copyright notice, this list of conditions and the following
disclaimer in the documentation and/or other materials provided
with the distribution.
- Neither the name of the HTS working group nor the names of its
contributors may be used to endorse or promote products derived
from this software without specific prior written permission.
THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND
CONTRIBUTORS "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES,
INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF
MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS
BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL,
EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED
TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON
ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY
OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
POSSIBILITY OF SUCH DAMAGE.
"""
from typing import Optional
# (カタカナ, 子音, 母音)の順。子音がない場合は None を入れる。
# 但し「ン」と「ッ」は母音のみという扱いで、「ン」は「N」、「ッ」は「q」とする。
# （元々「ッ」は「cl」）
# また「デェ = dy e」は pyopenjtalk の出力（de e）と合わないため削除
__MORA_LIST_MINIMUM: list[tuple[str, Optional[str], str]] = [
("ヴォ", "v", "o"),
("ヴェ", "v", "e"),
("ヴィ", "v", "i"),
("ヴァ", "v", "a"),
("ヴ", "v", "u"),
("ン", None, "N"),
("ワ", "w", "a"),
("ロ", "r", "o"),
("レ", "r", "e"),
("ル", "r", "u"),
("リョ", "ry", "o"),
("リュ", "ry", "u"),
("リャ", "ry", "a"),
("リェ", "ry", "e"),
("リ", "r", "i"),
("ラ", "r", "a"),
("ヨ", "y", "o"),
("ユ", "y", "u"),
("ヤ", "y", "a"),
("モ", "m", "o"),
("メ", "m", "e"),
("ム", "m", "u"),
("ミョ", "my", "o"),
("ミュ", "my", "u"),
("ミャ", "my", "a"),
("ミェ", "my", "e"),
("ミ", "m", "i"),
("マ", "m", "a"),
("ポ", "p", "o"),
("ボ", "b", "o"),
("ホ", "h", "o"),
("ペ", "p", "e"),
("ベ", "b", "e"),
("ヘ", "h", "e"),
("プ", "p", "u"),
("ブ", "b", "u"),
("フォ", "f", "o"),
("フェ", "f", "e"),
("フィ", "f", "i"),
("ファ", "f", "a"),
("フ", "f", "u"),
("ピョ", "py", "o"),
("ピュ", "py", "u"),
("ピャ", "py", "a"),
("ピェ", "py", "e"),
("ピ", "p", "i"),
("ビョ", "by", "o"),
("ビュ", "by", "u"),
("ビャ", "by", "a"),
("ビェ", "by", "e"),
("ビ", "b", "i"),
("ヒョ", "hy", "o"),
("ヒュ", "hy", "u"),
("ヒャ", "hy", "a"),
("ヒェ", "hy", "e"),
("ヒ", "h", "i"),
("パ", "p", "a"),
("バ", "b", "a"),
("ハ", "h", "a"),
("ノ", "n", "o"),
("ネ", "n", "e"),
("ヌ", "n", "u"),
("ニョ", "ny", "o"),
("ニュ", "ny", "u"),
("ニャ", "ny", "a"),
("ニェ", "ny", "e"),
("ニ", "n", "i"),
("ナ", "n", "a"),
("ドゥ", "d", "u"),
("ド", "d", "o"),
("トゥ", "t", "u"),
("ト", "t", "o"),
("デョ", "dy", "o"),
("デュ", "dy", "u"),
("デャ", "dy", "a"),
# ("デェ", "dy", "e"),
("ディ", "d", "i"),
("デ", "d", "e"),
("テョ", "ty", "o"),
("テュ", "ty", "u"),
("テャ", "ty", "a"),
("ティ", "t", "i"),
("テ", "t", "e"),
("ツォ", "ts", "o"),
("ツェ", "ts", "e"),
("ツィ", "ts", "i"),
("ツァ", "ts", "a"),
("ツ", "ts", "u"),
("ッ", None, "q"), # 「cl」から「q」に変更
("チョ", "ch", "o"),
("チュ", "ch", "u"),
("チャ", "ch", "a"),
("チェ", "ch", "e"),
("チ", "ch", "i"),
("ダ", "d", "a"),
("タ", "t", "a"),
("ゾ", "z", "o"),
("ソ", "s", "o"),
("ゼ", "z", "e"),
("セ", "s", "e"),
("ズィ", "z", "i"),
("ズ", "z", "u"),
("スィ", "s", "i"),
("ス", "s", "u"),
("ジョ", "j", "o"),
("ジュ", "j", "u"),
("ジャ", "j", "a"),
("ジェ", "j", "e"),
("ジ", "j", "i"),
("ショ", "sh", "o"),
("シュ", "sh", "u"),
("シャ", "sh", "a"),
("シェ", "sh", "e"),
("シ", "sh", "i"),
("ザ", "z", "a"),
("サ", "s", "a"),
("ゴ", "g", "o"),
("コ", "k", "o"),
("ゲ", "g", "e"),
("ケ", "k", "e"),
("グヮ", "gw", "a"),
("グ", "g", "u"),
("クヮ", "kw", "a"),
("ク", "k", "u"),
("ギョ", "gy", "o"),
("ギュ", "gy", "u"),
("ギャ", "gy", "a"),
("ギェ", "gy", "e"),
("ギ", "g", "i"),
("キョ", "ky", "o"),
("キュ", "ky", "u"),
("キャ", "ky", "a"),
("キェ", "ky", "e"),
("キ", "k", "i"),
("ガ", "g", "a"),
("カ", "k", "a"),
("オ", None, "o"),
("エ", None, "e"),
("ウォ", "w", "o"),
("ウェ", "w", "e"),
("ウィ", "w", "i"),
("ウ", None, "u"),
("イェ", "y", "e"),
("イ", None, "i"),
("ア", None, "a"),
]
__MORA_LIST_ADDITIONAL: list[tuple[str, Optional[str], str]] = [
("ヴョ", "by", "o"),
("ヴュ", "by", "u"),
("ヴャ", "by", "a"),
("ヲ", None, "o"),
("ヱ", None, "e"),
("ヰ", None, "i"),
("ヮ", "w", "a"),
("ョ", "y", "o"),
("ュ", "y", "u"),
("ヅ", "z", "u"),
("ヂ", "j", "i"),
("ヶ", "k", "e"),
("ャ", "y", "a"),
("ォ", None, "o"),
("ェ", None, "e"),
("ゥ", None, "u"),
("ィ", None, "i"),
("ァ", None, "a"),
]
# モーラの音素表記とカタカナの対応表
# 例: "vo" -> "ヴォ", "a" -> "ア"
MORA_PHONEMES_TO_MORA_KATA: dict[str, str] = {
(consonant or "") + vowel: kana for [kana, consonant, vowel] in __MORA_LIST_MINIMUM
}
# モーラのカタカナ表記と音素の対応表
# 例: "ヴォ" -> ("v", "o"), "ア" -> (None, "a")
MORA_KATA_TO_MORA_PHONEMES: dict[str, tuple[Optional[str], str]] = {
kana: (consonant, vowel)
for [kana, consonant, vowel] in __MORA_LIST_MINIMUM + __MORA_LIST_ADDITIONAL
}

161
style_bert_vits2/nlp/japanese/normalizer.py Normal file
View File

@@ -0,0 +1,161 @@
import re
import unicodedata
from num2words import num2words
from style_bert_vits2.nlp.symbols import PUNCTUATIONS
def normalize_text(text: str) -> str:
"""
日本語のテキストを正規化する。
結果は、ちょうど次の文字のみからなる：
- ひらがな
- カタカナ（全角長音記号「ー」が入る！）
- 漢字
- 半角アルファベット（大文字と小文字）
- ギリシャ文字
- `.` （句点`。`や`…`の一部や改行等）
- `,` （読点`、`や`:`等）
- `?` （疑問符`？`）
- `!` （感嘆符`！`）
- `'` （`「`や`」`等）
- `-` （`―`（ダッシュ、長音記号ではない）や`-`等）
注意点:
- 三点リーダー`…`は`...`に変換される（`なるほど…。` → `なるほど....`）
- 数字は漢字に変換される（`1,100円` → `千百円`、`52.34` → `五十二点三四`）
- 読点や疑問符等の位置・個数等は保持される（`??あ、、！！！` → `??あ,,!!!`）
Args:
text (str): 正規化するテキスト
Returns:
str: 正規化されたテキスト
"""
res = unicodedata.normalize("NFKC", text) # ここでアルファベットは半角になる
res = __convert_numbers_to_words(res) # 「100円」→「百円」等
# 「～」と「~」も長音記号として扱う
res = res.replace("~", "ー")
res = res.replace("～", "ー")
res = replace_punctuation(res) # 句読点等正規化、読めない文字を削除
# 結合文字の濁点・半濁点を削除
# 通常の「ば」等はそのままのこされる、「あ゛」は上で「あ゙」になりここで「あ」になる
res = res.replace("\u3099", "") # 結合文字の濁点を削除、る゙ → る
res = res.replace("\u309A", "") # 結合文字の半濁点を削除、な゚ → な
return res
def replace_punctuation(text: str) -> str:
"""
句読点等を「.」「,」「!」「?」「'」「-」に正規化し、OpenJTalk で読みが取得できるもののみ残す：
漢字・平仮名・カタカナ、アルファベット、ギリシャ文字
Args:
text (str): 正規化するテキスト
Returns:
str: 正規化されたテキスト
"""
# 記号類の正規化変換マップ
REPLACE_MAP = {
"：": ",",
"；": ",",
"，": ",",
"。": ".",
"！": "!",
"？": "?",
"\n": ".",
"．": ".",
"…": "...",
"···": "...",
"・・・": "...",
"·": ",",
"・": ",",
"、": ",",
"$": ".",
"“": "'",
"”": "'",
'"': "'",
"‘": "'",
"’": "'",
"（": "'",
"）": "'",
"(": "'",
")": "'",
"《": "'",
"》": "'",
"【": "'",
"】": "'",
"[": "'",
"]": "'",
# NFKC 正規化後のハイフン・ダッシュの変種を全て通常半角ハイフン - \u002d に変換
"\u02d7": "\u002d", # ˗, Modifier Letter Minus Sign
"\u2010": "\u002d", # ‐, Hyphen,
# "\u2011": "\u002d", # ‑, Non-Breaking Hyphen, NFKC により \u2010 に変換される
"\u2012": "\u002d", # ‒, Figure Dash
"\u2013": "\u002d", # –, En Dash
"\u2014": "\u002d", # —, Em Dash
"\u2015": "\u002d", # ―, Horizontal Bar
"\u2043": "\u002d", # ⁃, Hyphen Bullet
"\u2212": "\u002d", # −, Minus Sign
"\u23af": "\u002d", # ⎯, Horizontal Line Extension
"\u23e4": "\u002d", # ⏤, Straightness
"\u2500": "\u002d", # ─, Box Drawings Light Horizontal
"\u2501": "\u002d", # ━, Box Drawings Heavy Horizontal
"\u2e3a": "\u002d", # ⸺, Two-Em Dash
"\u2e3b": "\u002d", # ⸻, Three-Em Dash
# "～": "-", # これは長音記号「ー」として扱うよう変更
# "~": "-", # これも長音記号「ー」として扱うよう変更
"「": "'",
"」": "'",
}
pattern = re.compile("|".join(re.escape(p) for p in REPLACE_MAP.keys()))
# 句読点を辞書で置換
replaced_text = pattern.sub(lambda x: REPLACE_MAP[x.group()], text)
replaced_text = re.sub(
# ↓ ひらがな、カタカナ、漢字
r"[^\u3040-\u309F\u30A0-\u30FF\u4E00-\u9FFF\u3400-\u4DBF\u3005"
# ↓ 半角アルファベット（大文字と小文字）
+ r"\u0041-\u005A\u0061-\u007A"
# ↓ 全角アルファベット（大文字と小文字）
+ r"\uFF21-\uFF3A\uFF41-\uFF5A"
# ↓ ギリシャ文字
+ r"\u0370-\u03FF\u1F00-\u1FFF"
# ↓ "!", "?", "…", ",", ".", "'", "-", 但し`…`はすでに`...`に変換されている
+ "".join(PUNCTUATIONS) + r"]+",
# 上述以外の文字を削除
"",
replaced_text,
)
return replaced_text
def __convert_numbers_to_words(text: str) -> str:
"""
記号や数字を日本語の文字表現に変換する。
Args:
text (str): 変換するテキスト
Returns:
str: 変換されたテキスト
"""
NUMBER_WITH_SEPARATOR_PATTERN = re.compile("[0-9]{1,3}(,[0-9]{3})+")
CURRENCY_MAP = {"$": "ドル", "¥": "円", "£": "ポンド", "€": "ユーロ"}
CURRENCY_PATTERN = re.compile(r"([$¥£€])([0-9.]*[0-9])")
NUMBER_PATTERN = re.compile(r"[0-9]+(\.[0-9]+)?")
res = NUMBER_WITH_SEPARATOR_PATTERN.sub(lambda m: m[0].replace(",", ""), text)
res = CURRENCY_PATTERN.sub(lambda m: m[2] + CURRENCY_MAP.get(m[1], m[1]), res)
res = NUMBER_PATTERN.sub(lambda m: num2words(m[0], lang="ja"), res)
return res

126
style_bert_vits2/nlp/japanese/pyopenjtalk_worker/__init__.py Normal file
View File

@@ -0,0 +1,126 @@
"""
Run the pyopenjtalk worker in a separate process
to avoid user dictionary access error
"""
from typing import Any, Optional
from style_bert_vits2.logging import logger
from style_bert_vits2.nlp.japanese.pyopenjtalk_worker.worker_client import WorkerClient
from style_bert_vits2.nlp.japanese.pyopenjtalk_worker.worker_common import WORKER_PORT
WORKER_CLIENT: Optional[WorkerClient] = None
# pyopenjtalk interface
# g2p(): not used
def run_frontend(text: str) -> list[dict[str, Any]]:
assert WORKER_CLIENT
ret = WORKER_CLIENT.dispatch_pyopenjtalk("run_frontend", text)
assert isinstance(ret, list)
return ret
def make_label(njd_features: Any) -> list[str]:
assert WORKER_CLIENT
ret = WORKER_CLIENT.dispatch_pyopenjtalk("make_label", njd_features)
assert isinstance(ret, list)
return ret
def mecab_dict_index(path: str, out_path: str, dn_mecab: Optional[str] = None):
assert WORKER_CLIENT
WORKER_CLIENT.dispatch_pyopenjtalk("mecab_dict_index", path, out_path, dn_mecab)
def update_global_jtalk_with_user_dict(path: str):
assert WORKER_CLIENT
WORKER_CLIENT.dispatch_pyopenjtalk("update_global_jtalk_with_user_dict", path)
def unset_user_dict():
assert WORKER_CLIENT
WORKER_CLIENT.dispatch_pyopenjtalk("unset_user_dict")
# initialize module when imported
def initialize(port: int = WORKER_PORT) -> None:
import time
import socket
import sys
import atexit
import signal
logger.debug("initialize")
global WORKER_CLIENT
if WORKER_CLIENT:
return
client = None
try:
client = WorkerClient(port)
except (socket.timeout, socket.error):
logger.debug("try starting pyopenjtalk worker server")
import os
import subprocess
worker_pkg_path = os.path.relpath(
os.path.dirname(__file__), os.getcwd()
).replace(os.sep, ".")
args = [sys.executable, "-m", worker_pkg_path, "--port", str(port)]
# new session, new process group
if sys.platform.startswith("win"):
cf = subprocess.CREATE_NEW_CONSOLE | subprocess.CREATE_NEW_PROCESS_GROUP # type: ignore
si = subprocess.STARTUPINFO() # type: ignore
si.dwFlags |= subprocess.STARTF_USESHOWWINDOW # type: ignore
si.wShowWindow = subprocess.SW_HIDE # type: ignore
subprocess.Popen(args, creationflags=cf, startupinfo=si)
else:
# align with Windows behavior
# start_new_session is same as specifying setsid in preexec_fn
subprocess.Popen(args, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, start_new_session=True) # type: ignore
# wait until server listening
count = 0
while True:
try:
client = WorkerClient(port)
break
except socket.error:
time.sleep(1)
count += 1
# 10: max number of retries
if count == 10:
raise TimeoutError("サーバーに接続できませんでした")
WORKER_CLIENT = client
atexit.register(terminate)
# when the process is killed
def signal_handler(signum: int, frame: Any):
terminate()
signal.signal(signal.SIGTERM, signal_handler)
# top-level declaration
def terminate() -> None:
logger.debug("terminate")
global WORKER_CLIENT
if not WORKER_CLIENT:
return
# repare for unexpected errors
try:
if WORKER_CLIENT.status() == 1:
WORKER_CLIENT.quit_server()
except Exception as e:
logger.error(e)
WORKER_CLIENT.close()
WORKER_CLIENT = None

16
style_bert_vits2/nlp/japanese/pyopenjtalk_worker/__main__.py Normal file
View File

@@ -0,0 +1,16 @@
import argparse
from style_bert_vits2.nlp.japanese.pyopenjtalk_worker.worker_common import WORKER_PORT
from style_bert_vits2.nlp.japanese.pyopenjtalk_worker.worker_server import WorkerServer
def main() -> None:
parser = argparse.ArgumentParser()
parser.add_argument("--port", type=int, default=WORKER_PORT)
args = parser.parse_args()
server = WorkerServer()
server.start_server(port=args.port)
if __name__ == "__main__":
main()

63
style_bert_vits2/nlp/japanese/pyopenjtalk_worker/worker_client.py Normal file
View File

@@ -0,0 +1,63 @@
import socket
from typing import Any, cast
from style_bert_vits2.logging import logger
from style_bert_vits2.nlp.japanese.pyopenjtalk_worker.worker_common import RequestType, receive_data, send_data
class WorkerClient:
""" pyopenjtalk worker client """
def __init__(self, port: int) -> None:
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
# 5: timeout
sock.settimeout(5)
sock.connect((socket.gethostname(), port))
self.sock = sock
def __enter__(self) -> "WorkerClient":
return self
def __exit__(self, exc_type: Any, exc_value: Any, traceback: Any) -> None:
self.close()
def close(self) -> None:
self.sock.close()
def dispatch_pyopenjtalk(self, func: str, *args: Any, **kwargs: Any) -> Any:
data = {
"request-type": RequestType.PYOPENJTALK,
"func": func,
"args": args,
"kwargs": kwargs,
}
logger.trace(f"client sends request: {data}")
send_data(self.sock, data)
logger.trace("client sent request successfully")
response = receive_data(self.sock)
logger.trace(f"client received response: {response}")
return response.get("return")
def status(self) -> int:
data = {"request-type": RequestType.STATUS}
logger.trace(f"client sends request: {data}")
send_data(self.sock, data)
logger.trace("client sent request successfully")
response = receive_data(self.sock)
logger.trace(f"client received response: {response}")
return cast(int, response.get("client-count"))
def quit_server(self) -> None:
data = {"request-type": RequestType.QUIT_SERVER}
logger.trace(f"client sends request: {data}")
send_data(self.sock, data)
logger.trace("client sent request successfully")
response = receive_data(self.sock)
logger.trace(f"client received response: {response}")

45
style_bert_vits2/nlp/japanese/pyopenjtalk_worker/worker_common.py Normal file
View File

@@ -0,0 +1,45 @@
import json
import socket
from enum import IntEnum, auto
from typing import Any, Final
WORKER_PORT: Final[int] = 7861
HEADER_SIZE: Final[int] = 4
class RequestType(IntEnum):
STATUS = auto()
QUIT_SERVER = auto()
PYOPENJTALK = auto()
class ConnectionClosedException(Exception):
pass
# socket communication
def send_data(sock: socket.socket, data: dict[str, Any]):
json_data = json.dumps(data).encode()
header = len(json_data).to_bytes(HEADER_SIZE, byteorder="big")
sock.sendall(header + json_data)
def __receive_until(sock: socket.socket, size: int):
data = b""
while len(data) < size:
part = sock.recv(size - len(data))
if part == b"":
raise ConnectionClosedException("接続が閉じられました")
data += part
return data
def receive_data(sock: socket.socket) -> dict[str, Any]:
header = __receive_until(sock, HEADER_SIZE)
data_length = int.from_bytes(header, byteorder="big")
body = __receive_until(sock, data_length)
return json.loads(body.decode())

126
style_bert_vits2/nlp/japanese/pyopenjtalk_worker/worker_server.py Normal file
View File

@@ -0,0 +1,126 @@
import select
import socket
import time
from typing import Any, cast
import pyopenjtalk
from style_bert_vits2.logging import logger
from style_bert_vits2.nlp.japanese.pyopenjtalk_worker.worker_common import (
ConnectionClosedException,
RequestType,
receive_data,
send_data,
)
# To make it as fast as possible
# Probably faster than calling getattr every time
__PYOPENJTALK_FUNC_DICT = {
"run_frontend": pyopenjtalk.run_frontend,
"make_label": pyopenjtalk.make_label,
"mecab_dict_index": pyopenjtalk.mecab_dict_index,
"update_global_jtalk_with_user_dict": pyopenjtalk.update_global_jtalk_with_user_dict,
"unset_user_dict": pyopenjtalk.unset_user_dict,
}
class WorkerServer:
""" pyopenjtalk worker server """
def __init__(self) -> None:
self.client_count: int = 0
self.quit: bool = False
def handle_request(self, request: dict[str, Any]) -> dict[str, Any]:
request_type = None
try:
request_type = RequestType(cast(int, request.get("request-type")))
except Exception:
return {
"success": False,
"reason": "request-type is invalid",
}
response: dict[str, Any] = {}
if request_type:
if request_type == RequestType.STATUS:
response = {
"success": True,
"client-count": self.client_count,
}
elif request_type == RequestType.QUIT_SERVER:
self.quit = True
response = {"success": True}
elif request_type == RequestType.PYOPENJTALK:
func_name = request.get("func")
assert isinstance(func_name, str)
func = __PYOPENJTALK_FUNC_DICT[func_name]
args = request.get("args")
kwargs = request.get("kwargs")
assert isinstance(args, list)
assert isinstance(kwargs, dict)
ret = func(*args, **kwargs)
response = {"success": True, "return": ret}
else:
# NOT REACHED
response = request
return response
def start_server(self, port: int, no_client_timeout: int = 30) -> None:
logger.info("start pyopenjtalk worker server")
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as server_socket:
server_socket.bind((socket.gethostname(), port))
server_socket.listen()
sockets = [server_socket]
no_client_since = time.time()
while True:
if self.client_count == 0:
if no_client_since is None:
no_client_since = time.time()
elif (time.time() - no_client_since) > no_client_timeout:
logger.info("quit because there is no client")
return
else:
no_client_since = None
ready_sockets, _, _ = select.select(sockets, [], [], 0.1)
for sock in ready_sockets:
if sock is server_socket:
logger.info("new client connected")
client_socket, _ = server_socket.accept()
sockets.append(client_socket)
self.client_count += 1
else:
# client
try:
request = receive_data(sock)
except Exception as e:
sock.close()
sockets.remove(sock)
self.client_count -= 1
# unexpected disconnections
if not isinstance(e, ConnectionClosedException):
logger.error(e)
logger.info("close connection")
continue
logger.trace(f"server received request: {request}")
response = self.handle_request(request)
logger.trace(f"server sends response: {response}")
try:
send_data(sock, response)
logger.trace("server sent response successfully")
except Exception:
logger.warning(
"an exception occurred during sending responce"
)
if self.quit:
logger.info("quit pyopenjtalk worker server")
return

27
style_bert_vits2/nlp/japanese/user_dict/README.md Normal file
View File

@@ -0,0 +1,27 @@
## ユーザー辞書関連のコードについて
このフォルダに含まれるユーザー辞書関連のコードは、[VOICEVOX ENGINE](https://github.com/VOICEVOX/voicevox_engine) プロジェクトのコードを改変したものを使用しています。
VOICEVOX プロジェクトのチームに深く感謝し、その貢献を尊重します。
### 元のコード
- [voicevox_engine/user_dict/](https://github.com/VOICEVOX/voicevox_engine/tree/f181411ec69812296989d9cc583826c22eec87ae/voicevox_engine/user_dict)
- [voicevox_engine/model.py](https://github.com/VOICEVOX/voicevox_engine/blob/f181411ec69812296989d9cc583826c22eec87ae/voicevox_engine/model.py#L207)
### 改変の詳細
- ファイル名の書き換えおよびそれに伴う import 文の書き換え。
- VOICEVOX 固有の部分をコメントアウト。
- mutex を使用している部分をコメントアウト。
- 参照している pyopenjtalk の違いによるメソッド名の書き換え。
- UserDictWord の mora_count のデフォルト値を None に指定。
- `model.py` のうち、必要な Pydantic モデルのみを抽出。
### ライセンス
元の VOICEVOX ENGINE のリポジトリのコードは、LGPL v3 と、ソースコードの公開が不要な別ライセンスのデュアルライセンスの下で使用されています。
当プロジェクトにおけるこのモジュールも LGPL ライセンスの下にあります。
詳細については、プロジェクトのルートディレクトリにある [LGPL_LICENSE](/LGPL_LICENSE) ファイルをご参照ください。
また、元の VOICEVOX ENGINE プロジェクトのライセンスについては、[こちら](https://github.com/VOICEVOX/voicevox_engine/blob/master/LICENSE) をご覧ください。

464
style_bert_vits2/nlp/japanese/user_dict/__init__.py Normal file
View File

@@ -0,0 +1,464 @@
"""
このファイルは、VOICEVOX プロジェクトの VOICEVOX ENGINE からお借りしています。
引用元: https://github.com/VOICEVOX/voicevox_engine/blob/f181411ec69812296989d9cc583826c22eec87ae/voicevox_engine/user_dict/user_dict.py
ライセンス: LGPL-3.0
詳しくは、このファイルと同じフォルダにある README.md を参照してください。
"""
import json
import sys
import traceback
from pathlib import Path
from typing import Dict, List, Optional
from uuid import UUID, uuid4
import numpy as np
from fastapi import HTTPException
from style_bert_vits2.constants import DEFAULT_USER_DICT_DIR
from style_bert_vits2.nlp.japanese import pyopenjtalk_worker as pyopenjtalk
from style_bert_vits2.nlp.japanese.user_dict.word_model import UserDictWord, WordTypes
from style_bert_vits2.nlp.japanese.user_dict.part_of_speech_data import MAX_PRIORITY, MIN_PRIORITY, part_of_speech_data
pyopenjtalk.initialize()
# root_dir = engine_root()
# save_dir = get_save_dir()
# if not save_dir.is_dir():
# save_dir.mkdir(parents=True)
default_dict_path = DEFAULT_USER_DICT_DIR / "default.csv" # VOICEVOXデフォルト辞書ファイルのパス
user_dict_path = DEFAULT_USER_DICT_DIR / "user_dict.json" # ユーザー辞書ファイルのパス
compiled_dict_path = DEFAULT_USER_DICT_DIR / "user.dic" # コンパイル済み辞書ファイルのパス
# # 同時書き込みの制御
# mutex_user_dict = threading.Lock()
# mutex_openjtalk_dict = threading.Lock()
# @mutex_wrapper(mutex_user_dict)
def _write_to_json(user_dict: Dict[str, UserDictWord], user_dict_path: Path) -> None:
"""
ユーザー辞書ファイルへのユーザー辞書データ書き込み
Parameters
----------
user_dict : Dict[str, UserDictWord]
ユーザー辞書データ
user_dict_path : Path
ユーザー辞書ファイルのパス
"""
converted_user_dict = {}
for word_uuid, word in user_dict.items():
word_dict = word.model_dump()
word_dict["cost"] = _priority2cost(
word_dict["context_id"], word_dict["priority"]
)
del word_dict["priority"]
converted_user_dict[word_uuid] = word_dict
# 予めjsonに変換できることを確かめる
user_dict_json = json.dumps(converted_user_dict, ensure_ascii=False)
# ユーザー辞書ファイルへの書き込み
user_dict_path.write_text(user_dict_json, encoding="utf-8")
# @mutex_wrapper(mutex_openjtalk_dict)
def update_dict(
default_dict_path: Path = default_dict_path,
user_dict_path: Path = user_dict_path,
compiled_dict_path: Path = compiled_dict_path,
) -> None:
"""
辞書の更新
Parameters
----------
default_dict_path : Path
デフォルト辞書ファイルのパス
user_dict_path : Path
ユーザー辞書ファイルのパス
compiled_dict_path : Path
コンパイル済み辞書ファイルのパス
"""
random_string = uuid4()
tmp_csv_path = compiled_dict_path.with_suffix(
f".dict_csv-{random_string}.tmp"
) # csv形式辞書データの一時保存ファイル
tmp_compiled_path = compiled_dict_path.with_suffix(
f".dict_compiled-{random_string}.tmp"
) # コンパイル済み辞書データの一時保存ファイル
try:
# 辞書.csvを作成
csv_text = ""
# デフォルト辞書データの追加
if not default_dict_path.is_file():
print("Warning: Cannot find default dictionary.", file=sys.stderr)
return
default_dict = default_dict_path.read_text(encoding="utf-8")
if default_dict == default_dict.rstrip():
default_dict += "\n"
csv_text += default_dict
# ユーザー辞書データの追加
user_dict = read_dict(user_dict_path=user_dict_path)
for word_uuid in user_dict:
word = user_dict[word_uuid]
csv_text += (
"{surface},{context_id},{context_id},{cost},{part_of_speech},"
+ "{part_of_speech_detail_1},{part_of_speech_detail_2},"
+ "{part_of_speech_detail_3},{inflectional_type},"
+ "{inflectional_form},{stem},{yomi},{pronunciation},"
+ "{accent_type}/{mora_count},{accent_associative_rule}\n"
).format(
surface=word.surface,
context_id=word.context_id,
cost=_priority2cost(word.context_id, word.priority),
part_of_speech=word.part_of_speech,
part_of_speech_detail_1=word.part_of_speech_detail_1,
part_of_speech_detail_2=word.part_of_speech_detail_2,
part_of_speech_detail_3=word.part_of_speech_detail_3,
inflectional_type=word.inflectional_type,
inflectional_form=word.inflectional_form,
stem=word.stem,
yomi=word.yomi,
pronunciation=word.pronunciation,
accent_type=word.accent_type,
mora_count=word.mora_count,
accent_associative_rule=word.accent_associative_rule,
)
# 辞書データを辞書.csv へ一時保存
tmp_csv_path.write_text(csv_text, encoding="utf-8")
# 辞書.csvをOpenJTalk用にコンパイル
# pyopenjtalk.create_user_dict(str(tmp_csv_path), str(tmp_compiled_path))
pyopenjtalk.mecab_dict_index(str(tmp_csv_path), str(tmp_compiled_path))
if not tmp_compiled_path.is_file():
raise RuntimeError("辞書のコンパイル時にエラーが発生しました。")
# コンパイル済み辞書の置き換え・読み込み
pyopenjtalk.unset_user_dict()
tmp_compiled_path.replace(compiled_dict_path)
if compiled_dict_path.is_file():
# pyopenjtalk.set_user_dict(str(compiled_dict_path.resolve(strict=True)))
pyopenjtalk.update_global_jtalk_with_user_dict(str(compiled_dict_path))
except Exception as e:
print("Error: Failed to update dictionary.", file=sys.stderr)
traceback.print_exc(file=sys.stderr)
raise e
finally:
# 後処理
if tmp_csv_path.exists():
tmp_csv_path.unlink()
if tmp_compiled_path.exists():
tmp_compiled_path.unlink()
# @mutex_wrapper(mutex_user_dict)
def read_dict(user_dict_path: Path = user_dict_path) -> Dict[str, UserDictWord]:
"""
ユーザー辞書の読み出し
Parameters
----------
user_dict_path : Path
ユーザー辞書ファイルのパス
Returns
-------
result : Dict[str, UserDictWord]
ユーザー辞書
"""
# 指定ユーザー辞書が存在しない場合、空辞書を返す
if not user_dict_path.is_file():
return {}
with user_dict_path.open(encoding="utf-8") as f:
result: Dict[str, UserDictWord] = {}
for word_uuid, word in json.load(f).items():
# cost2priorityで変換を行う際にcontext_idが必要となるが、
# 0.12以前の辞書は、context_idがハードコーディングされていたためにユーザー辞書内に保管されていない
# ハードコーディングされていたcontext_idは固有名詞を意味するものなので、固有名詞のcontext_idを補完する
if word.get("context_id") is None:
word["context_id"] = part_of_speech_data[
WordTypes.PROPER_NOUN
].context_id
word["priority"] = _cost2priority(word["context_id"], word["cost"])
del word["cost"]
result[str(UUID(word_uuid))] = UserDictWord(**word)
return result
def _create_word(
surface: str,
pronunciation: str,
accent_type: int,
word_type: Optional[WordTypes] = None,
priority: Optional[int] = None,
) -> UserDictWord:
"""
単語オブジェクトの生成
Parameters
----------
surface : str
単語情報
pronunciation : str
単語情報
accent_type : int
単語情報
word_type : Optional[WordTypes]
品詞
priority : Optional[int]
優先度
Returns
-------
: UserDictWord
単語オブジェクト
"""
if word_type is None:
word_type = WordTypes.PROPER_NOUN
if word_type not in part_of_speech_data.keys():
raise HTTPException(status_code=422, detail="不明な品詞です")
if priority is None:
priority = 5
if not MIN_PRIORITY <= priority <= MAX_PRIORITY:
raise HTTPException(status_code=422, detail="優先度の値が無効です")
pos_detail = part_of_speech_data[word_type]
return UserDictWord(
surface=surface,
context_id=pos_detail.context_id,
priority=priority,
part_of_speech=pos_detail.part_of_speech,
part_of_speech_detail_1=pos_detail.part_of_speech_detail_1,
part_of_speech_detail_2=pos_detail.part_of_speech_detail_2,
part_of_speech_detail_3=pos_detail.part_of_speech_detail_3,
inflectional_type="*",
inflectional_form="*",
stem="*",
yomi=pronunciation,
pronunciation=pronunciation,
accent_type=accent_type,
accent_associative_rule="*",
)
def apply_word(
surface: str,
pronunciation: str,
accent_type: int,
word_type: Optional[WordTypes] = None,
priority: Optional[int] = None,
user_dict_path: Path = user_dict_path,
compiled_dict_path: Path = compiled_dict_path,
) -> str:
"""
新規単語の追加
Parameters
----------
surface : str
単語情報
pronunciation : str
単語情報
accent_type : int
単語情報
word_type : Optional[WordTypes]
品詞
priority : Optional[int]
優先度
user_dict_path : Path
ユーザー辞書ファイルのパス
compiled_dict_path : Path
コンパイル済み辞書ファイルのパス
Returns
-------
word_uuid : UserDictWord
追加された単語に発行されたUUID
"""
# 新規単語の追加による辞書データの更新
word = _create_word(
surface=surface,
pronunciation=pronunciation,
accent_type=accent_type,
word_type=word_type,
priority=priority,
)
user_dict = read_dict(user_dict_path=user_dict_path)
word_uuid = str(uuid4())
user_dict[word_uuid] = word
# 更新された辞書データの保存と適用
_write_to_json(user_dict, user_dict_path)
update_dict(user_dict_path=user_dict_path, compiled_dict_path=compiled_dict_path)
return word_uuid
def rewrite_word(
word_uuid: str,
surface: str,
pronunciation: str,
accent_type: int,
word_type: Optional[WordTypes] = None,
priority: Optional[int] = None,
user_dict_path: Path = user_dict_path,
compiled_dict_path: Path = compiled_dict_path,
) -> None:
"""
既存単語の上書き更新
Parameters
----------
word_uuid : str
単語UUID
surface : str
単語情報
pronunciation : str
単語情報
accent_type : int
単語情報
word_type : Optional[WordTypes]
品詞
priority : Optional[int]
優先度
user_dict_path : Path
ユーザー辞書ファイルのパス
compiled_dict_path : Path
コンパイル済み辞書ファイルのパス
"""
word = _create_word(
surface=surface,
pronunciation=pronunciation,
accent_type=accent_type,
word_type=word_type,
priority=priority,
)
# 既存単語の上書きによる辞書データの更新
user_dict = read_dict(user_dict_path=user_dict_path)
if word_uuid not in user_dict:
raise HTTPException(
status_code=422, detail="UUIDに該当するワードが見つかりませんでした"
)
user_dict[word_uuid] = word
# 更新された辞書データの保存と適用
_write_to_json(user_dict, user_dict_path)
update_dict(user_dict_path=user_dict_path, compiled_dict_path=compiled_dict_path)
def delete_word(
word_uuid: str,
user_dict_path: Path = user_dict_path,
compiled_dict_path: Path = compiled_dict_path,
) -> None:
"""
単語の削除
Parameters
----------
word_uuid : str
単語UUID
user_dict_path : Path
ユーザー辞書ファイルのパス
compiled_dict_path : Path
コンパイル済み辞書ファイルのパス
"""
# 既存単語の削除による辞書データの更新
user_dict = read_dict(user_dict_path=user_dict_path)
if word_uuid not in user_dict:
raise HTTPException(
status_code=422, detail="IDに該当するワードが見つかりませんでした"
)
del user_dict[word_uuid]
# 更新された辞書データの保存と適用
_write_to_json(user_dict, user_dict_path)
update_dict(user_dict_path=user_dict_path, compiled_dict_path=compiled_dict_path)
def import_user_dict(
dict_data: Dict[str, UserDictWord],
override: bool = False,
user_dict_path: Path = user_dict_path,
default_dict_path: Path = default_dict_path,
compiled_dict_path: Path = compiled_dict_path,
) -> None:
"""
ユーザー辞書のインポート
Parameters
----------
dict_data : Dict[str, UserDictWord]
インポートするユーザー辞書のデータ
override : bool
重複したエントリがあった場合、上書きするかどうか
user_dict_path : Path
ユーザー辞書ファイルのパス
default_dict_path : Path
デフォルト辞書ファイルのパス
compiled_dict_path : Path
コンパイル済み辞書ファイルのパス
"""
# インポートする辞書データのバリデーション
for word_uuid, word in dict_data.items():
UUID(word_uuid)
assert isinstance(word, UserDictWord)
for pos_detail in part_of_speech_data.values():
if word.context_id == pos_detail.context_id:
assert word.part_of_speech == pos_detail.part_of_speech
assert (
word.part_of_speech_detail_1 == pos_detail.part_of_speech_detail_1
)
assert (
word.part_of_speech_detail_2 == pos_detail.part_of_speech_detail_2
)
assert (
word.part_of_speech_detail_3 == pos_detail.part_of_speech_detail_3
)
assert (
word.accent_associative_rule in pos_detail.accent_associative_rules
)
break
else:
raise ValueError("対応していない品詞です")
# 既存辞書の読み出し
old_dict = read_dict(user_dict_path=user_dict_path)
# 辞書データの更新
# 重複エントリの上書き
if override:
new_dict = {**old_dict, **dict_data}
# 重複エントリの保持
else:
new_dict = {**dict_data, **old_dict}
# 更新された辞書データの保存と適用
_write_to_json(user_dict=new_dict, user_dict_path=user_dict_path)
update_dict(
default_dict_path=default_dict_path,
user_dict_path=user_dict_path,
compiled_dict_path=compiled_dict_path,
)
def _search_cost_candidates(context_id: int) -> List[int]:
for value in part_of_speech_data.values():
if value.context_id == context_id:
return value.cost_candidates
raise HTTPException(status_code=422, detail="品詞IDが不正です")
def _cost2priority(context_id: int, cost: int) -> int:
assert -32768 <= cost <= 32767
cost_candidates = _search_cost_candidates(context_id)
# cost_candidatesの中にある値で最も近い値を元にpriorityを返す
# 参考: https://qiita.com/Krypf/items/2eada91c37161d17621d
# この関数とpriority2cost関数によって、辞書ファイルのcostを操作しても最も近いpriorityのcostに上書きされる
return MAX_PRIORITY - np.argmin(np.abs(np.array(cost_candidates) - cost)).item()
def _priority2cost(context_id: int, priority: int) -> int:
assert MIN_PRIORITY <= priority <= MAX_PRIORITY
cost_candidates = _search_cost_candidates(context_id)
return cost_candidates[MAX_PRIORITY - priority]

151
style_bert_vits2/nlp/japanese/user_dict/part_of_speech_data.py Normal file
View File

@@ -0,0 +1,151 @@
"""
このファイルは、VOICEVOX プロジェクトの VOICEVOX ENGINE からお借りしています。
引用元: https://github.com/VOICEVOX/voicevox_engine/blob/f181411ec69812296989d9cc583826c22eec87ae/voicevox_engine/user_dict/part_of_speech_data.py
ライセンス: LGPL-3.0
詳しくは、このファイルと同じフォルダにある README.md を参照してください。
"""
from typing import Dict
from style_bert_vits2.nlp.japanese.user_dict.word_model import (
USER_DICT_MAX_PRIORITY,
USER_DICT_MIN_PRIORITY,
PartOfSpeechDetail,
WordTypes,
)
MIN_PRIORITY = USER_DICT_MIN_PRIORITY
MAX_PRIORITY = USER_DICT_MAX_PRIORITY
part_of_speech_data: Dict[WordTypes, PartOfSpeechDetail] = {
WordTypes.PROPER_NOUN: PartOfSpeechDetail(
part_of_speech="名詞",
part_of_speech_detail_1="固有名詞",
part_of_speech_detail_2="一般",
part_of_speech_detail_3="*",
context_id=1348,
cost_candidates=[
-988,
3488,
4768,
6048,
7328,
8609,
8734,
8859,
8984,
9110,
14176,
],
accent_associative_rules=[
"*",
"C1",
"C2",
"C3",
"C4",
"C5",
],
),
WordTypes.COMMON_NOUN: PartOfSpeechDetail(
part_of_speech="名詞",
part_of_speech_detail_1="一般",
part_of_speech_detail_2="*",
part_of_speech_detail_3="*",
context_id=1345,
cost_candidates=[
-4445,
49,
1473,
2897,
4321,
5746,
6554,
7362,
8170,
8979,
15001,
],
accent_associative_rules=[
"*",
"C1",
"C2",
"C3",
"C4",
"C5",
],
),
WordTypes.VERB: PartOfSpeechDetail(
part_of_speech="動詞",
part_of_speech_detail_1="自立",
part_of_speech_detail_2="*",
part_of_speech_detail_3="*",
context_id=642,
cost_candidates=[
3100,
6160,
6360,
6561,
6761,
6962,
7414,
7866,
8318,
8771,
13433,
],
accent_associative_rules=[
"*",
],
),
WordTypes.ADJECTIVE: PartOfSpeechDetail(
part_of_speech="形容詞",
part_of_speech_detail_1="自立",
part_of_speech_detail_2="*",
part_of_speech_detail_3="*",
context_id=20,
cost_candidates=[
1527,
3266,
3561,
3857,
4153,
4449,
5149,
5849,
6549,
7250,
10001,
],
accent_associative_rules=[
"*",
],
),
WordTypes.SUFFIX: PartOfSpeechDetail(
part_of_speech="名詞",
part_of_speech_detail_1="接尾",
part_of_speech_detail_2="一般",
part_of_speech_detail_3="*",
context_id=1358,
cost_candidates=[
4399,
5373,
6041,
6710,
7378,
8047,
9440,
10834,
12228,
13622,
15847,
],
accent_associative_rules=[
"*",
"C1",
"C2",
"C3",
"C4",
"C5",
],
),
}

131
style_bert_vits2/nlp/japanese/user_dict/word_model.py Normal file
View File

@@ -0,0 +1,131 @@
"""
このファイルは、VOICEVOX プロジェクトの VOICEVOX ENGINE からお借りしています。
引用元: https://github.com/VOICEVOX/voicevox_engine/blob/f181411ec69812296989d9cc583826c22eec87ae/voicevox_engine/model.py#L207
ライセンス: LGPL-3.0
詳しくは、このファイルと同じフォルダにある README.md を参照してください。
"""
from enum import Enum
from re import findall, fullmatch
from typing import List, Optional
from pydantic import BaseModel, Field, validator
USER_DICT_MIN_PRIORITY = 0
USER_DICT_MAX_PRIORITY = 10
class UserDictWord(BaseModel):
"""
辞書のコンパイルに使われる情報
"""
surface: str = Field(title="表層形")
priority: int = Field(
title="優先度", ge=USER_DICT_MIN_PRIORITY, le=USER_DICT_MAX_PRIORITY
)
context_id: int = Field(title="文脈ID", default=1348)
part_of_speech: str = Field(title="品詞")
part_of_speech_detail_1: str = Field(title="品詞細分類1")
part_of_speech_detail_2: str = Field(title="品詞細分類2")
part_of_speech_detail_3: str = Field(title="品詞細分類3")
inflectional_type: str = Field(title="活用型")
inflectional_form: str = Field(title="活用形")
stem: str = Field(title="原形")
yomi: str = Field(title="読み")
pronunciation: str = Field(title="発音")
accent_type: int = Field(title="アクセント型")
mora_count: Optional[int] = Field(title="モーラ数", default=None)
accent_associative_rule: str = Field(title="アクセント結合規則")
class Config:
validate_assignment = True
@validator("surface")
def convert_to_zenkaku(cls, surface):
return surface.translate(
str.maketrans(
"".join(chr(0x21 + i) for i in range(94)),
"".join(chr(0xFF01 + i) for i in range(94)),
)
)
@validator("pronunciation", pre=True)
def check_is_katakana(cls, pronunciation):
if not fullmatch(r"[ァ-ヴー]+", pronunciation):
raise ValueError("発音は有効なカタカナでなくてはいけません。")
sutegana = ["ァ", "ィ", "ゥ", "ェ", "ォ", "ャ", "ュ", "ョ", "ヮ", "ッ"]
for i in range(len(pronunciation)):
if pronunciation[i] in sutegana:
# 「キャット」のように、捨て仮名が連続する可能性が考えられるので、
# 「ッ」に関しては「ッ」そのものが連続している場合と、「ッ」の後にほかの捨て仮名が連続する場合のみ無効とする
if i < len(pronunciation) - 1 and (
pronunciation[i + 1] in sutegana[:-1]
or (
pronunciation[i] == sutegana[-1]
and pronunciation[i + 1] == sutegana[-1]
)
):
raise ValueError("無効な発音です。(捨て仮名の連続)")
if pronunciation[i] == "ヮ":
if i != 0 and pronunciation[i - 1] not in ["ク", "グ"]:
raise ValueError(
"無効な発音です。(「くゎ」「ぐゎ」以外の「ゎ」の使用)"
)
return pronunciation
@validator("mora_count", pre=True, always=True)
def check_mora_count_and_accent_type(cls, mora_count, values):
if "pronunciation" not in values or "accent_type" not in values:
# 適切な場所でエラーを出すようにする
return mora_count
if mora_count is None:
rule_others = (
"[イ][ェ]|[ヴ][ャュョ]|[トド][ゥ]|[テデ][ィャュョ]|[デ][ェ]|[クグ][ヮ]"
)
rule_line_i = "[キシチニヒミリギジビピ][ェャュョ]"
rule_line_u = "[ツフヴ][ァ]|[ウスツフヴズ][ィ]|[ウツフヴ][ェォ]"
rule_one_mora = "[ァ-ヴー]"
mora_count = len(
findall(
f"(?:{rule_others}|{rule_line_i}|{rule_line_u}|{rule_one_mora})",
values["pronunciation"],
)
)
if not 0 <= values["accent_type"] <= mora_count:
raise ValueError(
"誤ったアクセント型です({})。 expect: 0 <= accent_type <= {}".format(
values["accent_type"], mora_count
)
)
return mora_count
class PartOfSpeechDetail(BaseModel):
"""
品詞ごとの情報
"""
part_of_speech: str = Field(title="品詞")
part_of_speech_detail_1: str = Field(title="品詞細分類1")
part_of_speech_detail_2: str = Field(title="品詞細分類2")
part_of_speech_detail_3: str = Field(title="品詞細分類3")
# context_idは辞書の左・右文脈IDのこと
# https://github.com/VOICEVOX/open_jtalk/blob/427cfd761b78efb6094bea3c5bb8c968f0d711ab/src/mecab-naist-jdic/_left-id.def # noqa
context_id: int = Field(title="文脈ID")
cost_candidates: List[int] = Field(title="コストのパーセンタイル")
accent_associative_rules: List[str] = Field(title="アクセント結合規則の一覧")
class WordTypes(str, Enum):
"""
fastapiでword_type引数を検証する時に使用するクラス
"""
PROPER_NOUN = "PROPER_NOUN"
COMMON_NOUN = "COMMON_NOUN"
VERB = "VERB"
ADJECTIVE = "ADJECTIVE"
SUFFIX = "SUFFIX"