File size: 8,915 Bytes
989436d | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 | 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
|