Feat: add editor server backend
This commit is contained in:
4
.gitignore
vendored
4
.gitignore
vendored
@@ -28,3 +28,7 @@ venv/
|
||||
|
||||
safetensors.ipynb
|
||||
*.wav
|
||||
/Web/
|
||||
|
||||
# pyopenjtalk's dictionary
|
||||
*.dic
|
||||
|
||||
3
app.py
3
app.py
@@ -3,6 +3,7 @@ import datetime
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from typing import Optional
|
||||
|
||||
import gradio as gr
|
||||
@@ -271,7 +272,7 @@ if __name__ == "__main__":
|
||||
help="Do not launch app automatically",
|
||||
)
|
||||
args = parser.parse_args()
|
||||
model_dir = args.dir
|
||||
model_dir = Path(args.dir)
|
||||
|
||||
if args.cpu:
|
||||
device = "cpu"
|
||||
|
||||
@@ -5,6 +5,10 @@ import enum
|
||||
GRADIO_THEME: str = "NoCrypt/miku"
|
||||
|
||||
LATEST_VERSION: str = "2.3"
|
||||
|
||||
USER_DICT_CSV_PATH: str = "dict/user_dict.csv"
|
||||
USER_DICT_PATH: str = "dict/user.dic"
|
||||
|
||||
DEFAULT_STYLE: str = "Neutral"
|
||||
DEFAULT_STYLE_WEIGHT: float = 5.0
|
||||
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
import os
|
||||
import warnings
|
||||
from typing import Dict, List, Optional, Union
|
||||
from pathlib import Path
|
||||
from typing import Optional, Union
|
||||
|
||||
import gradio as gr
|
||||
import numpy as np
|
||||
@@ -29,12 +30,13 @@ from .log import logger
|
||||
|
||||
def adjust_voice(fs, wave, pitch_scale, intonation_scale):
|
||||
if pitch_scale == 1.0 and intonation_scale == 1.0:
|
||||
# 初期値の場合は、音質劣化を避けるためにそのまま返す
|
||||
return fs, wave
|
||||
|
||||
# pyworldでf0を加工して合成
|
||||
# pyworldよりもよいのがあるかもしれないが……
|
||||
|
||||
wave = wave.astype(np.double)
|
||||
logger.debug(f"wave: shape={wave.shape}, max={max(wave)}, min={min(wave)}")
|
||||
f0, t = pyworld.harvest(wave, fs)
|
||||
# 質が高そうだしとりあえずharvestにしておく
|
||||
|
||||
@@ -43,7 +45,6 @@ def adjust_voice(fs, wave, pitch_scale, intonation_scale):
|
||||
|
||||
non_zero_f0 = [f for f in f0 if f != 0]
|
||||
f0_mean = sum(non_zero_f0) / len(non_zero_f0)
|
||||
logger.debug(f"f0: shape={f0.shape}, mean={f0_mean}, max={max(f0)}")
|
||||
|
||||
for i, f in enumerate(f0):
|
||||
if f == 0:
|
||||
@@ -56,21 +57,21 @@ def adjust_voice(fs, wave, pitch_scale, intonation_scale):
|
||||
|
||||
class Model:
|
||||
def __init__(
|
||||
self, model_path: str, config_path: str, style_vec_path: str, device: str
|
||||
self, model_path: Path, config_path: Path, style_vec_path: Path, device: str
|
||||
):
|
||||
self.model_path: str = model_path
|
||||
self.config_path: str = config_path
|
||||
self.model_path: Path = model_path
|
||||
self.config_path: Path = config_path
|
||||
self.style_vec_path: Path = style_vec_path
|
||||
self.device: str = device
|
||||
self.style_vec_path: str = style_vec_path
|
||||
self.hps: utils.HParams = utils.get_hparams_from_file(self.config_path)
|
||||
self.spk2id: Dict[str, int] = self.hps.data.spk2id
|
||||
self.id2spk: Dict[int, str] = {v: k for k, v in self.spk2id.items()}
|
||||
self.spk2id: dict[str, int] = self.hps.data.spk2id
|
||||
self.id2spk: dict[int, str] = {v: k for k, v in self.spk2id.items()}
|
||||
|
||||
self.num_styles: int = self.hps.data.num_styles
|
||||
if hasattr(self.hps.data, "style2id"):
|
||||
self.style2id: Dict[str, int] = self.hps.data.style2id
|
||||
self.style2id: dict[str, int] = self.hps.data.style2id
|
||||
else:
|
||||
self.style2id: Dict[str, int] = {str(i): i for i in range(self.num_styles)}
|
||||
self.style2id: dict[str, int] = {str(i): i for i in range(self.num_styles)}
|
||||
if len(self.style2id) != self.num_styles:
|
||||
raise ValueError(
|
||||
f"Number of styles ({self.num_styles}) does not match the number of style2id ({len(self.style2id)})"
|
||||
@@ -86,7 +87,7 @@ class Model:
|
||||
|
||||
def load_net_g(self):
|
||||
self.net_g = get_net_g(
|
||||
model_path=self.model_path,
|
||||
model_path=str(self.model_path),
|
||||
version=self.hps.version,
|
||||
device=self.device,
|
||||
hps=self.hps,
|
||||
@@ -128,6 +129,7 @@ class Model:
|
||||
given_tone: Optional[list[int]] = None,
|
||||
pitch_scale: float = 1.0,
|
||||
intonation_scale: float = 1.0,
|
||||
ignore_unknown: bool = False,
|
||||
) -> tuple[int, np.ndarray]:
|
||||
logger.info(f"Start generating audio data from text:\n{text}")
|
||||
if language != "JP" and self.hps.version.endswith("JP-Extra"):
|
||||
@@ -165,6 +167,7 @@ class Model:
|
||||
assist_text_weight=assist_text_weight,
|
||||
style_vec=style_vector,
|
||||
given_tone=given_tone,
|
||||
ignore_unknown=ignore_unknown,
|
||||
)
|
||||
else:
|
||||
texts = text.split("\n")
|
||||
@@ -187,14 +190,12 @@ class Model:
|
||||
assist_text=assist_text,
|
||||
assist_text_weight=assist_text_weight,
|
||||
style_vec=style_vector,
|
||||
ignore_unknown=ignore_unknown,
|
||||
)
|
||||
)
|
||||
if i != len(texts) - 1:
|
||||
audios.append(np.zeros(int(44100 * split_interval)))
|
||||
audio = np.concatenate(audios)
|
||||
with warnings.catch_warnings():
|
||||
warnings.simplefilter("ignore")
|
||||
audio = convert_to_16_bit_wav(audio)
|
||||
logger.info("Audio data generated successfully")
|
||||
fs, audio = adjust_voice(
|
||||
fs=self.hps.data.sampling_rate,
|
||||
@@ -202,46 +203,79 @@ class Model:
|
||||
pitch_scale=pitch_scale,
|
||||
intonation_scale=intonation_scale,
|
||||
)
|
||||
with warnings.catch_warnings():
|
||||
warnings.simplefilter("ignore")
|
||||
audio = convert_to_16_bit_wav(audio)
|
||||
return (fs, audio)
|
||||
|
||||
|
||||
class ModelHolder:
|
||||
def __init__(self, root_dir: str, device: str):
|
||||
self.root_dir: str = root_dir
|
||||
def __init__(self, root_dir: Path, device: str):
|
||||
self.root_dir: Path = root_dir
|
||||
self.device: str = device
|
||||
self.model_files_dict: Dict[str, List[str]] = {}
|
||||
self.model_files_dict: dict[str, list[Path]] = {}
|
||||
self.current_model: Optional[Model] = None
|
||||
self.model_names: List[str] = []
|
||||
self.models: List[Model] = []
|
||||
self.model_names: list[str] = []
|
||||
self.models: list[Model] = []
|
||||
self.refresh()
|
||||
|
||||
def refresh(self):
|
||||
self.model_files_dict = {}
|
||||
self.model_names = []
|
||||
self.current_model = None
|
||||
model_dirs = [
|
||||
d
|
||||
for d in os.listdir(self.root_dir)
|
||||
if os.path.isdir(os.path.join(self.root_dir, d))
|
||||
]
|
||||
for model_name in model_dirs:
|
||||
model_dir = os.path.join(self.root_dir, model_name)
|
||||
|
||||
model_dirs = [d for d in self.root_dir.iterdir() if d.is_dir()]
|
||||
for model_dir in model_dirs:
|
||||
model_files = [
|
||||
os.path.join(model_dir, f)
|
||||
for f in os.listdir(model_dir)
|
||||
if f.endswith(".pth") or f.endswith(".pt") or f.endswith(".safetensors")
|
||||
f
|
||||
for f in model_dir.iterdir()
|
||||
if f.suffix in [".pth", ".pt", ".safetensors"]
|
||||
]
|
||||
if len(model_files) == 0:
|
||||
logger.warning(
|
||||
f"No model files found in {self.root_dir}/{model_name}, so skip it"
|
||||
)
|
||||
logger.warning(f"No model files found in {model_dir}, so skip it")
|
||||
continue
|
||||
self.model_files_dict[model_name] = model_files
|
||||
self.model_names.append(model_name)
|
||||
self.model_files_dict[model_dir.name] = model_files
|
||||
self.model_names.append(model_dir.name)
|
||||
|
||||
def models_info(self):
|
||||
if hasattr(self, "_models_info"):
|
||||
return self._models_info
|
||||
result = []
|
||||
for name, files in self.model_files_dict.items():
|
||||
# Get styles
|
||||
config_path = self.root_dir / name / "config.json"
|
||||
hps = utils.get_hparams_from_file(config_path)
|
||||
style2id: dict[str, int] = hps.data.style2id
|
||||
styles = list(style2id.keys())
|
||||
result.append(
|
||||
{
|
||||
"name": name,
|
||||
"files": [str(f) for f in files],
|
||||
"styles": styles,
|
||||
}
|
||||
)
|
||||
self._models_info = result
|
||||
return result
|
||||
|
||||
def load_model(self, model_name: str, model_path_str: str):
|
||||
model_path = Path(model_path_str)
|
||||
if model_name not in self.model_files_dict:
|
||||
raise ValueError(f"Model `{model_name}` is not found")
|
||||
if model_path not in self.model_files_dict[model_name]:
|
||||
raise ValueError(f"Model file `{model_path}` is not found")
|
||||
if self.current_model is None or self.current_model.model_path != model_path:
|
||||
self.current_model = Model(
|
||||
model_path=model_path,
|
||||
config_path=self.root_dir / model_name / "config.json",
|
||||
style_vec_path=self.root_dir / model_name / "style_vectors.npy",
|
||||
device=self.device,
|
||||
)
|
||||
return self.current_model
|
||||
|
||||
def load_model_gr(
|
||||
self, model_name: str, model_path: str
|
||||
self, model_name: str, model_path_str: str
|
||||
) -> tuple[gr.Dropdown, gr.Button, gr.Dropdown]:
|
||||
model_path = Path(model_path_str)
|
||||
if model_name not in self.model_files_dict:
|
||||
raise ValueError(f"Model `{model_name}` is not found")
|
||||
if model_path not in self.model_files_dict[model_name]:
|
||||
@@ -260,8 +294,8 @@ class ModelHolder:
|
||||
)
|
||||
self.current_model = Model(
|
||||
model_path=model_path,
|
||||
config_path=os.path.join(self.root_dir, model_name, "config.json"),
|
||||
style_vec_path=os.path.join(self.root_dir, model_name, "style_vectors.npy"),
|
||||
config_path=self.root_dir / model_name / "config.json",
|
||||
style_vec_path=self.root_dir / model_name / "style_vectors.npy",
|
||||
device=self.device,
|
||||
)
|
||||
speakers = list(self.current_model.spk2id.keys())
|
||||
|
||||
15
infer.py
15
infer.py
@@ -52,9 +52,12 @@ def get_text(
|
||||
assist_text=None,
|
||||
assist_text_weight=0.7,
|
||||
given_tone=None,
|
||||
ignore_unknown=False,
|
||||
):
|
||||
use_jp_extra = hps.version.endswith("JP-Extra")
|
||||
norm_text, phone, tone, word2ph = clean_text(text, language_str, use_jp_extra)
|
||||
norm_text, phone, tone, word2ph = clean_text(
|
||||
text, language_str, use_jp_extra, ignore_unknown=ignore_unknown
|
||||
)
|
||||
if given_tone is not None:
|
||||
if len(given_tone) != len(phone):
|
||||
raise InvalidToneError(
|
||||
@@ -71,7 +74,13 @@ def get_text(
|
||||
word2ph[i] = word2ph[i] * 2
|
||||
word2ph[0] += 1
|
||||
bert_ori = get_bert(
|
||||
norm_text, word2ph, language_str, device, assist_text, assist_text_weight
|
||||
norm_text,
|
||||
word2ph,
|
||||
language_str,
|
||||
device,
|
||||
assist_text,
|
||||
assist_text_weight,
|
||||
ignore_unknown,
|
||||
)
|
||||
del word2ph
|
||||
assert bert_ori.shape[-1] == len(phone), phone
|
||||
@@ -118,6 +127,7 @@ def infer(
|
||||
assist_text=None,
|
||||
assist_text_weight=0.7,
|
||||
given_tone=None,
|
||||
ignore_unknown=False,
|
||||
):
|
||||
is_jp_extra = hps.version.endswith("JP-Extra")
|
||||
bert, ja_bert, en_bert, phones, tones, lang_ids = get_text(
|
||||
@@ -128,6 +138,7 @@ def infer(
|
||||
assist_text=assist_text,
|
||||
assist_text_weight=assist_text_weight,
|
||||
given_tone=given_tone,
|
||||
ignore_unknown=ignore_unknown,
|
||||
)
|
||||
if skip_start:
|
||||
phones = phones[3:]
|
||||
|
||||
285
server_editor.py
Normal file
285
server_editor.py
Normal file
@@ -0,0 +1,285 @@
|
||||
import csv
|
||||
import os
|
||||
import sys
|
||||
from io import BytesIO
|
||||
from pathlib import Path
|
||||
|
||||
import numpy as np
|
||||
import pyopenjtalk
|
||||
import uvicorn
|
||||
from fastapi import FastAPI, HTTPException, status
|
||||
from fastapi.middleware.cors import CORSMiddleware
|
||||
from fastapi.responses import JSONResponse, Response
|
||||
from fastapi.staticfiles import StaticFiles
|
||||
from pydantic import BaseModel
|
||||
from scipy.io import wavfile
|
||||
|
||||
from common.constants import (
|
||||
DEFAULT_ASSIST_TEXT_WEIGHT,
|
||||
DEFAULT_LENGTH,
|
||||
DEFAULT_NOISE,
|
||||
DEFAULT_NOISEW,
|
||||
DEFAULT_SDP_RATIO,
|
||||
DEFAULT_STYLE,
|
||||
DEFAULT_STYLE_WEIGHT,
|
||||
USER_DICT_CSV_PATH,
|
||||
USER_DICT_PATH,
|
||||
Languages,
|
||||
LATEST_VERSION,
|
||||
)
|
||||
from common.log import logger
|
||||
from common.tts_model import ModelHolder
|
||||
from text.japanese import g2kata_tone, kata_tone2phone_tone
|
||||
|
||||
|
||||
class AudioResponse(Response):
|
||||
media_type = "audio/wav"
|
||||
|
||||
|
||||
origins = [
|
||||
"http://localhost:3000",
|
||||
"http://localhost:8000",
|
||||
"http://127.0.0.1:3000",
|
||||
"http://127.0.0.1:8000",
|
||||
]
|
||||
|
||||
device = "cuda"
|
||||
model_dir = Path("model_assets/server_test")
|
||||
model_holder = ModelHolder(model_dir, device)
|
||||
if len(model_holder.model_names) == 0:
|
||||
logger.error(f"Models not found in {model_dir}.")
|
||||
sys.exit(1)
|
||||
|
||||
app = FastAPI()
|
||||
|
||||
|
||||
app.add_middleware(
|
||||
CORSMiddleware,
|
||||
allow_origins=origins,
|
||||
allow_credentials=True,
|
||||
allow_methods=["*"],
|
||||
allow_headers=["*"],
|
||||
)
|
||||
|
||||
|
||||
@app.get("/version")
|
||||
def version() -> str:
|
||||
return LATEST_VERSION
|
||||
|
||||
|
||||
class MoraTone(BaseModel):
|
||||
mora: str
|
||||
tone: int
|
||||
|
||||
|
||||
class G2PRequest(BaseModel):
|
||||
text: str
|
||||
|
||||
|
||||
@app.post("/g2p")
|
||||
async def read_item(item: G2PRequest):
|
||||
try:
|
||||
kata_tone_list = g2kata_tone(item.text, ignore_unknown=True)
|
||||
except Exception as e:
|
||||
raise HTTPException(
|
||||
status_code=400,
|
||||
detail=f"Failed to convert {item.text} to katakana and tone, {e}",
|
||||
)
|
||||
return [MoraTone(mora=kata, tone=tone) for kata, tone in kata_tone_list]
|
||||
|
||||
|
||||
@app.get("/models_info")
|
||||
def models_info():
|
||||
return model_holder.models_info()
|
||||
|
||||
|
||||
class SynthesisRequest(BaseModel):
|
||||
model: str
|
||||
modelFile: str
|
||||
text: str
|
||||
moraToneList: list[MoraTone]
|
||||
style: str = DEFAULT_STYLE
|
||||
styleWeight: float = DEFAULT_STYLE_WEIGHT
|
||||
assistText: str = ""
|
||||
assistTextWeight: float = DEFAULT_ASSIST_TEXT_WEIGHT
|
||||
speed: float = 1.0
|
||||
noise: float = DEFAULT_NOISE
|
||||
noisew: float = DEFAULT_NOISEW
|
||||
sdpRatio: float = DEFAULT_SDP_RATIO
|
||||
language: Languages = Languages.JP
|
||||
silenceAfter: float = 0.5
|
||||
pitchScale: float = 1.0
|
||||
intonationScale: float = 1.0
|
||||
|
||||
|
||||
@app.post("/synthesis", response_class=AudioResponse)
|
||||
def synthesis(request: SynthesisRequest):
|
||||
try:
|
||||
model = model_holder.load_model(
|
||||
model_name=request.model, model_path_str=request.modelFile
|
||||
)
|
||||
except Exception as e:
|
||||
logger.error(e)
|
||||
raise HTTPException(
|
||||
status_code=500,
|
||||
detail=f"Failed to load model {request.model} from {request.modelFile}, {e}",
|
||||
)
|
||||
text = request.text
|
||||
kata_tone_list = [
|
||||
(mora_tone.mora, mora_tone.tone) for mora_tone in request.moraToneList
|
||||
]
|
||||
phone_tone = kata_tone2phone_tone(kata_tone_list)
|
||||
tone = [t for _, t in phone_tone]
|
||||
sr, audio = model.infer(
|
||||
text=text,
|
||||
language=request.language.value,
|
||||
sdp_ratio=request.sdpRatio,
|
||||
noise=request.noise,
|
||||
noisew=request.noisew,
|
||||
length=1 / request.speed,
|
||||
given_tone=tone,
|
||||
style=request.style,
|
||||
style_weight=request.styleWeight,
|
||||
assist_text=request.assistText,
|
||||
assist_text_weight=request.assistTextWeight,
|
||||
use_assist_text=bool(request.assistText),
|
||||
line_split=False,
|
||||
ignore_unknown=True,
|
||||
pitch_scale=request.pitchScale,
|
||||
intonation_scale=request.intonationScale,
|
||||
)
|
||||
|
||||
with BytesIO() as wavContent:
|
||||
wavfile.write(wavContent, sr, audio)
|
||||
return Response(content=wavContent.getvalue(), media_type="audio/wav")
|
||||
|
||||
|
||||
class MultiSynthesisRequest(BaseModel):
|
||||
lines: list[SynthesisRequest]
|
||||
|
||||
|
||||
@app.post("/multi-synthesis", response_class=AudioResponse)
|
||||
def multi_synthesis(request: MultiSynthesisRequest):
|
||||
lines = request.lines
|
||||
audios = []
|
||||
for i, req in enumerate(lines):
|
||||
# Loade model
|
||||
try:
|
||||
model = model_holder.load_model(
|
||||
model_name=req.model, model_path_str=req.modelFile
|
||||
)
|
||||
except Exception as e:
|
||||
logger.error(e)
|
||||
raise HTTPException(
|
||||
status_code=500,
|
||||
detail=f"Failed to load model {req.model} from {req.modelFile}, {e}",
|
||||
)
|
||||
text = req.text
|
||||
kata_tone_list = [
|
||||
(mora_tone.mora, mora_tone.tone) for mora_tone in req.moraToneList
|
||||
]
|
||||
phone_tone = kata_tone2phone_tone(kata_tone_list)
|
||||
tone = [t for _, t in phone_tone]
|
||||
sr, audio = model.infer(
|
||||
text=text,
|
||||
language=req.language.value,
|
||||
sdp_ratio=req.sdpRatio,
|
||||
noise=req.noise,
|
||||
noisew=req.noisew,
|
||||
length=1 / req.speed,
|
||||
given_tone=tone,
|
||||
style=req.style,
|
||||
style_weight=req.styleWeight,
|
||||
assist_text=req.assistText,
|
||||
assist_text_weight=req.assistTextWeight,
|
||||
use_assist_text=bool(req.assistText),
|
||||
line_split=False,
|
||||
ignore_unknown=True,
|
||||
pitch_scale=req.pitchScale,
|
||||
intonation_scale=req.intonationScale,
|
||||
)
|
||||
audios.append(audio)
|
||||
if i < len(lines) - 1:
|
||||
silence = int(sr * req.silenceAfter)
|
||||
audios.append(np.zeros(silence, dtype=np.int16))
|
||||
audio = np.concatenate(audios)
|
||||
logger.debug(audio.dtype)
|
||||
|
||||
with BytesIO() as wavContent:
|
||||
wavfile.write(wavContent, sr, audio)
|
||||
return Response(content=wavContent.getvalue(), media_type="audio/wav")
|
||||
|
||||
|
||||
class AddUserDictWordRequest(BaseModel):
|
||||
surface: str
|
||||
pronunciation: str
|
||||
accentType: str # アクセント核位置/モーラ数、例:1/3, アクセント核位置は1から始まる
|
||||
priority: int = 5
|
||||
|
||||
|
||||
# コストの値は以下の値を参考にしている
|
||||
# https://github.com/VOICEVOX/voicevox_engine/blob/master/voicevox_engine/user_dict/part_of_speech_data.py
|
||||
COST_CANDIDATES = [-988, 3488, 4768, 6048, 7328, 8609, 8734, 8859, 8984, 9110, 14176]
|
||||
|
||||
|
||||
@app.post("/user_dict_word")
|
||||
def add_user_dict_word(request: AddUserDictWordRequest):
|
||||
if request.surface == "" or request.pronunciation == "":
|
||||
return JSONResponse(
|
||||
status_code=status.HTTP_400_BAD_REQUEST,
|
||||
content={"message": "単語か読みが空です。"},
|
||||
)
|
||||
user_dict_file = Path(USER_DICT_CSV_PATH)
|
||||
user_dict_file.parent.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
# 新規追加または更新する単語のCSV行
|
||||
new_csv_row = f"{request.surface},,,{COST_CANDIDATES[request.priority]},名詞,固有名詞,一般,*,*,*,{request.surface},{request.pronunciation},{request.pronunciation},{request.accentType},*\n"
|
||||
found = False
|
||||
updated = False
|
||||
|
||||
# ユーザー辞書ファイルが存在する場合、既存の単語をチェックし、必要に応じて更新する
|
||||
if user_dict_file.exists():
|
||||
with user_dict_file.open(encoding="utf-8") as f:
|
||||
lines = f.readlines()
|
||||
|
||||
with user_dict_file.open("w", encoding="utf-8") as f:
|
||||
for line in lines:
|
||||
if line.split(",")[0] == request.surface:
|
||||
found = True
|
||||
if line.strip() != new_csv_row.strip():
|
||||
f.write(new_csv_row)
|
||||
updated = True
|
||||
else:
|
||||
f.write(line)
|
||||
else:
|
||||
f.write(line)
|
||||
# 単語が新しい場合、新規に追加
|
||||
if not found:
|
||||
with user_dict_file.open("a", encoding="utf-8") as f:
|
||||
f.write(new_csv_row)
|
||||
|
||||
pyopenjtalk.unset_user_dict()
|
||||
pyopenjtalk.mecab_dict_index(USER_DICT_CSV_PATH, USER_DICT_PATH)
|
||||
pyopenjtalk.update_global_jtalk_with_user_dict(USER_DICT_PATH)
|
||||
|
||||
if updated:
|
||||
return JSONResponse(
|
||||
status_code=status.HTTP_200_OK,
|
||||
content={"message": "単語が既に存在し、更新されました。"},
|
||||
)
|
||||
elif found:
|
||||
return JSONResponse(
|
||||
status_code=status.HTTP_200_OK,
|
||||
content={"message": "既に登録されている単語です。更新はありません。"},
|
||||
)
|
||||
else:
|
||||
return JSONResponse(
|
||||
status_code=status.HTTP_201_CREATED,
|
||||
content={"message": "単語が新規追加されました。"},
|
||||
)
|
||||
|
||||
|
||||
app.mount("/", StaticFiles(directory="Web"), name="static")
|
||||
|
||||
if __name__ == "__main__":
|
||||
uvicorn.run(app, host="0.0.0.0", port=8000)
|
||||
@@ -19,14 +19,25 @@ def cleaned_text_to_sequence(cleaned_text, tones, language):
|
||||
|
||||
|
||||
def get_bert(
|
||||
norm_text, word2ph, language, device, assist_text=None, assist_text_weight=0.7
|
||||
text,
|
||||
word2ph,
|
||||
language,
|
||||
device,
|
||||
assist_text=None,
|
||||
assist_text_weight=0.7,
|
||||
ignore_unknown=False,
|
||||
):
|
||||
from .chinese_bert import get_bert_feature as zh_bert
|
||||
from .english_bert_mock import get_bert_feature as en_bert
|
||||
from .japanese_bert import get_bert_feature as jp_bert
|
||||
|
||||
lang_bert_func_map = {"ZH": zh_bert, "EN": en_bert, "JP": jp_bert}
|
||||
if language == "JP":
|
||||
bert = lang_bert_func_map[language](
|
||||
norm_text, word2ph, device, assist_text, assist_text_weight
|
||||
text, word2ph, device, assist_text, assist_text_weight, ignore_unknown
|
||||
)
|
||||
else:
|
||||
bert = lang_bert_func_map[language](
|
||||
text, word2ph, device, assist_text, assist_text_weight
|
||||
)
|
||||
return bert
|
||||
|
||||
@@ -4,11 +4,13 @@ from text import chinese, japanese, english, cleaned_text_to_sequence
|
||||
language_module_map = {"ZH": chinese, "JP": japanese, "EN": english}
|
||||
|
||||
|
||||
def clean_text(text, language, use_jp_extra=True):
|
||||
def clean_text(text, language, use_jp_extra=True, ignore_unknown=False):
|
||||
language_module = language_module_map[language]
|
||||
norm_text = language_module.text_normalize(text)
|
||||
if language == "JP":
|
||||
phones, tones, word2ph = language_module.g2p(norm_text, use_jp_extra)
|
||||
phones, tones, word2ph = language_module.g2p(
|
||||
norm_text, use_jp_extra, ignore_unknown=ignore_unknown
|
||||
)
|
||||
else:
|
||||
phones, tones, word2ph = language_module.g2p(norm_text)
|
||||
return norm_text, phones, tones, word2ph
|
||||
|
||||
@@ -2,11 +2,13 @@
|
||||
# compatible with Julius https://github.com/julius-speech/segmentation-kit
|
||||
import re
|
||||
import unicodedata
|
||||
from pathlib import Path
|
||||
|
||||
import pyopenjtalk
|
||||
from num2words import num2words
|
||||
from transformers import AutoTokenizer
|
||||
|
||||
from common.constants import USER_DICT_PATH, USER_DICT_CSV_PATH
|
||||
from common.log import logger
|
||||
from text import punctuation
|
||||
from text.japanese_mora_list import (
|
||||
@@ -14,6 +16,11 @@ from text.japanese_mora_list import (
|
||||
mora_phonemes_to_mora_kata,
|
||||
)
|
||||
|
||||
if Path(USER_DICT_CSV_PATH).exists():
|
||||
pyopenjtalk.mecab_dict_index(USER_DICT_CSV_PATH, USER_DICT_PATH)
|
||||
if Path(USER_DICT_PATH).exists():
|
||||
pyopenjtalk.update_global_jtalk_with_user_dict(USER_DICT_PATH)
|
||||
|
||||
# 子音の集合
|
||||
COSONANTS = set(
|
||||
[
|
||||
@@ -160,7 +167,7 @@ def japanese_convert_numbers_to_words(text: str) -> str:
|
||||
|
||||
|
||||
def g2p(
|
||||
norm_text: str, use_jp_extra: bool = True
|
||||
norm_text: str, use_jp_extra: bool = True, ignore_unknown: bool = False
|
||||
) -> tuple[list[str], list[int], list[int]]:
|
||||
"""
|
||||
他で使われるメインの関数。`text_normalize()`で正規化された`norm_text`を受け取り、
|
||||
@@ -182,7 +189,7 @@ def g2p(
|
||||
|
||||
# sep_text: 単語単位の単語のリスト
|
||||
# sep_kata: 単語単位の単語のカタカナ読みのリスト
|
||||
sep_text, sep_kata = text2sep_kata(norm_text)
|
||||
sep_text, sep_kata = text2sep_kata(norm_text, ignore_unknown=ignore_unknown)
|
||||
|
||||
# sep_phonemes: 各単語ごとの音素のリストのリスト
|
||||
sep_phonemes = handle_long([kata2phoneme_list(i) for i in sep_kata])
|
||||
@@ -231,8 +238,8 @@ def g2p(
|
||||
return phones, tones, word2ph
|
||||
|
||||
|
||||
def g2kata_tone(norm_text: str) -> list[tuple[str, int]]:
|
||||
phones, tones, _ = g2p(norm_text, use_jp_extra=True)
|
||||
def g2kata_tone(norm_text: str, ignore_unknown: bool = False) -> list[tuple[str, int]]:
|
||||
phones, tones, _ = g2p(norm_text, use_jp_extra=True, ignore_unknown=ignore_unknown)
|
||||
return phone_tone2kata_tone(list(zip(phones, tones)))
|
||||
|
||||
|
||||
@@ -325,7 +332,9 @@ def g2phone_tone_wo_punct(text: str) -> list[tuple[str, int]]:
|
||||
return result
|
||||
|
||||
|
||||
def text2sep_kata(norm_text: str) -> tuple[list[str], list[str]]:
|
||||
def text2sep_kata(
|
||||
norm_text: str, ignore_unknown: bool = False
|
||||
) -> tuple[list[str], list[str]]:
|
||||
"""
|
||||
`text_normalize`で正規化済みの`norm_text`を受け取り、それを単語分割し、
|
||||
分割された単語リストとその読み(カタカナor記号1文字)のリストのタプルを返す。
|
||||
@@ -361,6 +370,9 @@ def text2sep_kata(norm_text: str) -> tuple[list[str], list[str]]:
|
||||
# wordは正規化されているので、`.`, `,`, `!`, `'`, `-`, `--` のいずれか
|
||||
if not set(word).issubset(set(punctuation)): # 記号繰り返しか判定
|
||||
# ここはpyopenjtalkが読めない文字等のときに起こる
|
||||
if ignore_unknown:
|
||||
logger.error(f"Ignoring unknown: {word} in:\n{norm_text}")
|
||||
continue
|
||||
raise ValueError(f"Cannot read: {word} in:\n{norm_text}")
|
||||
# yomiは元の記号のままに変更
|
||||
yomi = word
|
||||
@@ -500,6 +512,9 @@ def handle_long(sep_phonemes: list[list[str]]) -> list[list[str]]:
|
||||
おそらく長音記号とダッシュを勘違いしていると思われるので、ダッシュに対応する音素`-`に変換する。
|
||||
"""
|
||||
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]
|
||||
|
||||
@@ -4,7 +4,7 @@ import torch
|
||||
from transformers import AutoModelForMaskedLM, AutoTokenizer
|
||||
|
||||
from config import config
|
||||
from text.japanese import text2sep_kata
|
||||
from text.japanese import text2sep_kata, text_normalize
|
||||
|
||||
LOCAL_PATH = "./bert/deberta-v2-large-japanese-char-wwm"
|
||||
|
||||
@@ -19,8 +19,10 @@ def get_bert_feature(
|
||||
device=config.bert_gen_config.device,
|
||||
assist_text=None,
|
||||
assist_text_weight=0.7,
|
||||
ignore_unknown=False,
|
||||
):
|
||||
text = "".join(text2sep_kata(text)[0])
|
||||
text = "".join(text2sep_kata(text, ignore_unknown=ignore_unknown)[0])
|
||||
# text = text_normalize(text)
|
||||
if assist_text:
|
||||
assist_text = "".join(text2sep_kata(assist_text)[0])
|
||||
if (
|
||||
|
||||
Reference in New Issue
Block a user