설치형 로컬 TTS를 MeloTTS 한국어(MIT)로 채택하고 엔드포인트로 연결
Higgs Audio v3 는 연구/비상업 라이선스라 config.py 가 environment != dev 에서 차단하고 있었다. 그 가드를 푸는 건 법적 판단이라 코드로 결정할 수 없어서, 상업 사용이 허용된 설치형을 다시 찾아 MeloTTS Korean 으로 바꿨다. 결과적으로 가드를 건드릴 필요 자체가 사라졌다 — Higgs 가드는 그대로 두고 provider 만 melotts 로 두면 운영에서도 동작한다. 검토 결과: - MeloTTS MIT 한국어 지원 -> 채택. CPU 실시간, 사전학습 다화자 - Kokoro-82M Apache2.0 한국어 없음 -> 탈락. 공식 VOICES.md 언어 목록에 부재 - Piper GPL -> 탈락 - XTTS-v2 / Fish Speech 비상업 -> 탈락. Higgs 와 같은 문제 사전학습 다화자 모델이라 실존 인물 reference 를 쓰지 않는다. Higgs 경로가 P1 프리셋 한정이던 이유가 없으므로 모든 페르소나 프리셋에 적용된다. 구현: - scripts/melotts-server.py loopback HTTP 사이드카(/health, POST /tts -> WAV) - voice_tts_provider=melotts 경로와 VIGNETTE_MELOTTS_TTS_* 설정 - scripts/start-melotts.ps1 런처(설치 순서 안내 포함) 실측: - CPU 정상 상태 RTF 0.27~0.28(실시간 3.6배). 첫 실행 13.25 는 모델 다운로드 - POST /tts 200, WAV 350,566 bytes, 3.61s, 헤더 provider/model/license - 빈 텍스트 422, 미지 경로 404 로 fail-closed - 왕복 검증: MeloTTS 합성음을 로컬 faster-whisper 가 완전 일치 전사 "그렇게 느끼셨군요. 조금 더 이야기해 주실 수 있을까요?" (word timestamp 8개) 설치 함정 3가지를 decisions/local-voice-stack.md 에 남겼다. librosa 0.9.1 의 pkg_resources(setuptools<81), MeloTTS 가 언어와 무관하게 임포트하는 일본어 unidic 사전, Windows 한국어 g2p 의 eunjeon. G7 게이트의 TTS 허용목록에 melotts 를 추가했다. 선언/실제 불일치 차단과 배치 STT 배제는 그대로다. 검증: API 914 passed, 사이드카 melotts 16/16 + whisper 37/37, SSOT FAIL 0, ruff clean.
This commit is contained in:
Yun Chan
parent
05aa7b312e
commit
2624d49984
15 changed files with 749 additions and 33 deletions
|
|
@ -32,14 +32,15 @@ MINIMUM_SOAK_SECONDS = 3_000.0
|
|||
|
||||
# 이 게이트는 특정 벤더가 아니라 **운영하기로 결정한 provider** 를 강제한다.
|
||||
# 2026-08-08 소유자 결정: STT 는 노트북 상주 faster-whisper(`local_whisper`),
|
||||
# TTS 는 노트북 Higgs(`higgs`). 이전 값은 벤더 하나(`deepgram`/`openai`)가
|
||||
# 하드코딩돼 있었을 뿐 결정 기록이 아니었다.
|
||||
# TTS 는 노트북 MeloTTS(`melotts`). 둘 다 MIT 라 운영 사용 제약이 없다.
|
||||
# 이전 값은 벤더 하나(`deepgram`/`openai`)가 하드코딩돼 있었을 뿐 결정 기록이 아니었다.
|
||||
# 근거는 docs/decisions/local-voice-stack.md.
|
||||
#
|
||||
# 목록은 닫혀 있다. 배치 STT(`openai`)는 interim/final 계약을 만족할 수 없어
|
||||
# 여기 들어오지 못한다. 어떤 경우에도 `expected_* == ready_*` 결속은 유지되므로
|
||||
# 선언한 provider 와 실제로 돈 provider 가 다르면 계속 실패한다.
|
||||
ALLOWED_STT_PROVIDERS = ("local_whisper", "deepgram")
|
||||
ALLOWED_TTS_PROVIDERS = ("higgs", "openai")
|
||||
ALLOWED_TTS_PROVIDERS = ("melotts", "higgs", "openai")
|
||||
|
||||
|
||||
def load_json(path: Path, errors: list[str], label: str) -> dict[str, Any]:
|
||||
|
|
|
|||
253
scripts/melotts-server.py
Normal file
253
scripts/melotts-server.py
Normal file
|
|
@ -0,0 +1,253 @@
|
|||
# -*- coding: utf-8 -*-
|
||||
"""로컬 상주 MeloTTS 한국어 TTS 사이드카.
|
||||
|
||||
Higgs 서버와 같은 loopback 상주 방식이지만 권리 조건이 다르다. MeloTTS 는 MIT
|
||||
라이선스라 상업·비상업 사용에 제약이 없고, 사전학습된 한국어 다화자 모델이라
|
||||
실존 인물 음성 reference 를 전혀 쓰지 않는다. 그래서 Higgs 처럼 dev 전용 가드나
|
||||
P1 프리셋 한정이 필요 없다.
|
||||
|
||||
엔드포인트:
|
||||
GET /health → {"status":"ok","model":...,"language":"KR","license":"MIT",...}
|
||||
POST /tts → {"text": "...", "speed": 1.0} 를 받아 WAV 바이트를 돌려준다
|
||||
|
||||
텍스트는 메모리에서만 다루고 디스크에 쓰지 않는다.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import array
|
||||
import io
|
||||
import json
|
||||
import math
|
||||
import sys
|
||||
import wave
|
||||
from dataclasses import dataclass
|
||||
from http.server import BaseHTTPRequestHandler, HTTPServer
|
||||
from typing import Any, Iterable, Protocol
|
||||
|
||||
|
||||
DEFAULT_HOST = "127.0.0.1"
|
||||
DEFAULT_PORT = 9883
|
||||
DEFAULT_LANGUAGE = "KR"
|
||||
DEFAULT_SPEED = 1.0
|
||||
MIN_SPEED = 0.5
|
||||
MAX_SPEED = 2.0
|
||||
MAX_TEXT_CHARS = 2_000
|
||||
MAX_BODY_BYTES = 64 * 1024
|
||||
MODEL_ID = "melotts-korean"
|
||||
LICENSE_ID = "MIT"
|
||||
|
||||
|
||||
class TtsError(RuntimeError):
|
||||
"""합성 실패. 입력 텍스트는 로그·응답에 다시 싣지 않는다."""
|
||||
|
||||
def __init__(self, code: str, *, http_status: int = 422):
|
||||
self.code = code
|
||||
self.http_status = http_status
|
||||
super().__init__(code)
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class TtsRequest:
|
||||
text: str
|
||||
speed: float
|
||||
|
||||
|
||||
class Synthesizer(Protocol):
|
||||
sample_rate: int
|
||||
|
||||
def synthesize(self, text: str, *, speed: float) -> Iterable[float]: ...
|
||||
|
||||
|
||||
def clamp_speed(value: Any) -> float:
|
||||
"""말 속도를 안전 범위로 접는다. 숫자가 아니면 기본값."""
|
||||
|
||||
try:
|
||||
speed = float(value)
|
||||
except (TypeError, ValueError):
|
||||
return DEFAULT_SPEED
|
||||
if math.isnan(speed) or math.isinf(speed):
|
||||
return DEFAULT_SPEED
|
||||
return max(MIN_SPEED, min(MAX_SPEED, speed))
|
||||
|
||||
|
||||
def parse_tts_request(payload: Any) -> TtsRequest:
|
||||
"""요청 본문을 검증한다. 빈 텍스트와 과대 입력은 fail-closed."""
|
||||
|
||||
if not isinstance(payload, dict):
|
||||
raise TtsError("invalid_body")
|
||||
text = payload.get("text")
|
||||
if not isinstance(text, str):
|
||||
raise TtsError("text_required")
|
||||
text = text.strip()
|
||||
if not text:
|
||||
raise TtsError("text_required")
|
||||
if len(text) > MAX_TEXT_CHARS:
|
||||
raise TtsError("text_too_long", http_status=413)
|
||||
return TtsRequest(text=text, speed=clamp_speed(payload.get("speed", DEFAULT_SPEED)))
|
||||
|
||||
|
||||
def encode_wav(samples: Iterable[float], sample_rate: int) -> bytes:
|
||||
"""float(-1..1) 시퀀스를 16-bit mono WAV 로 만든다."""
|
||||
|
||||
if sample_rate <= 0:
|
||||
raise TtsError("invalid_sample_rate", http_status=500)
|
||||
pcm = array.array("h")
|
||||
for sample in samples:
|
||||
value = float(sample)
|
||||
if math.isnan(value):
|
||||
value = 0.0
|
||||
clipped = max(-1.0, min(1.0, value))
|
||||
pcm.append(int(round(clipped * 32767)))
|
||||
if sys.byteorder != "little": # pragma: no cover - little-endian 개발 환경
|
||||
pcm.byteswap()
|
||||
buffer = io.BytesIO()
|
||||
with wave.open(buffer, "wb") as handle:
|
||||
handle.setnchannels(1)
|
||||
handle.setsampwidth(2)
|
||||
handle.setframerate(sample_rate)
|
||||
handle.writeframes(pcm.tobytes())
|
||||
return buffer.getvalue()
|
||||
|
||||
|
||||
def health_payload(synthesizer: Synthesizer, *, speakers: Iterable[str]) -> dict[str, Any]:
|
||||
return {
|
||||
"status": "ok",
|
||||
"model": MODEL_ID,
|
||||
"language": DEFAULT_LANGUAGE,
|
||||
"license": LICENSE_ID,
|
||||
"reference_policy": "pretrained-multispeaker-no-external-reference",
|
||||
"sample_rate": synthesizer.sample_rate,
|
||||
"speakers": sorted(speakers),
|
||||
}
|
||||
|
||||
|
||||
class MeloSynthesizer:
|
||||
"""상주 MeloTTS 모델 하나. 텍스트는 메모리에서만 다룬다."""
|
||||
|
||||
def __init__(self, *, device: str, language: str = DEFAULT_LANGUAGE) -> None:
|
||||
try:
|
||||
from melo.api import TTS
|
||||
except ImportError as exc: # pragma: no cover - 런타임 환경 의존
|
||||
raise TtsError("melotts_unavailable", http_status=500) from exc
|
||||
self._model = TTS(language=language, device=device)
|
||||
self.device = device
|
||||
self.language = language
|
||||
self.speaker_ids = dict(self._model.hps.data.spk2id)
|
||||
if not self.speaker_ids: # pragma: no cover - 모델 무결성
|
||||
raise TtsError("melotts_no_speaker", http_status=500)
|
||||
self._speaker_id = next(iter(self.speaker_ids.values()))
|
||||
self.sample_rate = int(self._model.hps.data.sampling_rate)
|
||||
|
||||
def synthesize(self, text: str, *, speed: float) -> Iterable[float]:
|
||||
audio = self._model.tts_to_file(
|
||||
text, self._speaker_id, None, speed=speed, quiet=True
|
||||
)
|
||||
return audio
|
||||
|
||||
|
||||
def make_handler(
|
||||
synthesizer: Synthesizer, *, speakers: Iterable[str]
|
||||
) -> type[BaseHTTPRequestHandler]:
|
||||
speaker_list = list(speakers)
|
||||
|
||||
class Handler(BaseHTTPRequestHandler):
|
||||
server_version = "VignetteMeloTTS/1"
|
||||
sys_version = ""
|
||||
|
||||
def log_message(self, format: str, *args: object) -> None:
|
||||
return None
|
||||
|
||||
def _json(self, status: int, payload: dict[str, Any]) -> None:
|
||||
body = json.dumps(payload, ensure_ascii=False).encode("utf-8")
|
||||
self.send_response(status)
|
||||
self.send_header("Content-Type", "application/json; charset=utf-8")
|
||||
self.send_header("Content-Length", str(len(body)))
|
||||
self.send_header("Cache-Control", "no-store")
|
||||
self.end_headers()
|
||||
self.wfile.write(body)
|
||||
|
||||
def do_GET(self) -> None: # noqa: N802
|
||||
if self.path != "/health":
|
||||
self._json(404, {"detail": "not_found"})
|
||||
return
|
||||
self._json(200, health_payload(synthesizer, speakers=speaker_list))
|
||||
|
||||
def do_POST(self) -> None: # noqa: N802
|
||||
if self.path != "/tts":
|
||||
self._json(404, {"detail": "not_found"})
|
||||
return
|
||||
try:
|
||||
length = int(self.headers.get("Content-Length", "0"))
|
||||
except ValueError:
|
||||
self._json(400, {"detail": "invalid_content_length"})
|
||||
return
|
||||
if length < 1 or length > MAX_BODY_BYTES:
|
||||
self._json(413, {"detail": "request_size_rejected"})
|
||||
return
|
||||
try:
|
||||
payload = json.loads(self.rfile.read(length))
|
||||
except (UnicodeDecodeError, ValueError):
|
||||
self._json(400, {"detail": "invalid_json"})
|
||||
return
|
||||
try:
|
||||
request = parse_tts_request(payload)
|
||||
audio = synthesizer.synthesize(request.text, speed=request.speed)
|
||||
wav = encode_wav(audio, synthesizer.sample_rate)
|
||||
except TtsError as exc:
|
||||
self._json(exc.http_status, {"detail": exc.code})
|
||||
return
|
||||
except Exception:
|
||||
self._json(500, {"detail": "synthesis_failed"})
|
||||
return
|
||||
self.send_response(200)
|
||||
self.send_header("Content-Type", "audio/wav")
|
||||
self.send_header("Content-Length", str(len(wav)))
|
||||
self.send_header("Cache-Control", "no-store")
|
||||
self.send_header("X-Vignette-TTS-Provider", "melotts")
|
||||
self.send_header("X-Vignette-TTS-Model", MODEL_ID)
|
||||
self.send_header("X-Vignette-TTS-License", LICENSE_ID)
|
||||
self.end_headers()
|
||||
self.wfile.write(wav)
|
||||
|
||||
return Handler
|
||||
|
||||
|
||||
def build_parser() -> argparse.ArgumentParser:
|
||||
parser = argparse.ArgumentParser(description=__doc__)
|
||||
parser.add_argument("--host", default=DEFAULT_HOST, choices=[DEFAULT_HOST, "localhost"])
|
||||
parser.add_argument("--port", type=int, default=DEFAULT_PORT)
|
||||
parser.add_argument("--device", default="auto")
|
||||
parser.add_argument("--language", default=DEFAULT_LANGUAGE)
|
||||
return parser
|
||||
|
||||
|
||||
def main(argv: Iterable[str] | None = None) -> int: # pragma: no cover - CLI
|
||||
args = build_parser().parse_args(list(argv) if argv is not None else None)
|
||||
synthesizer = MeloSynthesizer(device=args.device, language=args.language)
|
||||
print(
|
||||
json.dumps(
|
||||
{
|
||||
"ready": True,
|
||||
"host": args.host,
|
||||
"port": args.port,
|
||||
"model": MODEL_ID,
|
||||
"license": LICENSE_ID,
|
||||
"language": synthesizer.language,
|
||||
"device": synthesizer.device,
|
||||
"sample_rate": synthesizer.sample_rate,
|
||||
"speakers": sorted(synthesizer.speaker_ids),
|
||||
},
|
||||
ensure_ascii=False,
|
||||
separators=(",", ":"),
|
||||
),
|
||||
flush=True,
|
||||
)
|
||||
handler = make_handler(synthesizer, speakers=synthesizer.speaker_ids)
|
||||
HTTPServer((args.host, args.port), handler).serve_forever()
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__": # pragma: no cover - CLI
|
||||
raise SystemExit(main())
|
||||
93
scripts/start-melotts.ps1
Normal file
93
scripts/start-melotts.ps1
Normal file
|
|
@ -0,0 +1,93 @@
|
|||
param(
|
||||
[int]$Port = 9883,
|
||||
[string]$Language = 'KR',
|
||||
[ValidateSet('auto', 'cuda', 'cpu')]
|
||||
[string]$Device = 'auto',
|
||||
[int]$WaitReadySeconds = 300
|
||||
)
|
||||
|
||||
# 노트북 상주 MeloTTS 한국어 TTS 사이드카를 띄운다.
|
||||
# MIT 라이선스 사전학습 다화자 모델이라 상업 사용 제약도, 실존 인물 reference 문제도 없다.
|
||||
# 그래서 Higgs 와 달리 dev 전용 가드나 P1 프리셋 한정이 필요 없다.
|
||||
|
||||
$ErrorActionPreference = 'Stop'
|
||||
[Console]::OutputEncoding = [System.Text.UTF8Encoding]::new($false)
|
||||
$OutputEncoding = [System.Text.UTF8Encoding]::new($false)
|
||||
|
||||
$repoRoot = Split-Path -Parent $PSScriptRoot
|
||||
$serverScript = Join-Path $PSScriptRoot 'melotts-server.py'
|
||||
$venvPython = 'C:\Users\encep\.venvs\vignette-melotts\Scripts\python.exe'
|
||||
$healthUrl = "http://127.0.0.1:$Port/health"
|
||||
$runtimeLogDir = Join-Path $env:TEMP 'Vignette\melotts'
|
||||
$stdoutLog = Join-Path $runtimeLogDir 'server.out.log'
|
||||
$stderrLog = Join-Path $runtimeLogDir 'server.err.log'
|
||||
|
||||
if (!(Test-Path -LiteralPath $serverScript)) {
|
||||
throw "MeloTTS 서버 스크립트를 찾지 못했습니다: $serverScript"
|
||||
}
|
||||
if (!(Test-Path -LiteralPath $venvPython)) {
|
||||
throw @"
|
||||
MeloTTS 전용 venv 를 찾지 못했습니다: $venvPython
|
||||
설치 순서:
|
||||
python -m venv C:\Users\encep\.venvs\vignette-melotts
|
||||
& C:\Users\encep\.venvs\vignette-melotts\Scripts\python.exe -m pip install "git+https://github.com/myshell-ai/MeloTTS.git"
|
||||
& ... -m pip install "setuptools<81" eunjeon # librosa 0.9.1 의 pkg_resources, 한국어 g2p
|
||||
& ... -m unidic download # MeloTTS 가 무조건 임포트하는 일본어 사전
|
||||
"@
|
||||
}
|
||||
|
||||
try {
|
||||
$currentHealth = Invoke-RestMethod -Uri $healthUrl -Method Get -TimeoutSec 3
|
||||
if ($currentHealth.status -eq 'ok') {
|
||||
Write-Output "MeloTTS가 이미 준비됐습니다: $healthUrl"
|
||||
$currentHealth | ConvertTo-Json -Depth 4 -Compress
|
||||
return
|
||||
}
|
||||
} catch {
|
||||
# 아직 서버가 없으면 아래에서 시작한다.
|
||||
}
|
||||
|
||||
$listener = Get-NetTCPConnection -State Listen -LocalPort $Port -ErrorAction SilentlyContinue
|
||||
if ($null -ne $listener) {
|
||||
throw "포트 $Port 를 다른 프로세스가 사용 중입니다. 임의 종료하지 않았습니다."
|
||||
}
|
||||
|
||||
New-Item -ItemType Directory -Path $runtimeLogDir -Force | Out-Null
|
||||
$serverArgs = @(
|
||||
'-X', 'utf8', $serverScript,
|
||||
'--host', '127.0.0.1',
|
||||
'--port', "$Port",
|
||||
'--language', $Language,
|
||||
'--device', $Device
|
||||
)
|
||||
$process = Start-Process -WindowStyle Hidden -FilePath $venvPython `
|
||||
-ArgumentList $serverArgs `
|
||||
-WorkingDirectory $repoRoot `
|
||||
-RedirectStandardOutput $stdoutLog `
|
||||
-RedirectStandardError $stderrLog `
|
||||
-PassThru
|
||||
|
||||
Write-Output "MeloTTS 로드를 시작했습니다. PID=$($process.Id) language=$Language device=$Device"
|
||||
Write-Output "로그: $stdoutLog"
|
||||
if ($WaitReadySeconds -le 0) {
|
||||
return
|
||||
}
|
||||
|
||||
$deadline = (Get-Date).AddSeconds($WaitReadySeconds)
|
||||
while ((Get-Date) -lt $deadline) {
|
||||
if ($process.HasExited) {
|
||||
throw "MeloTTS가 준비되기 전에 종료됐습니다. stderr=$stderrLog"
|
||||
}
|
||||
try {
|
||||
$health = Invoke-RestMethod -Uri $healthUrl -Method Get -TimeoutSec 3
|
||||
if ($health.status -eq 'ok') {
|
||||
Write-Output "MeloTTS 준비 완료: $healthUrl"
|
||||
$health | ConvertTo-Json -Depth 4 -Compress
|
||||
return
|
||||
}
|
||||
} catch {
|
||||
Start-Sleep -Seconds 3
|
||||
}
|
||||
}
|
||||
|
||||
throw "MeloTTS 준비 시간이 ${WaitReadySeconds}초를 넘었습니다. 로그: $stdoutLog"
|
||||
|
|
@ -155,10 +155,10 @@ def public_soak() -> dict[str, object]:
|
|||
"ready_stt_provider": "local_whisper",
|
||||
"expected_stt_model": "large-v3",
|
||||
"ready_stt_model": "large-v3",
|
||||
"expected_tts_provider": "higgs",
|
||||
"ready_tts_provider": "higgs",
|
||||
"expected_tts_model": "higgs-audio-v3-tts-4b",
|
||||
"ready_tts_model": "higgs-audio-v3-tts-4b",
|
||||
"expected_tts_provider": "melotts",
|
||||
"ready_tts_provider": "melotts",
|
||||
"expected_tts_model": "melotts-korean",
|
||||
"ready_tts_model": "melotts-korean",
|
||||
"turns_attempted": 10,
|
||||
"turns_succeeded": 10,
|
||||
"turn_transcript_metrics": metrics,
|
||||
|
|
@ -287,12 +287,12 @@ class G7ExternalProofTests(unittest.TestCase):
|
|||
self.assertGreaterEqual(gain["held_out_participants"], 30)
|
||||
|
||||
def test_decided_local_stack_is_accepted(self) -> None:
|
||||
"""2026-08-08 소유자 결정: 노트북 faster-whisper STT + Higgs TTS."""
|
||||
"""2026-08-08 소유자 결정: 노트북 faster-whisper STT + MeloTTS TTS (둘 다 MIT)."""
|
||||
|
||||
errors: list[str] = []
|
||||
voice = public_soak()
|
||||
self.assertEqual(voice["expected_stt_provider"], "local_whisper")
|
||||
self.assertEqual(voice["expected_tts_provider"], "higgs")
|
||||
self.assertEqual(voice["expected_tts_provider"], "melotts")
|
||||
self.checker.validate_public_soak(voice, errors)
|
||||
self.assertEqual([], errors)
|
||||
|
||||
|
|
|
|||
139
scripts/test_melotts_server.py
Normal file
139
scripts/test_melotts_server.py
Normal file
|
|
@ -0,0 +1,139 @@
|
|||
from __future__ import annotations
|
||||
|
||||
import importlib.util
|
||||
import io
|
||||
import math
|
||||
import sys
|
||||
import unittest
|
||||
import wave
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
SCRIPT_PATH = Path(__file__).with_name("melotts-server.py")
|
||||
SPEC = importlib.util.spec_from_file_location("melotts_server", SCRIPT_PATH)
|
||||
assert SPEC is not None and SPEC.loader is not None
|
||||
MODULE = importlib.util.module_from_spec(SPEC)
|
||||
sys.modules[SPEC.name] = MODULE
|
||||
SPEC.loader.exec_module(MODULE)
|
||||
|
||||
|
||||
class FakeSynthesizer:
|
||||
sample_rate = 44_100
|
||||
|
||||
def __init__(self) -> None:
|
||||
self.calls: list[tuple[str, float]] = []
|
||||
|
||||
def synthesize(self, text: str, *, speed: float):
|
||||
self.calls.append((text, speed))
|
||||
return [0.0, 0.5, -0.5, 1.0, -1.0]
|
||||
|
||||
|
||||
class ClampSpeedTest(unittest.TestCase):
|
||||
def test_in_range_speed_is_kept(self) -> None:
|
||||
self.assertEqual(MODULE.clamp_speed(1.25), 1.25)
|
||||
|
||||
def test_out_of_range_speed_is_folded(self) -> None:
|
||||
self.assertEqual(MODULE.clamp_speed(99), MODULE.MAX_SPEED)
|
||||
self.assertEqual(MODULE.clamp_speed(-99), MODULE.MIN_SPEED)
|
||||
|
||||
def test_non_numeric_and_nan_fall_back_to_default(self) -> None:
|
||||
for value in ("fast", None, [], float("nan"), float("inf")):
|
||||
with self.subTest(value=value):
|
||||
self.assertEqual(MODULE.clamp_speed(value), MODULE.DEFAULT_SPEED)
|
||||
|
||||
|
||||
class ParseRequestTest(unittest.TestCase):
|
||||
def test_valid_request_is_normalized(self) -> None:
|
||||
request = MODULE.parse_tts_request({"text": " 안녕하세요 ", "speed": 1.1})
|
||||
self.assertEqual(request.text, "안녕하세요")
|
||||
self.assertAlmostEqual(request.speed, 1.1)
|
||||
|
||||
def test_speed_defaults_when_absent(self) -> None:
|
||||
request = MODULE.parse_tts_request({"text": "안녕"})
|
||||
self.assertEqual(request.speed, MODULE.DEFAULT_SPEED)
|
||||
|
||||
def test_empty_or_missing_text_fails_closed(self) -> None:
|
||||
for payload in ({}, {"text": ""}, {"text": " "}, {"text": 5}, []):
|
||||
with self.subTest(payload=payload):
|
||||
with self.assertRaises(MODULE.TtsError):
|
||||
MODULE.parse_tts_request(payload)
|
||||
|
||||
def test_oversized_text_is_rejected_with_413(self) -> None:
|
||||
with self.assertRaises(MODULE.TtsError) as ctx:
|
||||
MODULE.parse_tts_request({"text": "가" * (MODULE.MAX_TEXT_CHARS + 1)})
|
||||
self.assertEqual(ctx.exception.http_status, 413)
|
||||
self.assertEqual(ctx.exception.code, "text_too_long")
|
||||
|
||||
|
||||
class EncodeWavTest(unittest.TestCase):
|
||||
def _read(self, payload: bytes):
|
||||
with wave.open(io.BytesIO(payload), "rb") as handle:
|
||||
return handle.getnchannels(), handle.getsampwidth(), handle.getframerate(), handle.readframes(handle.getnframes())
|
||||
|
||||
def test_wav_header_is_mono_16bit(self) -> None:
|
||||
payload = MODULE.encode_wav([0.0, 0.25], 22_050)
|
||||
channels, width, rate, frames = self._read(payload)
|
||||
self.assertEqual((channels, width, rate), (1, 2, 22_050))
|
||||
self.assertEqual(len(frames), 4)
|
||||
|
||||
def test_samples_are_clipped_into_int16_range(self) -> None:
|
||||
payload = MODULE.encode_wav([2.0, -2.0], 16_000)
|
||||
_, _, _, frames = self._read(payload)
|
||||
self.assertEqual(
|
||||
int.from_bytes(frames[0:2], "little", signed=True), 32_767
|
||||
)
|
||||
self.assertEqual(
|
||||
int.from_bytes(frames[2:4], "little", signed=True), -32_767
|
||||
)
|
||||
|
||||
def test_nan_samples_become_silence(self) -> None:
|
||||
payload = MODULE.encode_wav([math.nan], 16_000)
|
||||
_, _, _, frames = self._read(payload)
|
||||
self.assertEqual(int.from_bytes(frames[0:2], "little", signed=True), 0)
|
||||
|
||||
def test_empty_audio_still_produces_a_valid_wav(self) -> None:
|
||||
payload = MODULE.encode_wav([], 16_000)
|
||||
channels, width, rate, frames = self._read(payload)
|
||||
self.assertEqual((channels, width, rate), (1, 2, 16_000))
|
||||
self.assertEqual(frames, b"")
|
||||
|
||||
def test_invalid_sample_rate_fails_closed(self) -> None:
|
||||
with self.assertRaises(MODULE.TtsError):
|
||||
MODULE.encode_wav([0.0], 0)
|
||||
|
||||
|
||||
class HealthPayloadTest(unittest.TestCase):
|
||||
def test_health_declares_the_permissive_license_and_no_reference(self) -> None:
|
||||
payload = MODULE.health_payload(FakeSynthesizer(), speakers=["KR"])
|
||||
self.assertEqual(payload["status"], "ok")
|
||||
self.assertEqual(payload["license"], "MIT")
|
||||
self.assertEqual(payload["language"], "KR")
|
||||
self.assertEqual(payload["sample_rate"], 44_100)
|
||||
self.assertEqual(payload["speakers"], ["KR"])
|
||||
self.assertIn("no-external-reference", payload["reference_policy"])
|
||||
|
||||
|
||||
class CliTest(unittest.TestCase):
|
||||
def test_host_is_loopback_only(self) -> None:
|
||||
parser = MODULE.build_parser()
|
||||
with self.assertRaises(SystemExit):
|
||||
parser.parse_args(["--host", "0.0.0.0"])
|
||||
|
||||
def test_defaults_target_the_reserved_sidecar_port(self) -> None:
|
||||
args = MODULE.build_parser().parse_args([])
|
||||
self.assertEqual(args.port, MODULE.DEFAULT_PORT)
|
||||
self.assertEqual(args.language, "KR")
|
||||
|
||||
|
||||
class SynthesisPipelineTest(unittest.TestCase):
|
||||
def test_request_to_wav_round_trip(self) -> None:
|
||||
synthesizer = FakeSynthesizer()
|
||||
request = MODULE.parse_tts_request({"text": "안녕하세요", "speed": 5})
|
||||
audio = synthesizer.synthesize(request.text, speed=request.speed)
|
||||
payload = MODULE.encode_wav(audio, synthesizer.sample_rate)
|
||||
self.assertEqual(synthesizer.calls, [("안녕하세요", MODULE.MAX_SPEED)])
|
||||
self.assertTrue(payload.startswith(b"RIFF"))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Loading…
Add table
Add a link
Reference in a new issue