every_holiday/VRCT
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge branch 'develop' into translate_api

Browse Source 
# Conflicts:
#	requirements.txt
#	requirements_cuda.txt
#	src-python/config.py
#	src-python/mainloop.py
#	src-python/model.py
#	src-python/models/osc/osc.py
#	src-python/models/translation/translation_languages.py
#	src-python/models/translation/translation_translator.py
#	src-python/models/translation/translation_utils.py
This commit is contained in:
misyaguziya
2025-10-14 12:47:47 +09:00
parent 224eaf3cef b0403cfa70
commit 6fc89fbb3e
86 changed files with 20001 additions and 1584 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
src-python/models/__init__.py Normal file
View File

@@ -0,0 +1,5 @@
"""models package init for static analysis and packaging."""
__all__ = [
# subpackages are discovered implicitly
]

172
src-python/models/osc/osc.py
View File

@@ -1,82 +1,116 @@
import asyncio
from typing import Any
"""OSC helpers and a thin OSCQuery-enabled server wrapper.
This module provides `OSCHandler`, a convenience wrapper used by the
application to send OSC messages and expose OSCQuery endpoints when the
target address is localhost. The implementation is defensive: missing
utilities are handled gracefully and logging helpers are used where
available.
"""
import time
from typing import Any, Callable, Dict, Optional
from time import sleep
from threading import Thread
from pythonosc import udp_client, dispatcher, osc_server
from tinyoscquery.queryservice import OSCQueryService
from tinyoscquery.query import OSCQueryBrowser, OSCQueryClient
from tinyoscquery.utility import get_open_udp_port, get_open_tcp_port
from tinyoscquery.shared.node import OSCAccess
try:
from tinyoscquery.queryservice import OSCQueryService
from tinyoscquery.query import OSCQueryBrowser, OSCQueryClient
from tinyoscquery.utility import get_open_udp_port, get_open_tcp_port
from tinyoscquery.shared.node import OSCAccess
except Exception:
# tinyoscquery is optional for non-local usage; functionality that
# depends on it will be disabled if it's missing.
OSCQueryService = None # type: ignore
OSCQueryBrowser = None # type: ignore
OSCQueryClient = None # type: ignore
def get_open_udp_port() -> int: # type: ignore
return 0
def get_open_tcp_port() -> int: # type: ignore
return 0
OSCAccess = None # type: ignore
try:
from utils import errorLogging
except ImportError:
def errorLogging():
except Exception:
def errorLogging() -> None:
import traceback
print("Error occurred:", traceback.format_exc())
class OSCHandler:
def __init__(self, ip_address="127.0.0.1", port=9000) -> None:
"""Thin wrapper managing OSC send/receive and optional OSCQuery advertising.
if ip_address in ["127.0.0.1", "localhost"]:
self.is_osc_query_enabled = True
else:
self.is_osc_query_enabled = False
Args:
ip_address: OSC server client target / bind address
port: UDP port to send to
"""
self.osc_ip_address = ip_address
self.osc_port = port
self.osc_parameter_muteself = "/avatar/parameters/MuteSelf"
self.osc_parameter_chatbox_typing = "/chatbox/typing"
self.osc_parameter_chatbox_input = "/chatbox/input"
def __init__(self, ip_address: str = "127.0.0.1", port: int = 9000) -> None:
self.is_osc_query_enabled: bool = ip_address in ["127.0.0.1", "localhost"]
self.osc_ip_address: str = ip_address
self.osc_port: int = port
self.osc_parameter_muteself: str = "/avatar/parameters/MuteSelf"
self.osc_parameter_chatbox_typing: str = "/chatbox/typing"
self.osc_parameter_chatbox_input: str = "/chatbox/input"
self.udp_client = udp_client.SimpleUDPClient(self.osc_ip_address, self.osc_port)
self.osc_server_name = "VRChat-Client"
self.osc_server_name: str = "VRChat-Client"
self.osc_server = None
self.osc_query_service = None
self.osc_query_service_name = "VRCT"
self.osc_server_ip_address = ip_address
self.http_port = None
self.osc_server_port = None
self.dict_filter_and_target = {}
self.osc_query_service_name: str = "VRCT"
self.osc_server_ip_address: str = ip_address
self.http_port: Optional[int] = None
self.osc_server_port: Optional[int] = None
self.dict_filter_and_target: Dict[str, Callable] = {}
self.browser = None
def getIsOscQueryEnabled(self) -> bool:
"""Return whether OSCQuery support is enabled (local addresses only)."""
return self.is_osc_query_enabled
def setOscIpAddress(self, ip_address:str) -> None:
if ip_address in ["127.0.0.1", "localhost"]:
self.is_osc_query_enabled = True
else:
self.is_osc_query_enabled = False
def setOscIpAddress(self, ip_address: str) -> None:
"""Change the OSC target IP address and reinitialize services."""
self.is_osc_query_enabled = ip_address in ["127.0.0.1", "localhost"]
self.oscServerStop()
self.osc_ip_address = ip_address
self.udp_client = udp_client.SimpleUDPClient(self.osc_ip_address, self.osc_port)
self.receiveOscParameters()
def setOscPort(self, port:int) -> None:
def setOscPort(self, port: int) -> None:
"""Change the OSC UDP port used for sending and reinitialize services."""
self.oscServerStop()
self.osc_port = port
self.udp_client = udp_client.SimpleUDPClient(self.osc_ip_address, self.osc_port)
self.receiveOscParameters()
# send OSC message typing
def sendTyping(self, flag:bool=False) -> None:
def sendTyping(self, flag: bool = False) -> None:
"""Send /chatbox/typing with a boolean flag."""
self.udp_client.send_message(self.osc_parameter_chatbox_typing, [flag])
# send OSC message
def sendMessage(self, message:str="", notification:bool=True) -> None:
def sendMessage(self, message: str = "", notification: bool = True) -> None:
"""Send /chatbox/input if message is non-empty.
The second argument historically was a boolean flag for clearing; we keep
compatibility by sending [message, True, notification].
"""
if len(message) > 0:
self.udp_client.send_message(self.osc_parameter_chatbox_input, [f"{message}", True, notification])
def getOSCParameterValue(self, address:str) -> Any:
def getOSCParameterValue(self, address: str) -> Any:
if not self.is_osc_query_enabled:
# OSCQueryが無効な場合はNoneを返す
return None
value = None
value: Any = None
try:
# browserインスタンスを再利用し、毎回の生成と破棄を避ける
if self.browser is None:
# OSCQueryBrowser may not be available; guard
if OSCQueryBrowser is None:
return None
self.browser = OSCQueryBrowser()
sleep(1) # 初回のみスリープ
@@ -84,7 +118,17 @@ class OSCHandler:
if service is not None:
osc_query_client = OSCQueryClient(service)
mute_self_node = osc_query_client.query_node(address)
value = mute_self_node.value[0]
# mute_self_node may be None when the node is not present on the
# remote OSCQuery service. Also mute_self_node.value may be None
# or an empty list. Guard against those cases to avoid
# AttributeError: 'NoneType' object has no attribute 'value'
if mute_self_node is None:
return None
# prefer explicit checks rather than relying on exceptions
node_value = getattr(mute_self_node, 'value', None)
if not node_value:
return None
value = node_value[0]
except Exception:
errorLogging()
# エラー発生時にbrowserをリセットして次回再初期化
@@ -99,15 +143,22 @@ class OSCHandler:
self.browser = None
return value
def getOSCParameterMuteSelf(self) -> bool:
def getOSCParameterMuteSelf(self) -> Optional[bool]:
"""Return the value of the MuteSelf parameter when available, else None."""
return self.getOSCParameterValue(self.osc_parameter_muteself)
def setDictFilterAndTarget(self, dict_filter_and_target:dict) -> None:
def setDictFilterAndTarget(self, dict_filter_and_target: Dict[str, Callable]) -> None:
"""Set the mapping from OSC address filters to handler callables."""
self.dict_filter_and_target = dict_filter_and_target
def receiveOscParameters(self) -> None:
if self.is_osc_query_enabled is False:
# OSCQueryが無効な場合は何もしない
"""Start a local OSC server and advertise OSCQuery endpoints when supported.
If `tinyoscquery` is not available or OSCQuery is disabled, this is a
no-op.
"""
if not self.is_osc_query_enabled or OSCQueryService is None:
# OSCQuery が無効またはライブラリが無い場合は何もしない
return
self.osc_server_port = get_open_udp_port()
@@ -118,30 +169,41 @@ class OSCHandler:
self.osc_server = osc_server.ThreadingOSCUDPServer((self.osc_server_ip_address, self.osc_server_port), osc_dispatcher)
Thread(target=self.oscServerServe, daemon=True).start()
def startOSCQueryService():
while True:
try:
self.osc_query_service = OSCQueryService(self.osc_query_service_name, self.http_port, self.osc_server_port)
for filter, target in self.dict_filter_and_target.items():
while True:
try:
# osc_server_name + UTC timestamp でユニークなサービス名を生成
service_name = f"{self.osc_query_service_name}:{int(time.time())}"
self.osc_query_service = OSCQueryService(service_name, self.http_port, self.osc_server_port)
for filter, target in self.dict_filter_and_target.items():
# OSCAccess may be None when tinyoscquery is not present; guard
if OSCAccess is not None:
self.osc_query_service.advertise_endpoint(filter, access=OSCAccess.READWRITE_VALUE)
break
except Exception:
errorLogging()
sleep(1)
Thread(target=startOSCQueryService, daemon=True).start()
break
except Exception:
errorLogging()
sleep(1)
def oscServerServe(self) -> None:
"""Run the OSC server loop with a longer poll interval to reduce CPU."""
# ポーリング間隔を長くして（2秒から10秒に）CPUの使用率を削減
self.osc_server.serve_forever(10)
if self.osc_server is not None:
self.osc_server.serve_forever(10)
def oscServerStop(self) -> None:
"""Stop and clean up any running OSC server and OSCQuery service."""
if isinstance(self.osc_server, osc_server.ThreadingOSCUDPServer):
self.osc_server.shutdown()
try:
self.osc_server.shutdown()
except Exception:
pass
self.osc_server = None
if isinstance(self.osc_query_service, OSCQueryService):
self.osc_query_service.http_server.shutdown()
if OSCQueryService is not None and isinstance(self.osc_query_service, OSCQueryService):
try:
self.osc_query_service.http_server.shutdown()
except Exception:
pass
self.osc_query_service = None
# browserがある場合はクリーンアップ
# browser がある場合はクリーンアップ
if self.browser is not None:
try:
if hasattr(self.browser, 'zc') and self.browser.zc is not None:

5
src-python/models/overlay/__init__.py Normal file
View File

@@ -0,0 +1,5 @@
"""models.overlay package init for static analysis."""
from . import overlay_utils # re-export helper for ease-of-use in tooling
__all__ = ["overlay_utils"]

81
src-python/models/overlay/overlay.py
View File

