追加: 型検査を mypy strict モードへ切り替え (#1144)

* fix: テストに型を追加

* fix: エンコード関数に型を追加

* fix: キャンセル可能合成に型と型チェックを追加

* fix: CDLL 関数呼び出しの型チェックを無効化

* fix: 外部関数返り値 Any に型を追加

* fix: バリデータ関数に型を追加

* fix: 型変更を伴う代入を削除

* fix: 入れ子関数の型チェックを回避

* fix: 入れ子関数の型チェックを回避

* fix: NumPy 型を追加

* fix: NumPy 型を追加

* add: mypy strict モードに切り替え

* fix: Any を型付けして ignore を回避

* 微調整

---------

Co-authored-by: Hiroshiba <[email protected]>

pyproject.toml

@@ -6,7 +6,7 @@ enable_black = true
 enable_flake8 = true
 enable_isort = true
 enable_mypy = true
-mypy_preset = "entry" # TODO: "strict"
+mypy_preset = "strict"
 mypy_plugins = [
   { function = "numpy.typing.mypy_plugin" },
   { function = "pydantic.mypy" },

run.py

@@ -98,7 +98,7 @@
 from voicevox_engine.utility.run_utility import decide_boolean_from_env
 
 
-def b64encode_str(s):
+def b64encode_str(s: bytes) -> str:
     return base64.b64encode(s).decode("utf-8")
 
 

test/setting/test_setting.py

@@ -79,7 +79,7 @@ def test_cors_policy_mode_type(self) -> None:
 
         self.assertIsInstance(settings.cors_policy_mode, CorsPolicyMode)
 
-    def test_invalid_cors_policy_mode_type(self):
+    def test_invalid_cors_policy_mode_type(self) -> None:
         with self.assertRaises(ValidationError):
             Setting(cors_policy_mode="invalid_value", allow_origin="*")
 

test/test_connect_base64_waves.py

@@ -18,7 +18,7 @@ def generate_sine_wave_ndarray(
     seconds: float, samplerate: int, frequency: float
 ) -> NDArray[np.float32]:
     x = np.linspace(0, seconds, int(seconds * samplerate), endpoint=False)
-    wave = np.sin(2 * np.pi * frequency * x).astype(np.float32)
+    wave: NDArray[np.float32] = np.sin(2 * np.pi * frequency * x).astype(np.float32)
 
     return wave
 

test/test_core_utility.py

@@ -6,13 +6,13 @@
 
 class TestHalfLogicalCores(TestCase):
     @patch("os.cpu_count", return_value=8)
-    def test_half_logical_cores_even(self, mock_cpu_count):
+    def test_half_logical_cores_even(self, mock_cpu_count: int) -> None:
         self.assertEqual(get_half_logical_cores(), 4)
 
     @patch("os.cpu_count", return_value=9)
-    def test_half_logical_cores_odd(self, mock_cpu_count):
+    def test_half_logical_cores_odd(self, mock_cpu_count: int) -> None:
         self.assertEqual(get_half_logical_cores(), 4)
 
     @patch("os.cpu_count", return_value=None)
-    def test_half_logical_cores_none(self, mock_cpu_count):
+    def test_half_logical_cores_none(self, mock_cpu_count: int) -> None:
         self.assertEqual(get_half_logical_cores(), 0)

test/test_library_manager.py

@@ -5,6 +5,7 @@
 from io import BytesIO
 from pathlib import Path
 from tempfile import TemporaryDirectory
+from typing import Any
 from unittest import TestCase
 from zipfile import ZipFile
 
@@ -54,7 +55,7 @@ def create_vvlib_without_manifest(self, filename: str) -> None:
                 if file.filename != vvlib_manifest_name:
                     zf_out.writestr(file, buffer)
 
-    def create_vvlib_manifest(self, **kwargs):
+    def create_vvlib_manifest(self, **kwargs: Any) -> dict[str, Any]:
         vvlib_manifest = copy.deepcopy(self.vvlib_manifest)
         return {**vvlib_manifest, **kwargs}
 

voicevox_engine/cancellable_engine.py

@@ -170,6 +170,11 @@ def _synthesis_impl(
         try:
             sub_proc_con1.send((query, style_id, core_version))
             f_name = sub_proc_con1.recv()
+            if isinstance(f_name, str):
+                audio_file_name = f_name
+            else:
+                # ここには来ないはず
+                raise HTTPException(status_code=500, detail="不正な値が生成されました")
         except EOFError:
             raise HTTPException(
                 status_code=500, detail="既にサブプロセスは終了されています"
@@ -179,9 +184,9 @@ def _synthesis_impl(
             raise
 
         self.finalize_con(request, proc, sub_proc_con1)
-        return f_name
+        return audio_file_name
 
-    async def catch_disconnection(self):
+    async def catch_disconnection(self) -> None:
         """
         接続監視を行うコルーチン
         """

voicevox_engine/core/core_adapter.py

@@ -109,7 +109,7 @@ def safe_yukarin_sa_forward(
         end_accent_phrase_list = np.r_[0, end_accent_phrase_list, 0]
 
         with self.mutex:
-            f0_list = self.core.yukarin_sa_forward(
+            f0_list: NDArray[np.float32] = self.core.yukarin_sa_forward(
                 length=vowel_phoneme_list.shape[0],
                 vowel_phoneme_list=vowel_phoneme_list[np.newaxis],
                 consonant_phoneme_list=consonant_phoneme_list[np.newaxis],

voicevox_engine/core/core_wrapper.py

@@ -595,7 +595,8 @@ def __init__(
             os.chdir(cwd)
 
     def metas(self) -> str:
-        return self.core.metas().decode("utf-8")
+        metas_bytes: bytes = self.core.metas()
+        return metas_bytes.decode("utf-8")
 
     def yukarin_s_forward(
         self,
@@ -897,7 +898,8 @@ def supported_devices(self) -> str:
         coreから取得した対応デバイスに関するjsonデータの文字列
         """
         if self.api_exists["supported_devices"]:
-            return self.core.supported_devices().decode("utf-8")
+            supported_devices_byte: bytes = self.core.supported_devices()
+            return supported_devices_byte.decode("utf-8")
         raise OldCoreError
 
     def finalize(self) -> None:
@@ -913,7 +915,8 @@ def load_model(self, style_id: int) -> None:
 
     def is_model_loaded(self, style_id: int) -> bool:
         if self.api_exists["is_model_loaded"]:
-            return self.core.is_model_loaded(c_long(style_id))
+            loaded_bool: bool = self.core.is_model_loaded(c_long(style_id))
+            return loaded_bool
         raise OldCoreError
 
     def assert_core_success(self, result: bool) -> None:

voicevox_engine/dev/tts_engine/mock.py

@@ -69,5 +69,5 @@ def forward(self, text: str, **kwargs: dict[str, Any]) -> NDArray[np.float32]:
         logger.info("[Mock] input text: %s" % text)
         wave, sr = tts(text)
         wave /= 2**15
-        wave = resample(wave, 48000, 24000)
-        return wave.astype(np.float32)
+        wave_resampled: NDArray[np.float64] = resample(wave, 48000, 24000)
+        return wave_resampled.astype(np.float32)

voicevox_engine/engine_manifest/EngineManifestLoader.py

@@ -13,7 +13,7 @@ def __init__(self, manifest_path: Path, root_dir: Path):
     def load_manifest(self) -> EngineManifest:
         manifest = json.loads(self.manifest_path.read_text(encoding="utf-8"))
 
-        manifest = EngineManifest(
+        return EngineManifest(
             manifest_version=manifest["manifest_version"],
             name=manifest["name"],
             brand_name=manifest["brand_name"],
@@ -49,4 +49,3 @@ def load_manifest(self) -> EngineManifest:
                 for key, item in manifest["supported_features"].items()
             },
         )
-        return manifest

voicevox_engine/model.py

@@ -235,7 +235,7 @@ class Config:
         validate_assignment = True
 
     @validator("surface")
-    def convert_to_zenkaku(cls, surface):
+    def convert_to_zenkaku(cls, surface: str) -> str:
         return surface.translate(
             str.maketrans(
                 "".join(chr(0x21 + i) for i in range(94)),
@@ -244,7 +244,7 @@ def convert_to_zenkaku(cls, surface):
         )
 
     @validator("pronunciation", pre=True)
-    def check_is_katakana(cls, pronunciation):
+    def check_is_katakana(cls, pronunciation: str) -> str:
         if not fullmatch(r"[ァ-ヴー]+", pronunciation):
             raise ValueError("発音は有効なカタカナでなくてはいけません。")
         sutegana = ["ァ", "ィ", "ゥ", "ェ", "ォ", "ャ", "ュ", "ョ", "ヮ", "ッ"]
@@ -268,7 +268,9 @@ def check_is_katakana(cls, pronunciation):
         return pronunciation
 
     @validator("mora_count", pre=True, always=True)
-    def check_mora_count_and_accent_type(cls, mora_count, values):
+    def check_mora_count_and_accent_type(
+        cls, mora_count: int | None, values: Any
+    ) -> int | None:
         if "pronunciation" not in values or "accent_type" not in values:
             # 適切な場所でエラーを出すようにする
             return mora_count

voicevox_engine/morphing.py

@@ -199,13 +199,14 @@ def synthesis_morphing(
         + morph_param.target_spectrogram * morph_rate
     )
 
-    y_h = pw.synthesize(
+    _y_h: NDArray[np.float64] = pw.synthesize(
         morph_param.base_f0,
         morph_spectrogram,
         morph_param.base_aperiodicity,
         morph_param.fs,
         morph_param.frame_period,
-    ).astype(np.float32)
+    )
+    y_h = _y_h.astype(np.float32)
 
     # TODO: tts_engine.py でのリサンプル処理と共通化する
     if output_fs != morph_param.fs:

voicevox_engine/tts_pipeline/tts_engine.py

@@ -153,7 +153,8 @@ def count_frame_per_unit(
 def _to_frame(sec: float) -> int:
     FRAMERATE = 93.75  # 24000 / 256 [frame/sec]
     # NOTE: `round` は偶数丸め。移植時に取扱い注意。詳細は voicevox_engine#552
-    return np.round(sec * FRAMERATE).astype(np.int32).item()
+    sec_rounded: NDArray[np.float64] = np.round(sec * FRAMERATE)
+    return sec_rounded.astype(np.int32).item()
 
 
 def apply_pitch_scale(moras: list[Mora], query: AudioQuery) -> list[Mora]:

voicevox_engine/utility/mutex_utility.py

@@ -6,14 +6,14 @@
 
 
 def mutex_wrapper(lock: threading.Lock) -> Callable[[F], F]:
-    def wrap(f):
-        def func(*args, **kw):
+    def wrap(f: F) -> F:
+        def func(*args: Any, **kw: Any) -> Any:
             lock.acquire()
             try:
                 return f(*args, **kw)
             finally:
                 lock.release()
 
-        return func
+        return func  # type: ignore
 
     return wrap