[改善] 型注釈の追加とドキュメントの更新

- config.py, controller.py, model.py, mainloop.py, utils.py での型注釈の追加
- CODING_RULES.md と api.md のドキュメントを更新
- 不要なコードの削除とリファクタリング

src-python/config.py

@@ -5,6 +5,7 @@ from os import path as os_path, makedirs as os_makedirs
 from json import load as json_load
 from json import dump as json_dump
 import threading
+from typing import Optional, Dict, Any
 import torch
 from device_manager import device_manager
 from models.translation.translation_languages import translation_lang
@@ -22,8 +23,8 @@ def json_serializable(var_name):
 
 class Config:
     _instance = None
-    _config_data = {}
+    _config_data: Dict[str, Any] = {}
-    _timer = None
+    _timer: Optional[threading.Timer] = None
     _debounce_time = 2
 
     def __new__(cls):

src-python/controller.py

@@ -1,5 +1,5 @@
 import copy
-from typing import Callable, Any
+from typing import Callable, Any, List, Optional
 from time import sleep
 from subprocess import Popen
 from threading import Thread
@@ -11,10 +11,14 @@ from utils import removeLog, printLog, errorLogging, isConnectedNetwork, isValid
 
 class Controller:
     def __init__(self) -> None:
-        self.init_mapping = {}
+        # typed attributes to satisfy static type checkers
-        self.run_mapping = {}
+        self.init_mapping: dict = {}
-        self.run = None
+        self.run_mapping: dict = {}
-        self.device_access_status = True
+        # initialize with a no-op callable so callers can safely call self.run
+        def _noop_run(status: int, endpoint: str, payload: Any = None) -> None:
+            return None
+        self.run: Callable[[int, str, Any], None] = _noop_run
+        self.device_access_status: bool = True
 
     def setInitMapping(self, init_mapping:dict) -> None:
         self.init_mapping = init_mapping
@@ -251,7 +255,7 @@ class Controller:
 
         elif isinstance(message, str) and len(message) > 0:
             translation = []
-            transliteration_message = []
+            transliteration_message: List[Any] = []
             transliteration_translation = []
             if model.checkKeywords(message):
                 self.run(
@@ -407,7 +411,7 @@ class Controller:
             )
         elif isinstance(message, str) and len(message) > 0:
             translation = []
-            transliteration_message = []
+            transliteration_message: List[Any] = []
             transliteration_translation = []
             if model.checkKeywords(message):
                 self.run(
@@ -566,12 +570,12 @@ class Controller:
                     translation_text = f" ({'/'.join(translation)})" if translation else ""
                     model.logger.info(f"[RECEIVED] {message}{translation_text}")
 
-    def chatMessage(self, data) -> None:
+    def chatMessage(self, data) -> dict:
         id = data["id"]
         message = data["message"]
         if len(message) > 0:
             translation = []
-            transliteration_message = []
+            transliteration_message: List[Any] = []
             transliteration_translation = []
             if config.ENABLE_TRANSLATION is False:
                 pass
@@ -739,6 +743,7 @@ class Controller:
             self.run_mapping["software_update_info"],
             software_update_info,
         )
+        return {"status":200, "result": software_update_info}
 
     @staticmethod
     def getComputeMode(*args, **kwargs) -> dict:
@@ -800,11 +805,15 @@ class Controller:
                     if is_vram_error:
                         # Defaultのデバイス設定に戻す
                         printLog("VRAM error detected, reverting device setting")
+                        self.run(
+                            400,
+                            self.run_mapping["error_translation_enable_vram_overflow"],
+                            {
+                                "message":"VRAM out of memory enabling translation",
+                                "data": error_message
+                            },
+                        )
                         self.setDisableTranslation()