@@ -3,6 +3,8 @@ import ctypes
import time
from psutil import process_iter
from threading import Thread
from typing import Any, Dict, Optional, Sequence
import openvr
import numpy as np
from PIL import Image
@@ -18,14 +20,26 @@ try:
except ImportError:
import overlay_utils as utils
def mat34Id(array):
def mat34Id(array: Sequence[Sequence[float]]) -> Any:
"""Convert a 3x4 nested sequence into an openvr.HmdMatrix34_t instance.
Args:
array: 3x4 numeric sequence
Returns:
openvr HmdMatrix34_t compatible object
"""
arr = openvr.HmdMatrix34_t()
for i in range(3):
for j in range(4):
arr[i][j] = array[i][j]
return arr
def getBaseMatrix(x_pos, y_pos, z_pos, x_rotation, y_rotation, z_rotation):
def getBaseMatrix(x_pos: float, y_pos: float, z_pos: float, x_rotation: float, y_rotation: float, z_rotation: float) -> np.ndarray:
"""Create a 3x4 base matrix for an overlay given position and Euler rotations.
Returns a numpy array of shape (3,4).
"""
arr = np.zeros((3, 4))
rot = utils.euler_to_rotation_matrix((x_rotation, y_rotation, z_rotation))
@@ -38,7 +52,7 @@ def getBaseMatrix(x_pos, y_pos, z_pos, x_rotation, y_rotation, z_rotation):
arr[2][3] = - z_pos
return arr
def getHMDBaseMatrix():
def getHMDBaseMatrix() -> np.ndarray:
x_pos = 0.0
y_pos = -0.4
z_pos = 1.0
@@ -48,7 +62,7 @@ def getHMDBaseMatrix():
arr = getBaseMatrix(x_pos, y_pos, z_pos, x_rotation, y_rotation, z_rotation)
return arr
def getLeftHandBaseMatrix():
def getLeftHandBaseMatrix() -> np.ndarray:
x_pos = 0.3
y_pos = 0.1
z_pos = -0.31
@@ -58,7 +72,7 @@ def getLeftHandBaseMatrix():
arr = getBaseMatrix(x_pos, y_pos, z_pos, x_rotation, y_rotation, z_rotation)
return arr
def getRightHandBaseMatrix():
def getRightHandBaseMatrix() -> np.ndarray:
x_pos = -0.3
y_pos = 0.1
z_pos = -0.31
@@ -69,24 +83,25 @@ def getRightHandBaseMatrix():
return arr
class Overlay:
def __init__(self, settings_dict):
self.system = None
self.overlay = None
self.handle = None
self.init_process = False
self.initialized = False
self.loop = False
self.thread_overlay = None
"""Manage OpenVR overlays for multiple sizes (e.g. 'small'/'large')."""
def __init__(self, settings_dict: Dict[str, Dict[str, Any]]) -> None:
self.system: Optional[Any] = None
self.overlay: Optional[Any] = None
self.handle: Dict[str, Any] = {}
self.init_process: bool = False
self.initialized: bool = False
self.loop: bool = False
self.thread_overlay: Optional[Thread] = None
self.settings = {}
self.lastUpdate = {}
self.fadeRatio = {}
self.settings: Dict[str, Dict[str, Any]] = {}
self.lastUpdate: Dict[str, float] = {}
self.fadeRatio: Dict[str, float] = {}
for key, value in settings_dict.items():
self.settings[key] = value
self.lastUpdate[key] = time.monotonic()
self.fadeRatio[key] = 1
self.fadeRatio[key] = 1.0
def init(self):
def init(self) -> None:
try:
self.system = openvr.init(openvr.VRApplication_Background)
self.overlay = openvr.IVROverlay()
@@ -119,7 +134,7 @@ class Overlay:
except Exception:
errorLogging()
def updateImage(self, img, size):
def updateImage(self, img: Image.Image, size: str) -> None:
if self.initialized is True:
width, height = img.size
img = img.tobytes()
@@ -139,7 +154,7 @@ class Overlay:
self.updateOpacity(self.settings[size]["opacity"], size)
self.lastUpdate[size] = time.monotonic()
def clearImage(self, size):
def clearImage(self, size: str) -> None:
if self.initialized is True:
self.updateImage(Image.new("RGBA", (1, 1), (0, 0, 0, 0)), size)
@@ -151,7 +166,7 @@ class Overlay:
r, g, b = col
self.overlay.setOverlayColor(self.handle[size], r, g, b)
def updateOpacity(self, opacity, size, with_fade=False):
def updateOpacity(self, opacity: float, size: str, with_fade: bool = False) -> None:
self.settings[size]["opacity"] = opacity
if self.initialized is True:
@@ -161,12 +176,12 @@ class Overlay:
else:
self.overlay.setOverlayAlpha(self.handle[size], self.settings[size]["opacity"])
def updateUiScaling(self, ui_scaling, size):
def updateUiScaling(self, ui_scaling: float, size: str) -> None:
self.settings[size]["ui_scaling"] = ui_scaling
if self.initialized is True:
self.overlay.setOverlayWidthInMeters(self.handle[size], self.settings[size]["ui_scaling"])
def updatePosition(self, x_pos, y_pos, z_pos, x_rotation, y_rotation, z_rotation, tracker, size):
def updatePosition(self, x_pos: float, y_pos: float, z_pos: float, x_rotation: float, y_rotation: float, z_rotation: float, tracker: str, size: str) -> None:
"""
x_pos, y_pos, z_pos are floats representing the position of overlay
x_rotation, y_rotation, z_rotation are floats representing the rotation of overlay
@@ -208,13 +223,13 @@ class Overlay:
transform
)
def updateDisplayDuration(self, display_duration, size):
def updateDisplayDuration(self, display_duration: float, size: str) -> None:
self.settings[size]["display_duration"] = display_duration
def updateFadeoutDuration(self, fadeout_duration, size):
def updateFadeoutDuration(self, fadeout_duration: float, size: str) -> None:
self.settings[size]["fadeout_duration"] = fadeout_duration
def checkActive(self):
def checkActive(self) -> bool:
try:
if self.system is not None and self.initialized is True:
new_event = openvr.VREvent_t()
@@ -226,7 +241,7 @@ class Overlay:
errorLogging()
return False
def evaluateOpacityFade(self, size):
def evaluateOpacityFade(self, size: str) -> None:
currentTime = time.monotonic()
if (currentTime - self.lastUpdate[size]) > self.settings[size]["display_duration"]:
timeThroughInterval = currentTime - self.lastUpdate[size] - self.settings[size]["display_duration"]
@@ -235,13 +250,13 @@ class Overlay:
self.fadeRatio[size] = 0
self.overlay.setOverlayAlpha(self.handle[size], self.fadeRatio[size] * self.settings[size]["opacity"])
def update(self, size):
def update(self, size: str) -> None:
if self.settings[size]["fadeout_duration"] != 0:
self.evaluateOpacityFade(size)
else:
self.updateOpacity(self.settings[size]["opacity"], size)
def mainloop(self):
def mainloop(self) -> None:
self.loop = True
while self.checkActive() is True and self.loop is True:
startTime = time.monotonic()
@@ -251,21 +266,21 @@ class Overlay:
if sleepTime > 0:
time.sleep(sleepTime)
def main(self):
def main(self) -> None:
while self.checkSteamvrRunning() is False:
time.sleep(10)
self.init()
if self.initialized is True:
self.mainloop()
def startOverlay(self):
def startOverlay(self) -> None:
if self.initialized is False and self.init_process is False:
self.init_process = True
self.thread_overlay = Thread(target=self.main)
self.thread_overlay.daemon = True
self.thread_overlay.start()
def shutdownOverlay(self):
def shutdownOverlay(self) -> None:
if self.initialized is True and self.init_process is False:
if isinstance(self.thread_overlay, Thread):
self.loop = False
@@ -281,7 +296,7 @@ class Overlay:
self.system = None
self.initialized = False
def reStartOverlay(self):
def reStartOverlay(self) -> None:
self.shutdownOverlay()
self.startOverlay()

64
src-python/models/overlay/overlay_image.py
View File

@@ -1,6 +1,6 @@
from os import path as os_path
from datetime import datetime
from typing import Tuple
from typing import Tuple, List, Optional
from PIL import Image, ImageDraw, ImageFont
try:
from utils import errorLogging
@@ -18,8 +18,14 @@ class OverlayImage:
"Chinese Traditional": "NotoSansTC-Regular.ttf",
}
def __init__(self, root_path: str=None):
self.message_log = []
def __init__(self, root_path: Optional[str] = None) -> None:
"""Overlay image helper.
Args:
root_path: optional project root to resolve bundled fonts. If omitted,
defaults to repository `fonts` directory.
"""
self.message_log: List[dict] = []
if root_path is None:
self.root_path = os_path.join(os_path.dirname(__file__), "..", "..", "..", "fonts")
else:
@@ -58,7 +64,7 @@ class OverlayImage:
}
return colors
def createTextboxSmallLog(self, text:str, language:str, text_color:tuple, base_width:int, base_height:int, font_size:int) -> Image:
def createTextboxSmallLog(self, text: str, language: str, text_color: Tuple[int, int, int], base_width: int, base_height: int, font_size: int) -> Image:
font_family = self.LANGUAGES.get(language, self.LANGUAGES["Default"])
img = Image.new("RGBA", (base_width, base_height), (0, 0, 0, 0))
draw = ImageDraw.Draw(img)
@@ -67,9 +73,16 @@ class OverlayImage:
font_path = os_path.join(self.root_path, font_family)
font = ImageFont.truetype(font_path, font_size)
except Exception:
errorLogging()
font_path = os_path.join(os_path.dirname(__file__), "..", "..", "..", "fonts", font_family)
font = ImageFont.truetype(font_path, font_size)
# overlayフォルダから操作している場合
if os_path.exists(os_path.join(os_path.dirname(__file__), "..", "..", "..", "fonts", font_family)):
font_path = os_path.join(os_path.dirname(__file__), "..", "..", "..", "fonts", font_family)
font = ImageFont.truetype(font_path, font_size)
elif os_path.exists(os_path.join(os_path.dirname(__file__), "fonts", font_family)):
# src-pythonフォルダから操作している場合
font_path = os_path.join(os_path.dirname(__file__), "fonts", font_family)
font = ImageFont.truetype(font_path, font_size)
else:
raise FileNotFoundError(f"Font file not found: {font_family}")
text_width = draw.textlength(text, font)
character_width = text_width // len(text)
@@ -85,7 +98,7 @@ class OverlayImage:
draw.text((text_x, text_y), text, text_color, anchor="mm", stroke_width=0, font=font, align="center")
return img
def createOverlayImageSmallLog(self, message: str, your_language: str, translation: list = [], target_language: list = []) -> Image:
def createOverlayImageSmallLog(self, message: str, your_language: str, translation: List[str] = [], target_language: List[str] = []) -> Image:
# UI設定を取得
ui_size = self.getUiSizeSmallLog()
width, height, font_size = ui_size["width"], ui_size["height"], ui_size["font_size"]
@@ -155,7 +168,7 @@ class OverlayImage:
"text_color_time": (120, 120, 120)
}
def createTextImageLargeLog(self, message_type:str, size:str, text:str, language:str) -> Image:
def createTextImageLargeLog(self, message_type: str, size: str, text: str, language: str) -> Image:
ui_size = self.getUiSizeLargeLog()
font_size = ui_size["font_size_large"] if size == "large" else ui_size["font_size_small"]
text_color = self.getUiColorLargeLog()[f"text_color_{size}"]
@@ -171,11 +184,17 @@ class OverlayImage:
font_path = os_path.join(self.root_path, font_family)
font = ImageFont.truetype(font_path, font_size)
except Exception:
errorLogging()
font_path = os_path.join(os_path.dirname(__file__), "..", "..", "..", "fonts", font_family)
font = ImageFont.truetype(font_path, font_size)
if os_path.exists(os_path.join(os_path.dirname(__file__), "..", "..", "..", "fonts", font_family)):
font_path = os_path.join(os_path.dirname(__file__), "..", "..", "..", "fonts", font_family)
font = ImageFont.truetype(font_path, font_size)
elif os_path.exists(os_path.join(os_path.dirname(__file__), "fonts", font_family)):
font_path = os_path.join(os_path.dirname(__file__), "fonts", font_family)
font = ImageFont.truetype(font_path, font_size)
else:
raise FileNotFoundError(f"Font file not found: {font_family}")
text_width = draw.textlength(text, font)
# 改行を含んだtextの最大の文字数を計算する
text_width = max(draw.textlength(line, font) for line in text.split("\n"))
character_width = text_width // len(text)
character_line_num = int((ui_size["width"] // character_width) - 1)
if len(text) > character_line_num:
@@ -187,7 +206,7 @@ class OverlayImage:
draw.multiline_text((text_x, text_y), text, text_color, anchor=anchor, stroke_width=0, font=font, align=align)
return img
def createTextImageMessageType(self, message_type:str, date_time:str) -> Image:
def createTextImageMessageType(self, message_type: str, date_time: str) -> Image:
ui_size = self.getUiSizeLargeLog()
font_size = ui_size["font_size_small"]
ui_padding = ui_size["padding"]
@@ -206,9 +225,16 @@ class OverlayImage:
font_path = os_path.join(self.root_path, self.LANGUAGES["Default"])
font = ImageFont.truetype(font_path, font_size)
except Exception:
errorLogging()
font_path = os_path.join(os_path.dirname(__file__), "..", "..", "..", "fonts", self.LANGUAGES["Default"])
font = ImageFont.truetype(font_path, font_size)
# overlayフォルダから操作している場合
if os_path.exists(os_path.join(os_path.dirname(__file__), "..", "..", "..", "fonts", self.LANGUAGES["Default"])):
font_path = os_path.join(os_path.dirname(__file__), "..", "..", "..", "fonts", self.LANGUAGES["Default"])
font = ImageFont.truetype(font_path, font_size)
elif os_path.exists(os_path.join(os_path.dirname(__file__), "fonts", self.LANGUAGES["Default"])):
# src-pythonフォルダから操作している場合
font_path = os_path.join(os_path.dirname(__file__), "fonts", self.LANGUAGES["Default"])
font = ImageFont.truetype(font_path, font_size)
else:
raise FileNotFoundError(f"Font file not found: {self.LANGUAGES['Default']}")
text_height = font_size + ui_padding
text_width = draw.textlength(date_time, font)
@@ -222,7 +248,7 @@ class OverlayImage:
draw.text((text_x, text_y), text, text_color, anchor=anchor, stroke_width=0, font=font)
return img
def createTextboxLargeLog(self, message_type: str, message: str = None, your_language: str = None, translation: list = [], target_language: list = [], date_time: str = None) -> Image:
def createTextboxLargeLog(self, message_type: str, message: Optional[str] = None, your_language: Optional[str] = None, translation: List[str] = [], target_language: List[str] = [], date_time: Optional[str] = None) -> Image:
# テキスト画像のリストを作成
images = [self.createTextImageMessageType(message_type, date_time)]
@@ -252,7 +278,7 @@ class OverlayImage:
return combined_img
def createOverlayImageLargeLog(self, message_type:str, message:str=None, your_language:str=None, translation:list=[], target_language:list=[]) -> Image:
def createOverlayImageLargeLog(self, message_type: str, message: Optional[str] = None, your_language: Optional[str] = None, translation: List[str] = [], target_language: List[str] = []) -> Image:
ui_color = self.getUiColorLargeLog()
background_color = ui_color["background_color"]
background_outline_color = ui_color["background_outline_color"]

67
src-python/models/overlay/overlay_utils.py
View File

@@ -1,11 +1,29 @@
import numpy as np
from typing import Sequence
def toHomogeneous(matrix):
def toHomogeneous(matrix: np.ndarray) -> np.ndarray:
"""Convert a 3x4 base matrix to a 4x4 homogeneous matrix.
Args:
matrix: 3x4 numpy array
Returns:
4x4 numpy array with last row [0, 0, 0, 1]
"""
homogeneous_matrix = np.vstack([matrix, [0, 0, 0, 1]])
return homogeneous_matrix
# 移動行列を生成する関数
def calcTranslationMatrix(translation):
def calcTranslationMatrix(translation: Sequence[float]) -> np.ndarray:
"""Create a 4x4 translation matrix from a 3-element translation.
Args:
translation: (tx, ty, tz)
Returns:
4x4 numpy translation matrix
"""
tx, ty, tz = translation
return np.array([
[1, 0, 0, tx],
@@ -15,9 +33,10 @@ def calcTranslationMatrix(translation):
])
# X軸周りの回転行列を生成する関数
def calcRotationMatrixX(angle):
c = np.cos(np.pi/180*angle)
s = np.sin(np.pi/180*angle)
def calcRotationMatrixX(angle: float) -> np.ndarray:
"""Rotation matrix around X axis for given angle in degrees."""
c = np.cos(np.pi / 180 * angle)
s = np.sin(np.pi / 180 * angle)
return np.array([
[1, 0, 0, 0],
[0, c, -s, 0],
@@ -26,9 +45,10 @@ def calcRotationMatrixX(angle):
])
# Y軸周りの回転行列を生成する関数
def calcRotationMatrixY(angle):
c = np.cos(np.pi/180*angle)
s = np.sin(np.pi/180*angle)
def calcRotationMatrixY(angle: float) -> np.ndarray:
"""Rotation matrix around Y axis for given angle in degrees."""
c = np.cos(np.pi / 180 * angle)
s = np.sin(np.pi / 180 * angle)
return np.array([
[c, 0, s, 0],
[0, 1, 0, 0],
@@ -37,9 +57,10 @@ def calcRotationMatrixY(angle):
])
# Z軸周りの回転行列を生成する関数
def calcRotationMatrixZ(angle):
c = np.cos(np.pi/180*angle)
s = np.sin(np.pi/180*angle)
def calcRotationMatrixZ(angle: float) -> np.ndarray:
"""Rotation matrix around Z axis for given angle in degrees."""
c = np.cos(np.pi / 180 * angle)
s = np.sin(np.pi / 180 * angle)
return np.array([
[c, -s, 0, 0],
[s, c, 0, 0],
@@ -48,7 +69,17 @@ def calcRotationMatrixZ(angle):
])
# 3x4行列の座標を基準として回転や移動を行う関数
def transform_matrix(base_matrix, translation, rotation):
def transform_matrix(base_matrix: np.ndarray, translation: Sequence[float], rotation: Sequence[float]) -> np.ndarray:
"""Apply translation and Euler rotations to a 3x4 base matrix.
Args:
base_matrix: 3x4 base transform matrix
translation: (tx, ty, tz)
rotation: (x_deg, y_deg, z_deg)
Returns:
Transformed 3x4 matrix (numpy.ndarray)
"""
homogeneous_base_matrix = toHomogeneous(base_matrix)
translation_matrix = calcTranslationMatrix(translation)
rotation_matrix_x = calcRotationMatrixX(rotation[0])
@@ -60,10 +91,18 @@ def transform_matrix(base_matrix, translation, rotation):
result_matrix = np.dot(homogeneous_base_matrix, transformation_matrix)
return result_matrix[:3, :]
def euler_to_rotation_matrix(angles):
def euler_to_rotation_matrix(angles: Sequence[float]) -> np.ndarray:
"""Convert Euler angles in degrees to a 3x3 rotation matrix.
Args:
angles: (x_deg, y_deg, z_deg)
Returns:
3x3 rotation matrix
"""
phi = angles[0] * np.pi / 180
theta = angles[1] * np.pi / 180
psi = angles[2]* np.pi / 180
psi = angles[2] * np.pi / 180
R_x = np.array([[1, 0, 0],
[0, np.cos(phi), -np.sin(phi)],
[0, np.sin(phi), np.cos(phi)]])

5
src-python/models/transcription/transcription_languages.py
View File

@@ -1,3 +1,8 @@
"""Language table used by transcription components.
Maps a display language and country to engine-specific language codes.
"""
transcription_lang = {
"Afrikaans":{
"South Africa":{

202
src-python/models/transcription/transcription_recorder.py
View File

@@ -1,10 +1,18 @@
"""Recorders that wrap speech_recognition microphone interfaces.
These classes provide small adapters that push raw audio bytes into queues.
They intentionally keep a thin API so the rest of the system can mock them
in tests.
"""
from typing import Any
from speech_recognition import Recognizer, Microphone
from pyaudiowpatch import get_sample_size, paInt16
from datetime import datetime
from queue import Queue
class BaseRecorder:
def __init__(self, source, energy_threshold, dynamic_energy_threshold, record_timeout):
def __init__(self, source: Any, energy_threshold: int, dynamic_energy_threshold: bool, record_timeout: int) -> None:
self.recorder = Recognizer()
self.recorder.energy_threshold = energy_threshold
self.recorder.dynamic_energy_threshold = dynamic_energy_threshold
@@ -16,39 +24,65 @@ class BaseRecorder:
self.source = source
def adjustForNoise(self):
def adjustForNoise(self) -> None:
with self.source:
self.recorder.adjust_for_ambient_noise(self.source)
def recordIntoQueue(self, audio_queue):
def recordIntoQueue(self, audio_queue: Any) -> None:
def record_callback(_, audio):
audio_queue.put((audio.get_raw_data(), datetime.now()))
self.stop, self.pause, self.resume = self.recorder.listen_in_background(self.source, record_callback, phrase_time_limit=self.record_timeout)
class SelectedMicRecorder(BaseRecorder):
def __init__(self, device, energy_threshold, dynamic_energy_threshold, record_timeout):
source=Microphone(
device_index=device['index'],
sample_rate=int(device["defaultSampleRate"]),
)
def __init__(self, device: dict, energy_threshold: int, dynamic_energy_threshold: bool, record_timeout: int) -> None:
# Safely construct Microphone source. If device dict is missing expected keys
# or index is out-of-range for the platform, fallback to default device (None)
try:
device_index = int(device.get('index', -1))
sample_rate = int(device.get("defaultSampleRate", 16000))
if device_index < 0:
# invalid index -> fallback
raise ValueError("invalid device index")
source = Microphone(
device_index=device_index,
sample_rate=sample_rate,
)
except Exception:
# Best-effort fallback: use system default microphone
try:
source = Microphone()
except Exception:
raise
super().__init__(source=source, energy_threshold=energy_threshold, dynamic_energy_threshold=dynamic_energy_threshold, record_timeout=record_timeout)
# self.adjustForNoise()
class SelectedSpeakerRecorder(BaseRecorder):
def __init__(self, device, energy_threshold, dynamic_energy_threshold, record_timeout):
source = Microphone(speaker=True,
device_index= device["index"],
sample_rate=int(device["defaultSampleRate"]),
chunk_size=get_sample_size(paInt16),
channels=device["maxInputChannels"]
)
class SelectedSpeakerRecorder(BaseRecorder):
def __init__(self, device: dict, energy_threshold: int, dynamic_energy_threshold: bool, record_timeout: int) -> None:
try:
device_index = int(device.get('index', -1))
sample_rate = int(device.get("defaultSampleRate", 16000))
channels = int(device.get("maxInputChannels", 1))
if device_index < 0:
raise ValueError("invalid device index")
source = Microphone(speaker=True,
device_index=device_index,
sample_rate=sample_rate,
chunk_size=get_sample_size(paInt16),
channels=channels
)
except Exception:
try:
source = Microphone(speaker=True)
except Exception:
raise
super().__init__(source=source, energy_threshold=energy_threshold, dynamic_energy_threshold=dynamic_energy_threshold, record_timeout=record_timeout)
# self.adjustForNoise()
class BaseEnergyRecorder:
def __init__(self, source):
def __init__(self, source: Any) -> None:
self.recorder = Recognizer()
self.recorder.energy_threshold = 0
self.recorder.dynamic_energy_threshold = False
@@ -60,38 +94,68 @@ class BaseEnergyRecorder:
self.source = source
def adjustForNoise(self):
def adjustForNoise(self) -> None:
with self.source:
self.recorder.adjust_for_ambient_noise(self.source)
def recordIntoQueue(self, energy_queue):
def recordIntoQueue(self, energy_queue: Any) -> None:
def recordCallback(_, energy):
energy_queue.put(energy)
self.stop, self.pause, self.resume = self.recorder.listen_energy_in_background(self.source, recordCallback)
class SelectedMicEnergyRecorder(BaseEnergyRecorder):
def __init__(self, device):
source=Microphone(
device_index=device['index'],
sample_rate=int(device["defaultSampleRate"]),
)
def __init__(self, device: dict) -> None:
try:
device_index = int(device.get('index', -1))
sample_rate = int(device.get("defaultSampleRate", 16000))
if device_index < 0:
raise ValueError("invalid device index")
source = Microphone(
device_index=device_index,
sample_rate=sample_rate,
)
except Exception:
try:
source = Microphone()
except Exception:
raise
super().__init__(source=source)
# self.adjustForNoise()
class SelectedSpeakerEnergyRecorder(BaseEnergyRecorder):
def __init__(self, device):
source = Microphone(speaker=True,
device_index= device["index"],
sample_rate=int(device["defaultSampleRate"]),
channels=device["maxInputChannels"]
)
class SelectedSpeakerEnergyRecorder(BaseEnergyRecorder):
def __init__(self, device: dict) -> None:
try:
device_index = int(device.get('index', -1))
sample_rate = int(device.get("defaultSampleRate", 16000))
channels = int(device.get("maxInputChannels", 1))
if device_index < 0:
raise ValueError("invalid device index")
source = Microphone(speaker=True,
device_index=device_index,
sample_rate=sample_rate,
channels=channels
)
except Exception:
try:
source = Microphone(speaker=True)
except Exception:
raise
super().__init__(source=source)
# self.adjustForNoise()
class BaseEnergyAndAudioRecorder:
def __init__(self, source, energy_threshold, dynamic_energy_threshold, phrase_time_limit, phrase_timeout, record_timeout):
def __init__(
self,
source: Any,
energy_threshold: int,
dynamic_energy_threshold: bool,
phrase_time_limit: int,
phrase_timeout: int,
record_timeout: int,
) -> None:
self.recorder = Recognizer()
self.recorder.energy_threshold = energy_threshold
self.recorder.dynamic_energy_threshold = dynamic_energy_threshold
@@ -105,11 +169,11 @@ class BaseEnergyAndAudioRecorder:
self.source = source
def adjustForNoise(self):
def adjustForNoise(self) -> None:
with self.source:
self.recorder.adjust_for_ambient_noise(self.source)
def recordIntoQueue(self, audio_queue, energy_queue=None):
def recordIntoQueue(self, audio_queue: Any, energy_queue: Any = None) -> None:
def audioRecordCallback(_, audio):
audio_queue.put((audio.get_raw_data(), datetime.now()))
@@ -122,14 +186,34 @@ class BaseEnergyAndAudioRecorder:
phrase_time_limit=self.phrase_time_limit,
callback_energy=energyRecordCallback if energy_queue is not None else None,
phrase_timeout=self.phrase_timeout,
record_timeout=self.record_timeout)
record_timeout=self.record_timeout,
)
class SelectedMicEnergyAndAudioRecorder(BaseEnergyAndAudioRecorder):
def __init__(self, device, energy_threshold, dynamic_energy_threshold, phrase_time_limit, phrase_timeout:int=1, record_timeout:int=5):
source=Microphone(
device_index=device['index'],
sample_rate=int(device["defaultSampleRate"]),
)
def __init__(
self,
device: dict,
energy_threshold: int,
dynamic_energy_threshold: bool,
phrase_time_limit: int,
phrase_timeout: int = 1,
record_timeout: int = 5,
) -> None:
try:
device_index = int(device.get('index', -1))
sample_rate = int(device.get("defaultSampleRate", 16000))
if device_index < 0:
raise ValueError("invalid device index")
source = Microphone(
device_index=device_index,
sample_rate=sample_rate,
)
except Exception:
try:
source = Microphone()
except Exception:
raise
super().__init__(
source=source,
energy_threshold=energy_threshold,
@@ -140,15 +224,35 @@ class SelectedMicEnergyAndAudioRecorder(BaseEnergyAndAudioRecorder):
)
# self.adjustForNoise()
class SelectedSpeakerEnergyAndAudioRecorder(BaseEnergyAndAudioRecorder):
def __init__(self, device, energy_threshold, dynamic_energy_threshold, phrase_time_limit, phrase_timeout:int=1, record_timeout:int=5):
source = Microphone(speaker=True,
device_index= device["index"],
sample_rate=int(device["defaultSampleRate"]),
chunk_size=get_sample_size(paInt16),
channels=device["maxInputChannels"]
)
class SelectedSpeakerEnergyAndAudioRecorder(BaseEnergyAndAudioRecorder):
def __init__(
self,
device: dict,
energy_threshold: int,
dynamic_energy_threshold: bool,
phrase_time_limit: int,
phrase_timeout: int = 1,
record_timeout: int = 5,
) -> None:
try:
device_index = int(device.get('index', -1))
sample_rate = int(device.get("defaultSampleRate", 16000))
channels = int(device.get("maxInputChannels", 1))
if device_index < 0:
raise ValueError("invalid device index")
source = Microphone(speaker=True,
device_index=device_index,
sample_rate=sample_rate,
chunk_size=get_sample_size(paInt16),
channels=channels,
)
except Exception:
try:
source = Microphone(speaker=True)
except Exception:
raise
super().__init__(
source=source,
energy_threshold=energy_threshold,

98
src-python/models/transcription/transcription_transcriber.py
View File

@@ -1,7 +1,14 @@
"""Runtime transcriber that wraps Google SpeechRecognition and faster-whisper.
This class focuses on converting incoming raw audio buffers into text using
either the Google web recognizer (online) or a local Whisper model (offline).
"""
import time
from io import BytesIO
from threading import Event
import wave
from typing import Any, Callable, Dict, List, Optional, Tuple
from speech_recognition import Recognizer, AudioData, AudioFile
from speech_recognition.exceptions import UnknownValueError
from datetime import timedelta
@@ -20,38 +27,71 @@ warnings.simplefilter('ignore', RuntimeWarning)
PHRASE_TIMEOUT = 3
MAX_PHRASES = 10
class AudioTranscriber:
def __init__(self, speaker, source, phrase_timeout, max_phrases, transcription_engine, root=None, whisper_weight_type=None, device="cpu", device_index=0):
"""Convert queued audio buffers into transcripts.
Public attributes set by the constructor:
- speaker: bool
- phrase_timeout: int
- max_phrases: int
Methods are intentionally permissive about input types to match the
existing codebase; this wrapper adds typing for clarity.
"""
def __init__(
self,
speaker: bool,
source: Any,
phrase_timeout: int,
max_phrases: int,
transcription_engine: str,
root: Optional[str] = None,
whisper_weight_type: Optional[str] = None,
device: str = "cpu",
device_index: int = 0,
compute_type: str = "auto",
) -> None:
self.speaker = speaker
self.phrase_timeout = phrase_timeout
self.max_phrases = max_phrases
self.transcript_data = []
self.transcript_data: List[Dict[str, Any]] = []
self.transcript_changed_event = Event()
self.audio_recognizer = Recognizer()
self.transcription_engine = "Google"
self.whisper_model = None
self.audio_sources = {
"sample_rate": source.SAMPLE_RATE,
"sample_width": source.SAMPLE_WIDTH,
"channels": source.channels,
"last_sample": bytes(),
"last_spoken": None,
"new_phrase": True,
"process_data_func": self.processSpeakerData if speaker else self.processSpeakerData
self.audio_sources: Dict[str, Any] = {
"sample_rate": source.SAMPLE_RATE,
"sample_width": source.SAMPLE_WIDTH,
"channels": source.channels,
"last_sample": bytes(),
"last_spoken": None,
"new_phrase": True,
"process_data_func": self.processSpeakerData if speaker else self.processSpeakerData,
}
if transcription_engine == "Whisper" and checkWhisperWeight(root, whisper_weight_type) is True:
self.whisper_model = getWhisperModel(root, whisper_weight_type, device=device, device_index=device_index)
self.whisper_model = getWhisperModel(
root, whisper_weight_type, device=device, device_index=device_index, compute_type=compute_type
)
self.transcription_engine = "Whisper"
def transcribeAudioQueue(self, audio_queue, languages, countries, avg_logprob=-0.8, no_speech_prob=0.6):
def transcribeAudioQueue(
self,
audio_queue: Any,
languages: List[str],
countries: List[str],
avg_logprob: float = -0.8,
no_speech_prob: float = 0.6,
) -> bool:
if audio_queue.empty():
time.sleep(0.01)
return False
audio, time_spoken = audio_queue.get()
self.updateLastSampleAndPhraseStatus(audio, time_spoken)
confidences = [{"confidence": 0, "text": "", "language": None}]
confidences: List[Dict[str, Any]] = [{"confidence": 0, "text": "", "language": None}]
try:
audio_data = self.audio_sources["process_data_func"]()
match self.transcription_engine:
@@ -67,13 +107,19 @@ class AudioTranscriber:
except Exception:
pass
case "Whisper":
audio_data = np.frombuffer(audio_data.get_raw_data(convert_rate=16000, convert_width=2), np.int16).flatten().astype(np.float32) / 32768.0
audio_data = np.frombuffer(
audio_data.get_raw_data(convert_rate=16000, convert_width=2), np.int16
).flatten().astype(np.float32) / 32768.0
if isinstance(audio_data, torch.Tensor):
audio_data = audio_data.detach().numpy()
for language, country in zip(languages, countries):
text = ""
source_language = transcription_lang[language][country][self.transcription_engine] if len(languages) == 1 else None
source_language = (
transcription_lang[language][country][self.transcription_engine]
if len(languages) == 1
else None
)
segments, info = self.whisper_model.transcribe(
audio_data,
beam_size=5,
@@ -85,13 +131,15 @@ class AudioTranscriber:
without_timestamps=True,
task="transcribe",
vad_filter=False,
)
)
for s in segments:
if s.avg_logprob < avg_logprob or s.no_speech_prob > no_speech_prob:
continue
text += s.text
confidences.append({"confidence": info.language_probability, "text": text, "language": language})
if (len(languages) == 1) or (transcription_lang[language][country][self.transcription_engine] == info.language):
if (len(languages) == 1) or (
transcription_lang[language][country][self.transcription_engine] == info.language
):
break
except UnknownValueError:
@@ -106,7 +154,7 @@ class AudioTranscriber:
self.updateTranscript(result)
return True
def updateLastSampleAndPhraseStatus(self, data, time_spoken):
def updateLastSampleAndPhraseStatus(self, data: bytes, time_spoken) -> None:
source_info = self.audio_sources
if source_info["last_spoken"] and time_spoken - source_info["last_spoken"] > timedelta(seconds=self.phrase_timeout):
source_info["last_sample"] = bytes()
@@ -117,11 +165,13 @@ class AudioTranscriber:
source_info["last_sample"] += data
source_info["last_spoken"] = time_spoken
def processMicData(self):
audio_data = AudioData(self.audio_sources["last_sample"], self.audio_sources["sample_rate"], self.audio_sources["sample_width"])
def processMicData(self) -> AudioData:
audio_data = AudioData(
self.audio_sources["last_sample"], self.audio_sources["sample_rate"], self.audio_sources["sample_width"]
)
return audio_data
def processSpeakerData(self):
def processSpeakerData(self) -> AudioData:
temp_file = BytesIO()
with wave.open(temp_file, 'wb') as wf:
wf.setnchannels(self.audio_sources["channels"])
@@ -141,7 +191,7 @@ class AudioTranscriber:
audio = self.audio_recognizer.record(source)
return audio
def updateTranscript(self, result):
def updateTranscript(self, result: dict) -> None:
source_info = self.audio_sources
transcript = self.transcript_data
@@ -152,14 +202,14 @@ class AudioTranscriber:
else:
transcript[0] = result
def getTranscript(self):
def getTranscript(self) -> dict:
if len(self.transcript_data) > 0:
result = self.transcript_data.pop(-1)
else:
result = {"confidence": 0, "text": "", "language": None}
return result
def clearTranscriptData(self):
def clearTranscriptData(self) -> None:
self.transcript_data.clear()
self.audio_sources["last_sample"] = bytes()
self.audio_sources["new_phrase"] = True

81
src-python/models/transcription/transcription_whisper.py
View File

@@ -1,6 +1,17 @@
"""Helpers for downloading and loading Whisper (faster-whisper) models.
This module exposes small utilities used by the transcription subsystem:
- downloadFile: stream-download a file with optional progress callback
- checkWhisperWeight: quick local availability check
- downloadWhisperWeight: download model artifacts from HF hub
- getWhisperModel: construct and return a WhisperModel instance
The functions are defensive: failures are caught and reported by the caller.
"""
from os import path as os_path, makedirs as os_makedirs
from requests import get as requests_get
from typing import Callable
from typing import Callable, Optional
import huggingface_hub
from faster_whisper import WhisperModel
import logging
@@ -30,24 +41,36 @@ _FILENAMES = [
"vocabulary.json",
]
def downloadFile(url, path, func=None):
def downloadFile(url: str, path: str, func: Optional[Callable[[float], None]] = None) -> None:
"""Download a file from `url` to `path`.
Args:
url: remote URL to download from
path: local filepath to write
func: optional callback(progress: float) called with a 0.0-1.0 progress
"""
try:
res = requests_get(url, stream=True)
res.raise_for_status()
file_size = int(res.headers.get('content-length', 0))
total_chunk = 0
with open(os_path.join(path), 'wb') as file:
for chunk in res.iter_content(chunk_size=1024*2000):
for chunk in res.iter_content(chunk_size=1024 * 2000):
file.write(chunk)
if isinstance(func, Callable):
if callable(func) and file_size:
total_chunk += len(chunk)
func(total_chunk/file_size)
func(total_chunk / file_size)
except Exception:
# Silent failure here; caller may re-check or log
pass
def checkWhisperWeight(root, weight_type):
def checkWhisperWeight(root: str, weight_type: str) -> bool:
"""Return True if a Whisper model for `weight_type` is loadable from disk.
This attempts to construct a local `WhisperModel` with local_files_only=True
to verify required files exist and are compatible.
"""
path = os_path.join(root, "weights", "whisper", weight_type)
result = False
try:
WhisperModel(
path,
@@ -58,25 +81,50 @@ def checkWhisperWeight(root, weight_type):
num_workers=1,
local_files_only=True,
)
result = True
return True
except Exception:
pass
return result
return False
def downloadWhisperWeight(root, weight_type, callback=None, end_callback=None):
def downloadWhisperWeight(
root: str,
weight_type: str,
callback: Optional[Callable[[float], None]] = None,
end_callback: Optional[Callable[[], None]] = None,
) -> None:
"""Ensure Whisper weight files are present locally; download them if missing.
Args:
root: project root where `weights/whisper` lives
weight_type: key from `_MODELS` (eg. "tiny", "base")
callback: progress callback for the main model file
end_callback: called when download completes
"""
path = os_path.join(root, "weights", "whisper", weight_type)
os_makedirs(path, exist_ok=True)
if checkWhisperWeight(root, weight_type) is False:
if not checkWhisperWeight(root, weight_type):
for filename in _FILENAMES:
file_path = os_path.join(path, filename)
url = huggingface_hub.hf_hub_url(_MODELS[weight_type], filename)
downloadFile(url, file_path, func=callback if filename == "model.bin" else None)
if isinstance(end_callback, Callable):
if callable(end_callback):
end_callback()
def getWhisperModel(root, weight_type, device="cpu", device_index=0):
def getWhisperModel(
root: str,
weight_type: str,
device: str = "cpu",
device_index: int = 0,
compute_type: str = "auto",
) -> WhisperModel:
"""Return a `WhisperModel` instance loaded from local weights.
Raises:
ValueError: when VRAM shortage is detected (wrapped from RuntimeError)
Exception: other loading errors are propagated.
"""
path = os_path.join(root, "weights", "whisper", weight_type)
compute_type = getBestComputeType(device, device_index)
if compute_type == "auto":
compute_type = getBestComputeType(device, device_index)
try:
model = WhisperModel(
path,
@@ -89,11 +137,10 @@ def getWhisperModel(root, weight_type, device="cpu", device_index=0):
)
return model
except RuntimeError as e:
# VRAM不足エラーの検出
# Detect VRAM out-of-memory-like errors and raise a clear ValueError
error_message = str(e)
if "CUDA out of memory" in error_message or "CUBLAS_STATUS_ALLOC_FAILED" in error_message:
raise ValueError("VRAM_OUT_OF_MEMORY", error_message)
# その他のエラーは通常通り再送出
raise
if __name__ == "__main__":

74
src-python/models/translation/translation_languages.py
View File

@@ -1,4 +1,13 @@
translation_lang = {}
"""Language code mappings for supported translation backends.
Provides `translation_lang` mapping keyed by backend name with `source` and
`target` maps used by `Translator.getLanguageCode`.
"""
from typing import Dict
translation_lang: Dict[str, Dict[str, Dict[str, str]]] = {}
dict_deepl_languages = {
"Arabic":"ar",
"Bulgarian":"bg",
@@ -37,10 +46,7 @@ dict_deepl_languages = {
"Chinese Simplified":"zh",
"Chinese Traditional":"zh"
}
translation_lang["DeepL"] = {
"source":dict_deepl_languages,
"target":dict_deepl_languages,
}
translation_lang["DeepL"] = {"source": dict_deepl_languages, "target": dict_deepl_languages}
dict_deepl_api_source_languages = {
"Japanese":"ja",
@@ -109,10 +115,7 @@ dict_deepl_api_target_languages = {
"Chinese Simplified":"zh",
"Chinese Traditional":"zh"
}
translation_lang["DeepL_API"] = {
"source": dict_deepl_api_source_languages,
"target": dict_deepl_api_target_languages,
}
translation_lang["DeepL_API"] = {"source": dict_deepl_api_source_languages, "target": dict_deepl_api_target_languages}
dict_google_languages = {
"Japanese":"ja",
@@ -176,13 +179,10 @@ dict_google_languages = {
"Belarusian":"be",
"Cebuano":"ceb",
"Esperanto":"eo",
"Basque":"eu",
# "Basque":"eu",
"Irish":"ga"
}
translation_lang["Google"] = {
"source":dict_google_languages,
"target":dict_google_languages,
}
translation_lang["Google"] = {"source": dict_google_languages, "target": dict_google_languages}
dict_bing_languages = {
"Japanese":"ja",
@@ -247,10 +247,7 @@ dict_bing_languages = {
"Punjabi":"pa",
"Irish":"ga"
}
translation_lang["Bing"] = {
"source":dict_bing_languages,
"target":dict_bing_languages,
}
translation_lang["Bing"] = {"source": dict_bing_languages, "target": dict_bing_languages}
dict_papago_languages = {
"German": "de",
@@ -270,10 +267,7 @@ dict_papago_languages = {
"Chinese Traditional":"zh-TW",
}
translation_lang["Papago"] = {
"source":dict_papago_languages,
"target":dict_papago_languages,
}
translation_lang["Papago"] = {"source": dict_papago_languages, "target": dict_papago_languages}
dict_m2m100_languages = {
"English": "en",
@@ -317,7 +311,7 @@ dict_m2m100_languages = {
"Malayalam": "ml",
"Welsh": "cy",
"Slovak": "sk",
"Telugu": "te",
# "Telugu": "te",
"Persian": "fa",
"Latvian": "lv",
"Bengali": "bn",
@@ -328,7 +322,7 @@ dict_m2m100_languages = {
"Estonian": "et",
"Macedonian": "mk",
"Breton": "br",
"Basque": "eu",
# "Basque": "eu",
"Icelandic": "is",
"Armenian": "hy",
"Nepali": "ne",
@@ -378,15 +372,8 @@ dict_m2m100_languages = {
"Sundanese": "su"
}
translation_lang["m2m100_418M-ct2-int8"] = {
"source":dict_m2m100_languages,
"target":dict_m2m100_languages,
}
translation_lang["m2m100_1.2B-ct2-int8"] = {
"source":dict_m2m100_languages,
"target":dict_m2m100_languages,
}
translation_lang["m2m100_418M-ct2-int8"] = {"source":dict_m2m100_languages, "target":dict_m2m100_languages}
translation_lang["m2m100_1.2B-ct2-int8"] = {"source":dict_m2m100_languages, "target":dict_m2m100_languages}
dict_nllb_languages = {
"Acehnese (Arabic script)": "ace_Arab",
@@ -595,15 +582,8 @@ dict_nllb_languages = {
"Zulu": "zul_Latn"
}
translation_lang["nllb-200-distilled-1.3B-ct2-int8"] = {
"source":dict_nllb_languages,
"target":dict_nllb_languages,
}
translation_lang["nllb-200-3.3B-ct2-int8"] = {
"source":dict_nllb_languages,
"target":dict_nllb_languages,
}
translation_lang["nllb-200-distilled-1.3B-ct2-int8"] = {"source":dict_nllb_languages, "target":dict_nllb_languages}
translation_lang["nllb-200-3.3B-ct2-int8"] = {"source":dict_nllb_languages, "target":dict_nllb_languages}
dict_plamo_languages = {
"English": "English",
@@ -638,10 +618,7 @@ dict_plamo_languages = {
"Traditional Chinese":"Traditional Chinese"
}
translation_lang["Plamo_API"] = {
"source":dict_plamo_languages,
"target":dict_plamo_languages,
}
translation_lang["Plamo_API"] = {"source":dict_plamo_languages, "target":dict_plamo_languages}
dict_gemini_languages = {
"Arabic": "Arabic",
@@ -685,7 +662,4 @@ dict_gemini_languages = {
"Vietnamese": "Vietnamese",
}
translation_lang["Gemini_API"] = {
"source":dict_gemini_languages,
"target":dict_gemini_languages,
}
translation_lang["Gemini_API"] = {"source":dict_gemini_languages, "target":dict_gemini_languages}

115
src-python/models/translation/translation_translator.py
View File

@@ -4,6 +4,7 @@ try:
from translators import translate_text as other_web_Translator
ENABLE_TRANSLATORS = True
except Exception:
other_web_Translator = None # type: ignore
ENABLE_TRANSLATORS = False
try:
@@ -25,24 +26,40 @@ import transformers
from utils import errorLogging, getBestComputeType
import warnings
from typing import Any, Optional, Tuple
warnings.filterwarnings("ignore")
# Translator
class Translator():
def __init__(self):
self.deepl_client = None
self.plamo_client = None
self.gemini_client = None
self.ctranslate2_translator = None
self.ctranslate2_tokenizer = None
self.is_loaded_ctranslate2_model = False
self.is_enable_translators = ENABLE_TRANSLATORS
class Translator:
"""High-level translator facade.
This class wraps multiple backends (DeepL, DeepL API, Google, Bing, Papago,
and CTranslate2 local models). Optional dependencies may be unavailable at
runtime; methods degrade gracefully and return False or an empty string on
failure (kept compatible with existing behavior).
"""
def __init__(self) -> None:
self.deepl_client: Optional[DeepLClient] = None
self.plamo_client: Optional[PlamoClient] = None
self.gemini_client: Optional[GeminiClient] = None
self.ctranslate2_translator: Any = None
self.ctranslate2_tokenizer: Any = None
self.is_loaded_ctranslate2_model: bool = False
self.is_changed_translator_parameters: bool = False
self.is_enable_translators: bool = ENABLE_TRANSLATORS
def authenticationDeepLAuthKey(self, auth_key: str) -> bool:
"""Authenticate DeepL API with the provided key.
Returns True on success, False on failure.
"""
result = True
try:
self.deepl_client = DeepLClient(auth_key)
self.deepl_client.translate_text("Hello World", target_lang="EN-US")
# quick smoke test
self.deepl_client.translate_text(" ", target_lang="EN-US")
except Exception:
errorLogging()
self.deepl_client = None
@@ -50,6 +67,10 @@ class Translator():
return result
def authenticationPlamoAuthKey(self, auth_key: str, model: str) -> bool:
"""Authenticate Plamo API with the provided key.
Returns True on success, False on failure.
"""
result = True
try:
self.plamo_client = PlamoClient(auth_key, model=model)
@@ -61,6 +82,10 @@ class Translator():
return result
def authenticationGeminiAuthKey(self, auth_key: str, model: str) -> bool:
"""Authenticate Gemini API with the provided key.
Returns True on success, False on failure.
"""
result = True
try:
self.gemini_client = GeminiClient(auth_key, model=model)
@@ -71,21 +96,27 @@ class Translator():
result = False
return result
def changeCTranslate2Model(self, path, model_type, device="cpu", device_index=0):
def changeCTranslate2Model(self, path: str, model_type: str, device: str = "cpu", device_index: int = 0, compute_type: str = "auto") -> None:
"""Load a CTranslate2 model from weights.
This sets internal translator/tokenizer objects and flips
``is_loaded_ctranslate2_model`` on success.
"""
self.is_loaded_ctranslate2_model = False
directory_name = ctranslate2_weights[model_type]["directory_name"]
tokenizer = ctranslate2_weights[model_type]["tokenizer"]
weight_path = os_path.join(path, "weights", "ctranslate2", directory_name)
tokenizer_path = os_path.join(path, "weights", "ctranslate2", directory_name, "tokenizer")
compute_type = getBestComputeType(device, device_index)
if compute_type == "auto":
compute_type = getBestComputeType(device, device_index)
self.ctranslate2_translator = ctranslate2.Translator(
weight_path,
device=device,
device_index=device_index,
compute_type=compute_type,
inter_threads=1,
intra_threads=4
intra_threads=4,
)
try:
self.ctranslate2_tokenizer = transformers.AutoTokenizer.from_pretrained(tokenizer, cache_dir=tokenizer_path)
@@ -95,11 +126,21 @@ class Translator():
self.ctranslate2_tokenizer = transformers.AutoTokenizer.from_pretrained(tokenizer, cache_dir=tokenizer_path)
self.is_loaded_ctranslate2_model = True
def isLoadedCTranslate2Model(self):
def isLoadedCTranslate2Model(self) -> bool:
return self.is_loaded_ctranslate2_model
def translateCTranslate2(self, message, source_language, target_language, weight_type):
result = False
def isChangedTranslatorParameters(self) -> bool:
return self.is_changed_translator_parameters
def setChangedTranslatorParameters(self, is_changed: bool) -> None:
self.is_changed_translator_parameters = is_changed
def translateCTranslate2(self, message: str, source_language: str, target_language, weight_type: str) -> Any:
"""Translate using a loaded CTranslate2 model.
Returns a string on success or False on failure (keeps legacy behavior).
"""
result: Any = False
if self.is_loaded_ctranslate2_model is True:
try:
self.ctranslate2_tokenizer.src_lang = source_language
@@ -119,7 +160,11 @@ class Translator():
return result
@staticmethod
def getLanguageCode(translator_name, weight_type, target_country, source_language, target_language):
def getLanguageCode(translator_name: str, weight_type: str, target_country: str, source_language: str, target_language: str) -> Tuple[str, str]:
"""Resolve a friendly language name to translator-specific codes.
Returns (source_code, target_code).
"""
match translator_name:
case "DeepL_API":
if target_language == "English":
@@ -136,23 +181,31 @@ class Translator():
translator_name = weight_type
case _:
pass
source_language=translation_lang[translator_name]["source"][source_language]
target_language=translation_lang[translator_name]["target"][target_language]
source_language = translation_lang[translator_name]["source"][source_language]
target_language = translation_lang[translator_name]["target"][target_language]
return source_language, target_language
def translate(self, translator_name, weight_type, source_language, target_language, target_country, message):
def translate(self, translator_name: str, weight_type: str, source_language: str, target_language: str, target_country: str, message: str) -> Any:
"""Translate `message` using the named translator backend.
Returns translated string on success, or False on failure. When
source_language == target_language the original message is returned.
"""
try:
result = ""
if source_language == target_language:
return message
result: Any = ""
source_language, target_language = self.getLanguageCode(translator_name, weight_type, target_country, source_language, target_language)
match translator_name:
case "DeepL":
if self.is_enable_translators is True:
if self.is_enable_translators is True and other_web_Translator is not None:
result = other_web_Translator(
query_text=message,
translator="deepl",
from_language=source_language,
to_language=target_language,
)
)
case "DeepL_API":
if self.is_enable_translators is True:
if self.deepl_client is None:
@@ -161,7 +214,7 @@ class Translator():
result = self.deepl_client.translate_text(
message,
source_lang=source_language,
target_lang=target_language,
target_lang=target_language
).text
case "Plamo_API":
if self.plamo_client is None:
@@ -182,29 +235,29 @@ class Translator():
output_lang=target_language,
)
case "Google":
if self.is_enable_translators is True:
if self.is_enable_translators is True and other_web_Translator is not None:
result = other_web_Translator(
query_text=message,
translator="google",
from_language=source_language,
to_language=target_language,
)
)
case "Bing":
if self.is_enable_translators is True:
if self.is_enable_translators is True and other_web_Translator is not None:
result = other_web_Translator(
query_text=message,
translator="bing",
from_language=source_language,
to_language=target_language,
)
)
case "Papago":
if self.is_enable_translators is True:
if self.is_enable_translators is True and other_web_Translator is not None:
result = other_web_Translator(
query_text=message,
translator="papago",
from_language=source_language,
to_language=target_language,
)
)
case "CTranslate2":
result = self.translateCTranslate2(
message=message,

15
src-python/models/translation/translation_utils.py
View File

@@ -1,6 +1,8 @@
from os import path as os_path
from os import makedirs as os_makedirs
from requests import get as requests_get
from typing import Callable
import hashlib
import transformers
import ctranslate2
from huggingface_hub import hf_hub_url, list_repo_files
@@ -14,6 +16,15 @@ except Exception:
sys.path.append(os_path.dirname(os_path.dirname(os_path.dirname(os_path.abspath(__file__)))))
from utils import errorLogging, getBestComputeType
"""Utilities for downloading and verifying CTranslate2 weights and tokenizers.
This module provides a small, dependency-light set of helpers used by the
translation layer. It purposely keeps behavior resilient: network errors are
logged (via utils.errorLogging) and the functions return/complete without
raising, which matches the repository's defensive style.
"""
ctranslate2_weights = {
"m2m100_418M-ct2-int8": {
"hf_repo": "jncraton/m2m100_418M-ct2-int8",
@@ -84,8 +95,8 @@ def downloadCTranslate2Tokenizer(path: str, weight_type: str = "m2m100_418M-ct2-
tokenizer = ctranslate2_weights[weight_type]["tokenizer"]
tokenizer_path = os_path.join(path, "weights", "ctranslate2", directory_name, "tokenizer")
try:
os_makedirs(tokenizer_path, exist_ok=True)
transformers.AutoTokenizer.from_pretrained(tokenizer, cache_dir=tokenizer_path)
os_makedirs(tokenizer_cache, exist_ok=True)
transformers.AutoTokenizer.from_pretrained(tokenizer_name, cache_dir=tokenizer_cache)
except Exception:
errorLogging()
tokenizer_path = os_path.join("./weights", "ctranslate2", directory_name, "tokenizer")

134
src-python/models/transliteration/transliteration_context_rules.py Normal file
View File

@@ -0,0 +1,134 @@
from typing import List, Dict, Any
import re
"""Contextual transliteration rules for tokenized results.
This module provides a compact rule engine that can modify token
readings (kana) based on neighboring tokens. Rules are embedded in
``DEFAULT_RULES`` to simplify packaging (no external JSON required).
Key points
- Rules are applied in descending ``priority`` order.
- Supported match modes: ``equals`` (exact match) and ``regex``.
- ``direction`` chooses whether to inspect the next or previous token.
- When a rule sets ``kana``, the engine overwrites ``kana`` and clears
``hira``/``hepburn``; callers should recompute them after rules run.
The engine mutates the provided ``results`` list in-place and also
returns it for convenience.
"""
DEFAULT_RULES = {
"rules": [
{
"name": "nan_next_tdna",
"target": "何",
"match_mode": "equals",
"direction": "next",
"kana_set": list("タチツテトダヂヅデドナニヌネノ"),
"on_true": {"kana": "ナン"},
"on_false": {"kana": "ナニ"}
}
]
}
def apply_context_rules(results: List[Dict[str, Any]], use_macron: bool = False) -> List[Dict[str, Any]]:
"""Apply contextual rewrite rules to `results`.
Parameters
- results: list of token dicts produced by Transliterator.split_kanji_okurigana
where each entry contains at least the keys: 'orig', 'kana', 'hira', 'hepburn'.
- use_macron: passed through for compatibility; rules themselves don't use it
Returns
- The (possibly modified) `results` list. The list is also modified in-place.
The engine supports 'equals' and 'regex' match modes, next/prev neighbor
inspection, and simple actions that overwrite `kana` (caller must recalc
`hira`/`hepburn` afterwards).
"""
# prepare rules: sort by priority (desc) and precompile regex where provided
raw_rules: List[Dict[str, Any]] = DEFAULT_RULES.get("rules", [])
rules = sorted(raw_rules, key=lambda r: r.get("priority", 0), reverse=True)
for r in rules:
if r.get("match_mode") == "regex" and r.get("pattern"):
try:
r["_re"] = re.compile(r["pattern"])
except Exception:
r["_re"] = None
i = 0
n = len(results)
while i < n:
entry = results[i]
orig = entry.get("orig", "")
# skip tokens with empty orig (symbols, whitespace, etc.)
if not orig:
i += 1
continue
for rule in rules:
target = rule.get("target")
mode = rule.get("match_mode", "equals")
direction = rule.get("direction", "next")
kana_set = set(rule.get("kana_set", []))
on_true = rule.get("on_true", {})
on_false = rule.get("on_false", {})
matched = False
if mode == "equals" and orig == target:
matched = True
elif mode == "regex":
cre = rule.get("_re")
if cre and cre.search(orig):
matched = True
# regex or other modes can be added later
if not matched:
continue
# decide neighbor token based on direction
neighbor_entry = None
if direction == "next":
j = i + 1
while j < n:
if results[j].get("orig"):
neighbor_entry = results[j]
break
j += 1
elif direction == "prev":
j = i - 1
while j >= 0:
if results[j].get("orig"):
neighbor_entry = results[j]
break
j -= 1
condition = False
if neighbor_entry:
nk = neighbor_entry.get("kana", "")
if nk:
first = nk[0]
if first in kana_set:
condition = True
else:
# fallback to orig-first-char check
fo = neighbor_entry.get("orig", "")[:1]
if fo and 'ァ' <= fo <= 'ン' and fo in kana_set:
condition = True
# Apply action: simple overwrite of kana/hira/hepburn for the matched token
action = on_true if condition else on_false
if "kana" in action:
entry["kana"] = action["kana"]
entry["hira"] = ""
entry["hepburn"] = ""
# once a rule applied, do not apply further rules to this token
break
i += 1
# return the (possibly modified) results for convenience/pure-function style usage
return results

217
src-python/models/transliteration/transliteration_kana_to_hepburn.py Normal file
View File

@@ -0,0 +1,217 @@
# katakana_to_hepburn.py
# カタカナ -> ヘボン式ローマ字（パッケージ不要）
from typing import List
def katakana_to_hepburn(kata: str, use_macron: bool = True) -> str:
"""
カタカナ文字列をヘボン式ローマ字に変換する。
use_macron=True のとき ā ī ū ē ō で長音を表現（マクロン）。
use_macron=False のときは単純に連続母音を残す（例: ou, oo）。
"""
# 基本音の対応（主要なカタカナ）
base: dict = {
'ア':'a','イ':'i','ウ':'u','エ':'e','オ':'o',
'カ':'ka','キ':'ki','ク':'ku','ケ':'ke','コ':'ko',
'サ':'sa','シ':'shi','ス':'su','セ':'se','ソ':'so',
'タ':'ta','チ':'chi','ツ':'tsu','テ':'te','ト':'to',
'ナ':'na','ニ':'ni','ヌ':'nu','ネ':'ne','ノ':'no',
'ハ':'ha','ヒ':'hi','フ':'fu','ヘ':'he','ホ':'ho',
'マ':'ma','ミ':'mi','ム':'mu','メ':'me','モ':'mo',
'ヤ':'ya','ユ':'yu','ヨ':'yo',
'ラ':'ra','リ':'ri','ル':'ru','レ':'re','ロ':'ro',
'ワ':'wa','ヲ':'wo','ン':'n',
'ガ':'ga','ギ':'gi','グ':'gu','ゲ':'ge','ゴ':'go',
'ザ':'za','ジ':'ji','ズ':'zu','ゼ':'ze','ゾ':'zo',
'ダ':'da','ヂ':'ji','ヅ':'zu','デ':'de','ド':'do',
'バ':'ba','ビ':'bi','ブ':'bu','ベ':'be','ボ':'bo',
'パ':'pa','ピ':'pi','プ':'pu','ペ':'pe','ポ':'po',
# 小書き（単独で使われることは少ないがマップしておく）
'ァ':'a','ィ':'i','ゥ':'u','ェ':'e','ォ':'o',
'ャ':'ya','ュ':'yu','ョ':'yo','ッ':'xtsu','ー':'-',
'ヴ':'vu','シェ':'she' # 特殊は下で組合せで処理
}
# 拡張：子音 + 小ャユョ の組合せ（主要なもの）
digraphs: dict = {
('キ','ャ'):'kya', ('キ','ュ'):'kyu', ('キ','ョ'):'kyo',
('ギ','ャ'):'gya', ('ギ','ュ'):'gyu', ('ギ','ョ'):'gyo',
('シ','ャ'):'sha', ('シ','ュ'):'shu', ('シ','ョ'):'sho',
('ジ','ャ'):'ja', ('ジ','ュ'):'ju', ('ジ','ョ'):'jo',
('チ','ャ'):'cha', ('チ','ュ'):'chu', ('チ','ョ'):'cho',
('ニ','ャ'):'nya', ('ニ','ュ'):'nyu', ('ニ','ョ'):'nyo',
('ヒ','ャ'):'hya', ('ヒ','ュ'):'hyu', ('ヒ','ョ'):'hyo',
('ビ','ャ'):'bya', ('ビ','ュ'):'byu', ('ビ','ョ'):'byo',
('ピ','ャ'):'pya', ('ピ','ュ'):'pyu', ('ピ','ョ'):'pyo',
('ミ','ャ'):'mya', ('ミ','ュ'):'myu', ('ミ','ョ'):'myo',
('リ','ャ'):'rya', ('リ','ュ'):'ryu', ('リ','ョ'):'ryo',
# 外来音対応（ファ/フィ/チェ 等）
('フ','ャ'):'fya', ('フ','ュ'):'fyu', ('フ','ョ'):'fyo',
('ト','ゥ'):'tu', ('ド','ゥ'):'du',
# F-sounds (ファ フィ フェ フォ)
('フ','ァ'):'fa', ('フ','ィ'):'fi', ('フ','ェ'):'fe', ('フ','ォ'):'fo',
# シェ チェ ティ etc.
('シ','ェ'):'she', ('チ','ェ'):'che',
('テ','ィ'):'ti',
('ウ','ァ'):'wa', ('ウ','ィ'):'wi', ('ウ','ェ'):'we', ('ウ','ォ'):'wo',
# その他外来語によくある組合せ
('ス','ィ'):'si', ('ズ','ィ'):'zi', ('ツ','ァ'):'tsa', ('ツ','ィ'):'tsi', ('ツ','ェ'):'tse', ('ツ','ォ'):'tso',
('キ','ェ'):'kye', ('ギ','ェ'):'gye',
('ヴ','ァ'):'va', ('ヴ','ィ'):'vi', ('ヴ','ェ'):'ve', ('ヴ','ォ'):'vo', ('ヴ','ュ'):'vyu'
}
# 小文字一覧（ゃゅょぁぃぅぇぉ など）
small_kana = set(['ャ','ュ','ョ','ァ','ィ','ゥ','ェ','ォ','ヮ','ヵ','ヶ','ッ','ャ','ュ','ョ'])
# マクロン変換マップ（連続母音 -> マクロン）
macron_map = {
'aa':'ā','ii':'ī','uu':'ū','ee':'ē','oo':'ō',
# ou -> ō という扱いを多くのヘボン式はする（特に日本語由来の長音）
'ou':'ō'
}
# Helper: 次のローマ字の先頭子音を取り出す（促音処理用）
def initial_consonant(rom: str) -> str:
# romはローマ字（例 'shi','chi','ta'）
# 子音は最初の母音直前までと考える（母音: a,i,u,e,o）
for i,ch in enumerate(rom):
if ch in 'aeiou':
return rom[:i]
return rom # 母音がないなら全部
# 変換メイン
res: List[str] = []
i = 0
kata = kata.strip()
length = len(kata)
while i < length:
ch = kata[i]
# 促音（ッ）：次の音の初めの子音を重ねる
if ch == 'ッ':
# lookahead
if i+1 < length:
# 先の1文字 or 合字を取り得る（小書きが続く可能性）
# まず合字優先で調べる
next_pair = None
if i+2 < length and (kata[i+1], kata[i+2]) in digraphs:
next_pair = digraphs[(kata[i+1], kata[i+2])]
elif kata[i+1] in base:
next_pair = base.get(kata[i+1])
if next_pair:
cons = initial_consonant(next_pair)
if cons == '':
# もし母音始まりなら促音は無視（稀）
pass
else:
# Hepburnでは "ch" の場合 "cch"（matcha）等の扱いになるように
# cons の先頭1文字を倍にするより、cons全体の先頭文字を重ねるのが一般的（例: 'shi' -> 'ssh' ? いい例は少ない）
# 実務上は先頭子音の最初の文字を重複する：
res.append(cons[0])
# advance only the 促音 itself here; next loop handles next kana
i += 1
continue
# 長音符（ー）：前の母音を伸ばす（マクロン処理は後でまとめて）
if ch == 'ー':
# append marker '-' to indicate prolong; we'll post-process
res.append('-')
i += 1
continue
# 合字（子 + 小ャュョ等）
if i+1 < length and (ch, kata[i+1]) in digraphs:
res.append(digraphs[(ch, kata[i+1])])
i += 2
continue
# 小書きが前に独立して出てきた場合（通常は合字で処理されるが念のため）
if ch in small_kana and ch != 'ッ':
# 小書きを単独で英字に変換（例: 'ァ' -> 'a'）
res.append(base.get(ch, ''))
i += 1
continue
# 普通のカタカナ
if ch in base:
res.append(base[ch])
i += 1
continue
# 英数字や記号・ひらがななどはそのまま（変換対象外）
res.append(ch)
i += 1
# ここまでで res はローマ字パーツのリスト（長音は '-' でマーク）
raw = ''.join(res)
# 撥音（ン）処理: n の前が b/p/m の場合 m にする
# ただし既に 'n' のまま次が母音や y の時は通常 n' を入れるべきだが簡易処理として n のまま保持。
# 我々は 'n' の後に b/p/m が来たら 'm' に置換
import re
raw = re.sub(r'n(?=[bmp])', 'm', raw)
# 長音処理（'-' マークを見て前の母音を伸ばす）
# raw 中の '-' を削って該当の母音を伸ばす
while '-' in raw:
idx = raw.find('-')
if idx == 0:
# 先頭に長音符が来るのはおかしいので削除
raw = raw[:idx] + raw[idx+1:]
continue
# 前の文字が母音ならそれを重ねる
prev = raw[idx-1]
if prev in 'aiueo':
# 直前に既に vowel がある場合、後でマクロン処理に任せて母音を2つにする
raw = raw[:idx] + prev + raw[idx+1:]
else:
# 直前が子音なら何もして取り除く
raw = raw[:idx] + raw[idx+1:]
# 小さな例外対応： 'ti' 等の表記は 'chi' と扱いたいが上述マップでカバー済み
# macron の適用（長音の正規化）
if use_macron:
# まず 'ou' を ō に（ただし語による例外はあるが、一般的ヘボンに合わせる）
# その前に 'oo' を 'ō' に（稀）
for pair, mac in macron_map.items():
raw = raw.replace(pair, mac)
# else: leave as is (ou/oo/aa...)
# 仕上げ：小文字統一（ヘボンは小文字）
raw = raw.lower()
# 最後に、n の後に母音または y が来る場合は「んあ->n'a」的扱いが必要だが
# シンプル実装では n の後に母音や y が来るときは n' を入れる（明瞭化）
# ただし多くの実例では省略されることも多いのでコメントアウトしておく
# raw = re.sub(r"n(?=[aiueoy])", "n'", raw)
return raw
# --- テスト例 ---
if __name__ == "__main__":
tests = [
"カタカナ",
"コンピューター",
"キャッチ",
"マッチャ",
"シェア",
"ジェット",
"ヴァイオリン",
"ホテル",
"スーパー",
"ギュウニュウ",
"パーティー",
"トウキョウ", # 東京（トウキョウ -> tōkyō）
"オーケー",
"ファイル",
"ニューヨーク",
"ラーメン",
"パン",
"チョコレート",
]
for s in tests:
print(s, "->", katakana_to_hepburn(s, use_macron=True))

230
src-python/models/transliteration/transliteration_transliterator.py Normal file
View File

@@ -0,0 +1,230 @@
from sudachipy import tokenizer
from sudachipy import dictionary
from typing import List, Dict, Any
import threading
try:
from .transliteration_kana_to_hepburn import katakana_to_hepburn
except ImportError:
from transliteration_kana_to_hepburn import katakana_to_hepburn
try:
from .transliteration_context_rules import apply_context_rules
except ImportError:
from transliteration_context_rules import apply_context_rules
class Transliterator:
def __init__(self) -> None:
self.tokenizer_obj = dictionary.Dictionary(dict_type="full").create()
self.mode = tokenizer.Tokenizer.SplitMode.C
# Lock to prevent concurrent access to sudachipy tokenizer which may
# internally use Rust/PyO3 borrow semantics and raise "Already borrowed".
self._tokenizer_lock = threading.Lock()
@staticmethod
def is_kanji(ch: str) -> bool:
return '\u4e00' <= ch <= '\u9fff'
@staticmethod
def kata_to_hira(text: str) -> str:
return "".join(
chr(ord(c) - 0x60) if 'ァ' <= c <= 'ン' else c
for c in text
)
@staticmethod
def split_kanji_okurigana(surface: str, reading_kana: str, use_macron: bool = True) -> List[Dict[str, str]]:
"""Split a single surface word and its kana reading into parts.
Inputs:
- surface: the surface form (may contain kanji + kana)
- reading_kana: the katakana reading for the whole surface
Output:
- a list of dicts: [{"orig": str, "kana": str, "hira": str, "hepburn": str}, ...]
Notes:
- The function allocates portions of ``reading_kana`` to each contiguous
kanji/non-kanji block in ``surface``. Allocation is heuristic: an
initial allocation based on block length is used and any remainder is
distributed left-to-right preferring kanji blocks.
- This function is pure (no external side effects) and returns the
constructed list.
"""
result: List[Dict[str, str]] = []
# 表層を「漢字ブロック」と「非漢字ブロック」に分割
buf = ""
prev_is_kanji = None
blocks = []
for ch in surface:
now_is_kanji = Transliterator.is_kanji(ch)
if prev_is_kanji is None or now_is_kanji == prev_is_kanji:
buf += ch
else:
blocks.append((prev_is_kanji, buf))
buf = ch
prev_is_kanji = now_is_kanji
if buf:
blocks.append((prev_is_kanji, buf))
# 読みを分配
kana_left = reading_kana
# We'll allocate kana to each block by initial guess = len(part) (characters)
# and distribute any remaining kana left-to-right preferring kanji blocks.
kana_len = len(kana_left)
# initial allocation per block
allocs = [len(part) for _, part in blocks]
allocated = sum(allocs)
remaining = kana_len - allocated
# distribute extra kana to kanji blocks first (left-to-right)
if remaining > 0:
for idx, (is_kan, _) in enumerate(blocks):
if remaining <= 0:
break
if is_kan:
allocs[idx] += 1
remaining -= 1
# if still remaining, distribute to all blocks left-to-right
idx = 0
while remaining > 0 and len(blocks) > 0:
allocs[idx] += 1
remaining -= 1
idx = (idx + 1) % len(blocks)
# if remaining < 0 (reading shorter than base), shrink allocations from right
if remaining < 0:
# remove from rightmost blocks as needed
need = -remaining
idx = len(blocks) - 1
while need > 0 and idx >= 0:
take = min(allocs[idx] - 1, need) if allocs[idx] > 1 else 0
allocs[idx] -= take
need -= take
idx -= 1
# now slice kana_left according to allocs
pos = 0
for (is_kan, part), cnt in zip(blocks, allocs):
kana_for_part = kana_left[pos:pos+cnt]
pos += cnt
result.append({
"orig": part,
"kana": kana_for_part,
"hira": Transliterator.kata_to_hira(kana_for_part),
"hepburn": katakana_to_hepburn(kana_for_part, use_macron=use_macron)
})
return result
def analyze(self, text: str, use_macron: bool = False) -> List[Dict[str, Any]]:
"""Tokenize ``text`` and produce per-subunit reading information.
Returns a list of dicts for each token/sub-part with keys:
- orig: original surface string (one or more characters)
- kana: katakana reading for this part (may be adapted by context rules)
- hira: hiragana reading (derived from kana)
- hepburn: Latin transcription (derived from kana)
Side-effects / notes:
- The function calls ``apply_context_rules(results, use_macron=...)``
which both mutates ``results`` in-place and returns it. This method
safely accepts the returned list and then recalculates ``hira`` and
``hepburn`` for entries whose ``kana`` was changed.
- If rule application fails, analysis still returns the best-effort
results.
"""
# Tokenizer may raise RuntimeError: Already borrowed when called
# concurrently. Protect the call with a lock to serialize access.
with self._tokenizer_lock:
tokens = self.tokenizer_obj.tokenize(text, self.mode)
results: List[Dict[str, Any]] = []
for t in tokens:
surface = t.surface()
reading = t.reading_form()
pos = t.part_of_speech()
if pos and pos[0] in ["記号", "補助記号", "空白"]:
reading = surface
if surface == reading:
results.append({
"orig": surface,
"kana": reading,
"hira": surface,
"hepburn": surface,
})
continue
# 単純に1文字ずつ処理
if len(surface) == 1:
# 1文字の場合はそのまま
results.append({
"orig": surface,
"kana": reading,
"hira": self.kata_to_hira(reading),
"hepburn": katakana_to_hepburn(reading, use_macron=use_macron)
})
else:
# 複数文字の場合は既存のユーティリティで分割
parts = self.split_kanji_okurigana(surface, reading, use_macron=use_macron)
results.extend(parts)
# 文脈ルールを適用（別ファイル）
try:
results = apply_context_rules(results, use_macron=use_macron) or results
except Exception:
# ルール適用で失敗しても解析結果は返す
pass
# apply_context_rules が kana を書き換えた場合、hira と hepburn を再計算
for entry in results:
kana = entry.get("kana", "")
if kana:
entry["hira"] = self.kata_to_hira(kana)
entry["hepburn"] = katakana_to_hepburn(kana, use_macron=use_macron)
return results
# --- テスト ---
if __name__ == "__main__":
import pprint
test_cases = [
"向こうへ行く",
"行事を行う",
"上がる",
"上る",
"入り込む",
"何",
"何が好き？",
"何色が好き？",
"何色ありますか？",
"何語ですか？",
"テーブルに色鉛筆は何色ありますか？"
"美しい花を見る",
"東京に行く",
"漢字とカタカナの混在",
"パーティーに行く",
"コンピューターを使う",
"シェアハウスに住む",
"ヴァイオリンを弾く",
"ギュウニュウを飲む",
"ニューヨークに行く",
"ラーメンを食べる",
"チョコレートが好き",
"SessionIDを取得する",
"取り敢えず検索してみる",
"見知らぬ土地で冒険する",
"彼は優れたエンジニアです",
" ".join(list("[]<>!@#$%^&*()_+-={}|\;:'\",.<>/?`~")),
" ".join(list("「」＜＞！＠＃＄％＾＆＊（）＿＋－＝｛｝｜＼；：＇＂，．／？｀～")),
" ".join(list("♪♫♬♭♮♯°℃℉№Å®©™✓✔✕✖★☆○●◎◇◆□■△▲▽▼※→←↑↓↔︎↕︎⇄⇅∞∴∵∷≪≫≦≧±×÷≠≈≡⊂⊃⊆⊇⊄⊅∪∩∈∋∅∀∃∠⊥⌒∂∇√∫∬∮∑∏∧∨¬⇒⇔∀∃∠⊥⌒∂∇√∫∬∮∑∏")),
" ".join(list("😀😃😄😁😆😅😂🤣😊😇🙂"))
]
transliterator = Transliterator()
for case in test_cases:
pprint.pprint(transliterator.analyze(case), sort_dicts=False)

102
src-python/models/watchdog/watchdog.py
View File

@@ -1,20 +1,104 @@
from typing import Callable
from typing import Callable, Optional
import time
from threading import Thread, Event
class Watchdog:
def __init__(self, timeout:int=60, interval:int=20):
"""A lightweight watchdog utility.
This class provides a minimal watchdog which records the last "feed"
timestamp and can invoke a user-supplied callback when the timeout
is exceeded. The design is intentionally simple: callers are expected
to either call `start()` periodically (e.g. from a loop) or extend the
class to run `start()` in a background thread.
Args:
timeout: seconds without feed after which the callback is invoked
interval: suggested sleep interval (seconds) for callers that poll
"""
def __init__(self, timeout: int = 60, interval: int = 20) -> None:
self.timeout = timeout
self.interval = interval
self.last_feed_time = time.time()
self.callback: Optional[Callable[[], None]] = None
# Background thread control
self._thread: Optional[Thread] = None
self._stop_event: Optional[Event] = None
def feed(self):
def feed(self) -> None:
"""Refresh the watchdog timer (set last feed time to now)."""
self.last_feed_time = time.time()
def setCallback(self, callback):
def setCallback(self, callback: Callable[[], None]) -> None:
"""Register a zero-argument callback invoked on timeout."""
self.callback = callback
def start(self):
if time.time() - self.last_feed_time > self.timeout:
if isinstance(self.callback, Callable):
self.callback()
time.sleep(self.interval)
def start(self) -> None:
"""Perform a single watchdog check and optionally sleep `interval` seconds.
The method checks if the duration since the last feed exceeds
`timeout`. If so and a callback is registered, the callback is called.
Note: `start()` does not run in the background by itself; callers
should call it repeatedly (or run it inside a thread) if continuous
monitoring is required.
"""
now = time.time()
if now - self.last_feed_time > self.timeout:
if callable(self.callback):
try:
self.callback()
except Exception:
# Do not let callback exceptions propagate out of watchdog
import traceback
traceback.print_exc()
time.sleep(self.interval)
def _run_loop(self) -> None:
"""Internal run loop used by `start_in_thread`.
It repeatedly calls `start()` until `_stop_event` is set. The
implementation relies on `start()` sleeping for `self.interval`.
"""
# Defensive: ensure stop_event exists
if self._stop_event is None:
return
while not self._stop_event.is_set():
self.start()
def start_in_thread(self, daemon: bool = True) -> None:
"""Start the watchdog in a background thread.
If the watchdog is already running, this is a no-op. The created
thread will repeatedly call `start()` until `stop()` is invoked.
Args:
daemon: if True, thread is a daemon thread (won't block process exit)
"""
if self._thread is not None and self._thread.is_alive():
return
self._stop_event = Event()
self._thread = Thread(target=self._run_loop, daemon=daemon)
self._thread.start()
def stop(self, timeout: Optional[float] = None) -> None:
"""Stop background thread started by `start_in_thread`.
If no background thread is running this is a no-op.
Args:
timeout: optional timeout to wait for thread join (seconds). If
None, join will block until the thread exits.
"""
if self._stop_event is None or self._thread is None:
return
# signal stop and wait for thread to finish
self._stop_event.set()
self._thread.join(timeout=timeout)
# cleanup
if self._thread.is_alive():
# thread did not stop within timeout; leave objects for another stop()
return
self._thread = None
self._stop_event = None