| import os |
| import re |
| from typing import List, Optional, Tuple, Union |
| import numpy as np |
| import torch |
|
|
|
|
| |
| |
| |
|
|
| 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) |
|
|
| |
| self.special_tokens = [ |
| "<|instruct|>", |
| "<|text|>", |
| "<|ref_audio|>", |
| "<|audio|>", |
| "<|audio_end|>", |
| ] |
| self.tokenizer.add_special_tokens({"additional_special_tokens": self.special_tokens}) |
|
|
| |
| 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) |
|
|
|
|
| |
| |
| |
|
|
| 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")) |
| |
| 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) |
|
|
| |
| audio_values = self.codec.decode(audio_codes.to(self.device))[0] |
| if audio_values.dim() == 3: |
| audio_values = audio_values.squeeze(1) |
| elif audio_values.dim() == 1: |
| audio_values = audio_values.unsqueeze(0) |
| return audio_values |
|
|
|
|
| |
| |
| |
|
|
| 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) |
| |
| target_sr = self.audio_tokenizer.sample_rate |
| 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() |
| |
| |
| 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) |
| 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 |
|
|
