diff --git a/common/subprocess_utils.py b/common/subprocess_utils.py index 9a0e323..40426f7 100644 --- a/common/subprocess_utils.py +++ b/common/subprocess_utils.py @@ -14,6 +14,7 @@ def run_script_with_log(cmd: list[str], ignore_warning=False) -> tuple[bool, str stdout=SAFE_STDOUT, # type: ignore stderr=subprocess.PIPE, text=True, + encoding="utf-8", ) if result.returncode != 0: logger.error(f"Error: {' '.join(cmd)}\n{result.stderr}") diff --git a/preprocess_text.py b/preprocess_text.py index e0c3399..b3aaa17 100644 --- a/preprocess_text.py +++ b/preprocess_text.py @@ -7,10 +7,10 @@ from typing import Optional import click from tqdm import tqdm +from common.log import logger +from common.stdout_wrapper import SAFE_STDOUT from config import config from text.cleaner import clean_text -from common.stdout_wrapper import SAFE_STDOUT -from common.log import logger preprocess_text_config = config.preprocess_text_config @@ -89,7 +89,10 @@ def preprocess( ) ) except Exception as e: - logger.error(f"An error occurred at line:\n{line.strip()}\n{e}") + logger.error( + f"An error occurred at line:\n{line.strip()}\n{e}", + encoding="utf-8", + ) with open(error_log_path, "a", encoding="utf-8") as error_log: error_log.write(f"{line.strip()}\n{e}\n\n") error_count += 1 @@ -172,8 +175,9 @@ def preprocess( f"An error occurred in {error_count} lines. Please check {error_log_path} for details." ) raise Exception( - f"An error occurred in {error_count} lines. Please check {error_log_path} for details." + f"An error occurred in {error_count} lines. Please check `Data/you_model_name/text_error.log` file for details." ) + # 何故か{error_log_path}をraiseすると文字コードエラーが起きるので上のように書いている else: logger.info( "Training set and validation set generation from texts is complete!" diff --git a/server_editor.py b/server_editor.py index afb9321..1ae323f 100644 --- a/server_editor.py +++ b/server_editor.py @@ -19,7 +19,6 @@ from pathlib import Path import yaml import numpy as np -import pyopenjtalk import requests import torch import uvicorn diff --git a/text/japanese.py b/text/japanese.py index b18bc68..fea0eaa 100644 --- a/text/japanese.py +++ b/text/japanese.py @@ -4,7 +4,9 @@ import re import unicodedata from pathlib import Path -import pyopenjtalk +from . import pyopenjtalk_worker as pyopenjtalk + +pyopenjtalk.initialize() from num2words import num2words from transformers import AutoTokenizer diff --git a/text/pyopenjtalk_worker/__init__.py b/text/pyopenjtalk_worker/__init__.py new file mode 100644 index 0000000..9677666 --- /dev/null +++ b/text/pyopenjtalk_worker/__init__.py @@ -0,0 +1,126 @@ +""" +Run the pyopenjtalk worker in a separate process +to avoid user dictionary access error +""" + +from typing import Optional, Any + +from .worker_common import WOKER_PORT +from .worker_client import WorkerClient + +from common.log import logger + +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) -> 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 = WOKER_PORT): + 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, frame): + terminate() + + signal.signal(signal.SIGTERM, signal_handler) + + +# top-level declaration +def terminate(): + 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 diff --git a/text/pyopenjtalk_worker/__main__.py b/text/pyopenjtalk_worker/__main__.py new file mode 100644 index 0000000..8b67aa0 --- /dev/null +++ b/text/pyopenjtalk_worker/__main__.py @@ -0,0 +1,16 @@ +import argparse + +from .worker_server import WorkerServer +from .worker_common import WOKER_PORT + + +def main(): + parser = argparse.ArgumentParser() + parser.add_argument("--port", type=int, default=WOKER_PORT) + args = parser.parse_args() + server = WorkerServer() + server.start_server(port=args.port) + + +if __name__ == "__main__": + main() diff --git a/text/pyopenjtalk_worker/worker_client.py b/text/pyopenjtalk_worker/worker_client.py new file mode 100644 index 0000000..86d8969 --- /dev/null +++ b/text/pyopenjtalk_worker/worker_client.py @@ -0,0 +1,55 @@ +from typing import Any +import socket + +from .worker_common import RequestType, receive_data, send_data + +from common.log import logger + + +class WorkerClient: + 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): + return self + + def __exit__(self, exc_type, exc_value, traceback): + self.close() + + def close(self): + self.sock.close() + + def dispatch_pyopenjtalk(self, func: str, *args, **kwargs): + 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): + 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 response.get("client-count") + + def quit_server(self): + 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}") diff --git a/text/pyopenjtalk_worker/worker_common.py b/text/pyopenjtalk_worker/worker_common.py new file mode 100644 index 0000000..bea552e --- /dev/null +++ b/text/pyopenjtalk_worker/worker_common.py @@ -0,0 +1,44 @@ +from typing import Any, Optional, Final +from enum import IntEnum, auto +import socket +import json + +WOKER_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()) diff --git a/text/pyopenjtalk_worker/worker_server.py b/text/pyopenjtalk_worker/worker_server.py new file mode 100644 index 0000000..6babd91 --- /dev/null +++ b/text/pyopenjtalk_worker/worker_server.py @@ -0,0 +1,118 @@ +import pyopenjtalk +import socket +import select +import time + +from .worker_common import ( + ConnectionClosedException, + RequestType, + receive_data, + send_data, +) + +from common.log import logger + +# 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: + def __init__(self) -> None: + self.client_count: int = 0 + self.quit: bool = False + + def handle_request(self, request): + request_type = None + try: + request_type = RequestType(request.get("request-type")) + except Exception: + return { + "success": False, + "reason": "request-type is invalid", + } + + 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): + 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 diff --git a/text/user_dict/__init__.py b/text/user_dict/__init__.py index c12b3d1..95515cb 100644 --- a/text/user_dict/__init__.py +++ b/text/user_dict/__init__.py @@ -12,7 +12,9 @@ from typing import Dict, List, Optional from uuid import UUID, uuid4 import numpy as np -import pyopenjtalk +from .. import pyopenjtalk_worker as pyopenjtalk + +pyopenjtalk.initialize() from fastapi import HTTPException from .word_model import UserDictWord, WordTypes diff --git a/transcribe.py b/transcribe.py index c28c8db..b3f9014 100644 --- a/transcribe.py +++ b/transcribe.py @@ -61,13 +61,9 @@ if __name__ == "__main__": logger.warning(f"Failed to load model, so use `auto` compute_type: {e}") model = WhisperModel(args.model, device=device) - # wav_files = [ - # os.path.join(input_dir, f) for f in os.listdir(input_dir) if f.endswith(".wav") - # ] wav_files = [f for f in input_dir.rglob("*.wav") if f.is_file()] if output_file.exists(): logger.warning(f"{output_file} exists, backing up to {output_file}.bak") - # if os.path.exists(output_file + ".bak"): backup_path = output_file.with_name(output_file.name + ".bak") if backup_path.exists(): logger.warning(f"{output_file}.bak exists, deleting...")