複数プロセス利用時にユーザー辞書を活用するためのpyopenjtalk別プロセス化

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}")

preprocess_text.py

@@ -7,10 +7,10 @@
 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!"

server_editor.py

@@ -19,7 +19,6 @@
 import yaml
 
 import numpy as np
-import pyopenjtalk
 import requests
 import torch
 import uvicorn

text/japanese.py

@@ -4,7 +4,9 @@
 import unicodedata
 from pathlib import Path
 
-import pyopenjtalk
+from . import pyopenjtalk_worker as pyopenjtalk
+
+pyopenjtalk.initialize()
 from num2words import num2words
 from transformers import AutoTokenizer
 

text/pyopenjtalk_worker/__init__.py

@@ -0,0 +1,129 @@
+"""
+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):
+        with open("signal_handler.txt", mode="w") as f:
+
+            pass
+        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

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()

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}")

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())