import os import re from typing import List, Optional, Tuple, Union import numpy as np import torch # ----------------------------------------------------------------------------- # 1. Text Tokenizer (Qwen2.5 / Breeze-TTS Standardı) # ----------------------------------------------------------------------------- class TextTokenizer: """ Qwen2.5 tabanlı gelişmiş BPE Metin Tokenizer'ı (Breeze-TTS standardı). Özel kontrol tokenları ile Standart TTS, Voice Design ve Voice Clone destekler: - <|instruct|> : Ses tasarım talimatı (Voice Design) - <|text|> : Seslendirilecek metin - <|ref_audio|>: Klonlanacak referans ses (Voice Clone) - <|audio|> : Ses tokenlarının başladığı yer - <|audio_end|>: Ses tokenlarının bittiği yer """ def __init__(self, model_id: str = "Qwen/Qwen2.5-0.5B"): from transformers import AutoTokenizer self.model_id = model_id try: self.tokenizer = AutoTokenizer.from_pretrained(model_id, local_files_only=True) except Exception: self.tokenizer = AutoTokenizer.from_pretrained(model_id) # Özel kontrol tokenlarını ekle self.special_tokens = [ "<|instruct|>", "<|text|>", "<|ref_audio|>", "<|audio|>", "<|audio_end|>", ] self.tokenizer.add_special_tokens({"additional_special_tokens": self.special_tokens}) # Token ID erişimleri self.pad_id = self.tokenizer.pad_token_id if self.tokenizer.pad_token_id is not None else 0 self.bos_id = self.tokenizer.bos_token_id if self.tokenizer.bos_token_id is not None else 1 self.eos_id = self.tokenizer.eos_token_id if self.tokenizer.eos_token_id is not None else 2 self.instruct_id = self.tokenizer.convert_tokens_to_ids("<|instruct|>") self.text_id = self.tokenizer.convert_tokens_to_ids("<|text|>") self.ref_audio_id = self.tokenizer.convert_tokens_to_ids("<|ref_audio|>") self.audio_start_id = self.tokenizer.convert_tokens_to_ids("<|audio|>") self.audio_end_id = self.tokenizer.convert_tokens_to_ids("<|audio_end|>") self.vocab_size = len(self.tokenizer) def encode(self, text: str, add_bos: bool = False, add_eos: bool = False) -> List[int]: """ Metni token ID'lerine dönüştürür. """ ids = self.tokenizer.encode(text, add_special_tokens=False) if add_bos and self.bos_id is not None: ids = [self.bos_id] + ids if add_eos and self.eos_id is not None: ids = ids + [self.eos_id] return ids def decode(self, ids: List[int], skip_special_tokens: bool = False) -> str: """ Token ID'lerini tekrar metne çevirir. """ return self.tokenizer.decode(ids, skip_special_tokens=skip_special_tokens) # ----------------------------------------------------------------------------- # 2. Audio Codec Tokenizer (Kyutai Mimi) # ----------------------------------------------------------------------------- class AudioCodecTokenizer: """ Kyutai Mimi Audio Codec Entegrasyonu: 24 kHz dalga boyunu saniyede 12.5 kare ve 8 codebook ile ayrık sayılara çevirir. Ağırlıklar dondurulmuştur (frozen), sadece encode/decode için kullanılır. """ def __init__(self, model_id: str = "kyutai/mimi", device: str = "cpu"): self.model_id = model_id self.device = device self.sample_rate = 24000 self.frame_rate = 12.5 self.num_codebooks = 8 self.codebook_size = 2048 from transformers import MimiModel, AutoFeatureExtractor print(f"Kyutai Mimi codec yükleniyor ({model_id})...") try: self.codec = MimiModel.from_pretrained(model_id, local_files_only=True).to(device) self.feature_extractor = AutoFeatureExtractor.from_pretrained(model_id, local_files_only=True) except Exception: self.codec = MimiModel.from_pretrained(model_id).to(device) self.feature_extractor = AutoFeatureExtractor.from_pretrained(model_id) self.codec.eval() print("Kyutai Mimi başarıyla yüklendi!") @torch.inference_mode() def encode(self, wav: Union[np.ndarray, torch.Tensor]) -> torch.Tensor: """ Giriş: (1, audio_len) 24 kHz ses Çıkış: (8, T_audio) ayrık token matrisi """ if isinstance(wav, torch.Tensor): wav_np = wav.squeeze().cpu().numpy() else: wav_np = wav.squeeze() inputs = self.feature_extractor( raw_audio=wav_np, sampling_rate=self.sample_rate, return_tensors="pt" ).to(self.device) encoder_outputs = self.codec.encode(inputs["input_values"], inputs.get("padding_mask")) # audio_codes shape: (1, num_codebooks, T) -> hedef codebook sayısına (8) dilimle codes = encoder_outputs.audio_codes.squeeze(0) return codes[:self.num_codebooks, :] @torch.inference_mode() def decode(self, audio_codes: torch.Tensor) -> torch.Tensor: """ Giriş: (8, T_audio) veya (1, 8, T_audio) ayrık token matrisi Çıkış: (1, audio_len) 24 kHz dalga boyu """ if audio_codes.dim() == 2: audio_codes = audio_codes.unsqueeze(0) # (1, 8, T) # Mimi decode: (1, 8, T) -> (1, 1, audio_len) veya (1, audio_len) audio_values = self.codec.decode(audio_codes.to(self.device))[0] if audio_values.dim() == 3: audio_values = audio_values.squeeze(1) # (1, audio_len) elif audio_values.dim() == 1: audio_values = audio_values.unsqueeze(0) return audio_values # ----------------------------------------------------------------------------- # 3. TTS Processor (Prompt Hazırlayıcı) # ----------------------------------------------------------------------------- class TTSProcessor: """ Metin, Voice Design talimatı ve Ses Klonlama girdilerini modele beslenecek formatta hazırlayan yönetici sınıf. """ def __init__(self, text_tokenizer: TextTokenizer, audio_tokenizer: AudioCodecTokenizer): self.text_tokenizer = text_tokenizer self.audio_tokenizer = audio_tokenizer def format_text_prompt(self, text: str, instruction: Optional[str] = None) -> str: """ Senaryolara göre uygun prompt string'i üretir: 1. Standart: <|text|> {text} <|audio|> 2. Voice Design: <|instruct|> {instruction} <|text|> {text} <|audio|> """ if instruction is not None and instruction.strip(): return f"<|instruct|> {instruction.strip()} <|text|> {text.strip()} <|audio|>" return f"<|text|> {text.strip()} <|audio|>" def prepare_inference_inputs( self, text: str, instruction: Optional[str] = None, ref_audio_path: Optional[str] = None, max_ref_sec: Optional[float] = None, device: str = "cpu" ) -> Tuple[torch.Tensor, Optional[torch.Tensor]]: """ İnference için gerekli tensörleri (text_ids, ref_audio_codes) üretir. """ prompt_str = self.format_text_prompt(text, instruction=instruction) text_ids = torch.tensor([self.text_tokenizer.encode(prompt_str)], device=device, dtype=torch.long) ref_codes = None if ref_audio_path is not None and os.path.exists(ref_audio_path): try: import soundfile as sf wav, sr = sf.read(ref_audio_path) if wav.ndim > 1: wav = wav.mean(axis=1) # Mono'ya dönüştür target_sr = self.audio_tokenizer.sample_rate # 24000 if sr != target_sr: import torchaudio.functional as AF wav_t = torch.from_numpy(wav).float().unsqueeze(0) wav = AF.resample(wav_t, orig_freq=sr, new_freq=target_sr).squeeze(0).numpy() # Kullanıcı sınır belirtmişse kırp, belirtmemişse sesin TAMAMINI al if max_ref_sec is not None and max_ref_sec > 0: max_ref_samples = int(target_sr * max_ref_sec) if len(wav) > max_ref_samples: wav = wav[:max_ref_samples] ref_codes = self.audio_tokenizer.encode(wav).unsqueeze(0).to(device) # (1, 8, T_ref) ref_sec = ref_codes.shape[-1] / self.audio_tokenizer.frame_rate print(f"[Inference] 🎙️ Referans ses işlendi ({ref_audio_path}): {ref_sec:.2f} sn ({ref_codes.shape[-1]} kare - Tamamı alındı)", flush=True) except Exception as e: print(f"[Inference] Referans ses okunamadı ({ref_audio_path}): {e}", flush=True) return text_ids, ref_codes