-                        config.SELECTED_TRANSLATION_COMPUTE_DEVICE = copy.deepcopy(config.SELECTABLE_COMPUTE_DEVICE_LIST[0])
-                        config.SELECTED_TRANSLATION_COMPUTE_TYPE = "auto"
-                        self.run(200, self.run_mapping["selected_translation_compute_device"], config.SELECTED_TRANSLATION_COMPUTE_DEVICE)
-                        self.run(200, self.run_mapping["selected_translation_compute_type"], config.SELECTED_TRANSLATION_COMPUTE_TYPE)
                         self.run(
                             400,
                             self.run_mapping["enable_translation"],
@@ -2234,13 +2243,13 @@ class Controller:
         th_stopCheckSpeakerEnergy.join()
 
     @staticmethod
-    def startThreadingDownloadCtranslate2Weight(weight_type:str, callback:Callable[[float], None], end_callback:Callable[[float], None]) -> None:
+    def startThreadingDownloadCtranslate2Weight(weight_type:str, callback:Callable[[float], None], end_callback:Optional[Callable[..., None]] = None) -> None:
         th_download = Thread(target=model.downloadCTranslate2ModelWeight, args=(weight_type, callback, end_callback))
         th_download.daemon = True
         th_download.start()
 
     @staticmethod
-    def startThreadingDownloadWhisperWeight(weight_type:str, callback:Callable[[float], None], end_callback:Callable[[float], None]) -> None:
+    def startThreadingDownloadWhisperWeight(weight_type:str, callback:Callable[[float], None], end_callback:Optional[Callable[..., None]] = None) -> None:
         th_download = Thread(target=model.downloadWhisperModelWeight, args=(weight_type, callback, end_callback))
         th_download.daemon = True
         th_download.start()
@@ -2258,6 +2267,7 @@ class Controller:
     @staticmethod
     def setWatchdogCallback(callback) -> dict:
         model.setWatchdogCallback(callback)
+        return {"status":200, "result":True}
 
     @staticmethod
     def stopWatchdog(*args, **kwargs) -> dict:

src-python/docs/CODING_RULES.md

@@ -29,8 +29,6 @@
 - 定数: UPPER_SNAKE_CASE（`config.py` の定数に合わせる）。
 - run_mapping のキー: 現在は短い key（例: `transcription_mic`）を内部で使い `run_mapping` に `/run/...` を置いている。この慣習は維持する。Controller 内で `self.run_mapping[...]` を直接参照する実装は許容される。
 
-例: `selected_translation_compute_device` は内部 key、`/run/selected_translation_compute_device` が外部イベント名である点を区別して使う。
-
 ## モジュール・パッケージ構成
 - 各サブ領域（ocr, overlay, transcription, translation, websocket 等）は `models/` 下に整理済みのため、同様の粒度で新機能は追加する。
 - パッケージは必ず `__init__.py` を置く（static analysis / mypy のため）。空の `__init__.py` でも可。これにより相対インポートが安定する。

src-python/docs/api.md

@@ -130,9 +130,6 @@ run イベント
 `/run/selected_transcription_compute_type` (200)
   - payload: string
 
-`/run/selected_translation_compute_device` (200)
-  - payload: device descriptor (e.g. {"name":"cuda:0","type":"gpu"})
-
 `/run/selected_translation_engines` (200)
   - payload: config.SELECTED_TRANSLATION_ENGINES (list/dict per tab)
 

src-python/mainloop.py

@@ -1,7 +1,7 @@
 import sys
 import json
 import time
-from typing import Any
+from typing import Any, Tuple
 from threading import Thread
 from queue import Queue
 import logging
@@ -359,7 +359,8 @@ controller.setInitMapping(init_mapping)
 
 class Main:
     def __init__(self, controller_instance, mapping_data) -> None:
-        self.queue = Queue()
+        # queue holds tuples of (endpoint, data)
+        self.queue: Queue[Tuple[str, Any]] = Queue()
         self.main_loop = True
         self.controller = controller_instance
         self.mapping = mapping_data

src-python/model.py

@@ -10,7 +10,7 @@ from time import sleep
 from queue import Queue
 from threading import Thread
 from requests import get as requests_get
-from typing import Callable
+from typing import Callable, Optional, cast
 from packaging.version import parse
 
 from flashtext import KeywordProcessor
@@ -106,6 +106,9 @@ class Model:
         self.websocket_server_loop = False
         self.websocket_server_alive = False
         self.th_websocket_server = None
+        # default no-op callbacks for energy check functions
+        self.check_mic_energy_fnc: Callable[[float], None] = lambda v: None
+        self.check_speaker_energy_fnc: Callable[[float], None] = lambda v: None
 
     def checkTranslatorCTranslate2ModelWeight(self, weight_type:str):
         return checkCTranslate2Weight(config.PATH_LOCAL, weight_type)
@@ -291,9 +294,9 @@ class Model:
         if self.transliterator is not None:
             self.transliterator = None
 
-    def convertMessageToTransliteration(self, message: str, hiragana: bool=True, romaji: bool=True) -> str:
+    def convertMessageToTransliteration(self, message: str, hiragana: bool=True, romaji: bool=True) -> list:
         if hiragana is False and romaji is False:
-            return message
+            return []
 
         keys_to_keep = {"orig"}
         if hiragana:
@@ -574,9 +577,10 @@ class Model:
         #     self.mic_get_energy.stop()
         #     self.mic_get_energy = None
 
-    def startCheckMicEnergy(self, fnc:Callable[[float], None]=None) -> None:
+    def startCheckMicEnergy(self, fnc:Optional[Callable[[float], None]]=None) -> None:
-        if isinstance(fnc, Callable):
+        # fnc may be None or a callable. Use cast after checking for None to satisfy type checker.
-            self.check_mic_energy_fnc = fnc
+        if fnc is not None:
+            self.check_mic_energy_fnc = cast(Callable[[float], None], fnc)
 
         mic_host_name = config.SELECTED_MIC_HOST
         mic_device_name = config.SELECTED_MIC_DEVICE
@@ -596,7 +600,7 @@ class Model:
                         errorLogging()
                 sleep(0.01)
 
-            mic_energy_queue = Queue()
+            mic_energy_queue: Queue = Queue()
             mic_device = selected_mic_device[0]
             self.mic_energy_recorder = SelectedMicEnergyRecorder(mic_device)
             self.mic_energy_recorder.recordIntoQueue(mic_energy_queue)
@@ -614,17 +618,18 @@ class Model:
             self.mic_energy_recorder.stop()
             self.mic_energy_recorder = None
 
-    def startSpeakerTranscript(self, fnc):
+    def startSpeakerTranscript(self, fnc:Optional[Callable[[dict], None]]=None) -> None:
         speaker_device_name = config.SELECTED_SPEAKER_DEVICE
 
         speaker_device_list = device_manager.getSpeakerDevices()
         selected_speaker_device = [device for device in speaker_device_list if device["name"] == speaker_device_name]
 
         if len(selected_speaker_device) == 0 or speaker_device_name == "NoDevice":
+            # fnc may be None; only call if callable
+            if callable(fnc):
                 fnc({"text": False, "language": None})
         else:
-            speaker_audio_queue = Queue()
+            speaker_audio_queue: Queue = Queue()
-            # speaker_energy_queue = Queue()
             speaker_device = selected_speaker_device[0]
             record_timeout = config.SPEAKER_RECORD_TIMEOUT
             phrase_timeout = config.SPEAKER_PHRASE_TIMEOUT
@@ -708,9 +713,10 @@ class Model:
         #     self.speaker_get_energy.stop()
         #     self.speaker_get_energy = None
 
-    def startCheckSpeakerEnergy(self, fnc:Callable[[float], None]=None) -> None:
+    def startCheckSpeakerEnergy(self, fnc:Optional[Callable[[float], None]]=None) -> None:
-        if isinstance(fnc, Callable):
+        # Accept None as default and assign safely with cast after None-check
-            self.check_speaker_energy_fnc = fnc
+        if fnc is not None:
+            self.check_speaker_energy_fnc = cast(Callable[[float], None], fnc)
 
         speaker_device_name = config.SELECTED_SPEAKER_DEVICE
         speaker_device_list = device_manager.getSpeakerDevices()
@@ -720,7 +726,7 @@ class Model:
             self.check_speaker_energy_fnc(False)
         else:
             def sendSpeakerEnergy():
-                if speaker_energy_queue.empty() is False:
+                if not speaker_energy_queue.empty():
                     energy = speaker_energy_queue.get()
                     try:
                         self.check_speaker_energy_fnc(energy)
@@ -728,7 +734,7 @@ class Model:
                         errorLogging()
                 sleep(0.01)
 
-            speaker_energy_queue = Queue()
+            speaker_energy_queue: Queue = Queue()
             speaker_device = selected_speaker_device[0]
             self.speaker_energy_recorder = SelectedSpeakerEnergyRecorder(speaker_device)
             self.speaker_energy_recorder.recordIntoQueue(speaker_energy_queue)
@@ -746,9 +752,12 @@ class Model:
             self.speaker_energy_recorder.stop()
             self.speaker_energy_recorder = None
 
-    def createOverlayImageSmallLog(self, message:str, your_language:str, translation:list, target_language:dict):
+    def createOverlayImageSmallLog(self, message:Optional[str], your_language:Optional[str], translation:list, target_language:Optional[dict]) -> object:
-        target_language = [data["language"] for data in target_language.values() if data["enable"] is True]
+        # target_language may be provided as dict or None
-        return self.overlay_image.createOverlayImageSmallLog(message, your_language, translation, target_language)
+        target_language_list = []
+        if isinstance(target_language, dict):
+            target_language_list = [data["language"] for data in target_language.values() if data.get("enable") is True]
+        return self.overlay_image.createOverlayImageSmallLog(message, your_language, translation, target_language_list)
 
     def createOverlayImageSmallMessage(self, message):
         ui_language = config.UI_LANGUAGE
@@ -797,9 +806,12 @@ class Model:
         if (self.overlay.settings[size]["ui_scaling"] != config.OVERLAY_SMALL_LOG_SETTINGS["ui_scaling"]):
             self.overlay.updateUiScaling(config.OVERLAY_SMALL_LOG_SETTINGS["ui_scaling"], size)
 
-    def createOverlayImageLargeLog(self, message_type:str, message:str, your_language:str,  translation:list, target_language:dict):
+    def createOverlayImageLargeLog(self, message_type:str, message:Optional[str], your_language:Optional[str],  translation:list, target_language:Optional[dict]=None):
-        target_language = [data["language"] for data in target_language.values() if data["enable"] is True]
+        # normalize target_language dict -> list of language strings
-        return self.overlay_image.createOverlayImageLargeLog(message_type, message, your_language, translation, target_language)
+        target_language_list = []
+        if isinstance(target_language, dict):
+            target_language_list = [data["language"] for data in target_language.values() if data.get("enable") is True]
+        return self.overlay_image.createOverlayImageLargeLog(message_type, message, your_language, translation, target_language_list)
 
     def createOverlayImageLargeMessage(self, message):
         ui_language = config.UI_LANGUAGE

src-python/models/__init__.py

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

src-python/models/overlay/__init__.py

@@ -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"]

src-python/models/transcription/transcription_recorder.py

@@ -1,7 +1,6 @@
 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):

src-python/scripts/find_doc_tokens.py

@@ -7,7 +7,6 @@ tokens = [
     'transcription_mic',
     'transcription_speaker',
     'selected_translation_compute_device',
-    '/run/selected_translation_compute_device',
     '/run/transcription_mic',
     '/run/transcription_speaker',
 ]

src-python/utils.py

@@ -1,5 +1,5 @@
 import base64
-from typing import Any
+from typing import Any, List, Dict
 import json
 import traceback
 import logging
@@ -79,7 +79,7 @@ def isValidIpAddress(ip_address: str) -> bool:
     except ValueError:
         return False
 
-def getComputeDeviceList() -> dict:
+def getComputeDeviceList() -> List[Dict[str, Any]]:
     compute_types = [
         {
             "device": "cpu",
@@ -191,8 +191,8 @@ def printLog(log:str, data:Any=None) -> None:
         "data": str(data),
     }
     process_logger.info(response)
-    response = json.dumps(response)
+    serialized = json.dumps(response)
-    print(response, flush=True)
+    print(serialized, flush=True)
 
 def printResponse(status:int, endpoint:str, result:Any=None) -> None:
     global process_logger