Upload tokenizer.py with huggingface_hub

tokenizer.py

@@ -0,0 +1,201 @@
+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