"""Qwen3-TTS 모델 서비스 래퍼 Voice Clone, Voice Design, Custom Voice 기능 제공 """ import os import io import logging from typing import Optional, Tuple, List from enum import Enum import torch import soundfile as sf import numpy as np logger = logging.getLogger(__name__) class VoiceType(str, Enum): """지원하는 음성 타입""" CLONE = "clone" # 레퍼런스 오디오 기반 복제 DESIGN = "design" # 텍스트 프롬프트 기반 설계 CUSTOM = "custom" # 프리셋 음성 # CustomVoice 모델의 프리셋 스피커 PRESET_SPEAKERS = [ "Chelsie", # 여성, 밝고 활기찬 "Ethan", # 남성, 차분한 "Vivian", # 여성, 부드러운 "Benjamin", # 남성, 깊은 "Aurora", # 여성, 우아한 "Oliver", # 남성, 친근한 "Luna", # 여성, 따뜻한 "Jasper", # 남성, 전문적인 "Aria", # 여성, 표현력 있는 ] # 언어 코드 매핑 LANGUAGE_MAP = { "ko": "Korean", "en": "English", "ja": "Japanese", "zh": "Chinese", "de": "German", "fr": "French", "ru": "Russian", "pt": "Portuguese", "es": "Spanish", "it": "Italian", } class QwenTTSService: """Qwen3-TTS 통합 서비스 세 가지 모델을 관리: - Base: Voice Clone용 - CustomVoice: 프리셋 음성용 - VoiceDesign: 음성 설계용 """ def __init__(self): self.device = "cuda:0" if torch.cuda.is_available() else "cpu" self.dtype = torch.bfloat16 if torch.cuda.is_available() else torch.float32 # 모델 ID (환경변수에서 가져오거나 기본값 사용) self.model_ids = { "base": os.getenv("MODEL_ID", "Qwen/Qwen3-TTS-12Hz-1.7B-Base"), "custom": os.getenv("CLONE_MODEL_ID", "Qwen/Qwen3-TTS-12Hz-1.7B-CustomVoice"), "design": os.getenv("DESIGN_MODEL_ID", "Qwen/Qwen3-TTS-12Hz-1.7B-VoiceDesign"), } # 모델 인스턴스 (lazy loading) self._models = {} self._initialized = False async def initialize(self, preload_models: Optional[List[str]] = None): """모델 초기화 (서버 시작 시 호출) Args: preload_models: 미리 로드할 모델 리스트 (예: ["custom", "design"]) """ if self._initialized: return logger.info(f"Qwen3-TTS 서비스 초기화 (device: {self.device})") # 기본적으로 CustomVoice 모델만 로드 (가장 많이 사용) if preload_models is None: preload_models = ["custom"] for model_type in preload_models: await self._load_model(model_type) self._initialized = True logger.info("Qwen3-TTS 서비스 초기화 완료") async def _load_model(self, model_type: str): """모델 로딩 (lazy loading)""" if model_type in self._models: return self._models[model_type] model_id = self.model_ids.get(model_type) if not model_id: raise ValueError(f"Unknown model type: {model_type}") logger.info(f"모델 로딩 중: {model_id}") try: from qwen_tts import Qwen3TTSModel # FlashAttention 사용 가능 여부 확인 attn_impl = "flash_attention_2" try: import flash_attn except ImportError: attn_impl = "sdpa" logger.warning("FlashAttention 미설치, SDPA 사용") model = Qwen3TTSModel.from_pretrained( model_id, device_map=self.device, dtype=self.dtype, attn_implementation=attn_impl, ) self._models[model_type] = model logger.info(f"모델 로드 완료: {model_id}") return model except Exception as e: logger.error(f"모델 로드 실패: {model_id} - {e}") raise def _get_language(self, lang_code: str) -> str: """언어 코드를 모델 언어명으로 변환""" return LANGUAGE_MAP.get(lang_code, "English") def _to_wav_bytes(self, audio: np.ndarray, sample_rate: int) -> bytes: """numpy 배열을 WAV 바이트로 변환""" buffer = io.BytesIO() sf.write(buffer, audio, sample_rate, format='WAV') buffer.seek(0) return buffer.read() async def synthesize_custom( self, text: str, speaker: str = "Chelsie", language: str = "ko", instruct: Optional[str] = None, ) -> Tuple[bytes, int]: """프리셋 음성으로 TTS 합성 Args: text: 합성할 텍스트 speaker: 프리셋 스피커 이름 language: 언어 코드 (ko, en, ja 등) instruct: 감정/스타일 지시 (선택) Returns: (WAV 바이트, 샘플레이트) """ model = await self._load_model("custom") lang = self._get_language(language) logger.info(f"Custom TTS 합성: speaker={speaker}, lang={lang}, text_len={len(text)}") wavs, sr = model.generate_custom_voice( text=text, language=lang, speaker=speaker, instruct=instruct, ) audio_bytes = self._to_wav_bytes(wavs[0], sr) return audio_bytes, sr async def synthesize_clone( self, text: str, ref_audio: bytes, ref_text: str, language: str = "ko", ) -> Tuple[bytes, int]: """Voice Clone으로 TTS 합성 Args: text: 합성할 텍스트 ref_audio: 레퍼런스 오디오 바이트 (3초 이상 권장) ref_text: 레퍼런스 오디오의 트랜스크립트 language: 언어 코드 Returns: (WAV 바이트, 샘플레이트) """ model = await self._load_model("base") lang = self._get_language(language) # 레퍼런스 오디오를 임시 파일로 저장 import tempfile with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as f: f.write(ref_audio) ref_audio_path = f.name try: logger.info(f"Voice Clone 합성: lang={lang}, text_len={len(text)}") wavs, sr = model.generate_voice_clone( text=text, language=lang, ref_audio=ref_audio_path, ref_text=ref_text, ) audio_bytes = self._to_wav_bytes(wavs[0], sr) return audio_bytes, sr finally: # 임시 파일 삭제 os.unlink(ref_audio_path) async def synthesize_design( self, text: str, instruct: str, language: str = "ko", ) -> Tuple[bytes, int]: """Voice Design으로 TTS 합성 Args: text: 합성할 텍스트 instruct: 음성 디자인 프롬프트 (예: "30대 남성, 부드럽고 차분한 목소리") language: 언어 코드 Returns: (WAV 바이트, 샘플레이트) """ model = await self._load_model("design") lang = self._get_language(language) logger.info(f"Voice Design 합성: lang={lang}, instruct={instruct[:50]}...") wavs, sr = model.generate_voice_design( text=text, language=lang, instruct=instruct, ) audio_bytes = self._to_wav_bytes(wavs[0], sr) return audio_bytes, sr def get_preset_speakers(self) -> List[str]: """프리셋 스피커 목록 반환""" return PRESET_SPEAKERS.copy() def get_supported_languages(self) -> dict: """지원 언어 목록 반환""" return LANGUAGE_MAP.copy() def is_initialized(self) -> bool: """초기화 상태 확인""" return self._initialized def get_loaded_models(self) -> List[str]: """현재 로드된 모델 목록""" return list(self._models.keys()) # 싱글톤 인스턴스 tts_service = QwenTTSService()