Merge pull request #89 from kale4eat/dev-pyopenjtalk-worker

複数プロセス利用時にユーザー辞書を活用するための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 @@ 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!"

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

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
 

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

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

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

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

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