Update app.py

app.py

@@ -1,24 +1,79 @@
-from fastapi import FastAPI, HTTPException
-from pydantic import BaseModel
-from transformers import AutoTokenizer, VitsModel
+from logging import error
+import gradio as gr
+import spaces
 import torch
-import numpy as np
+from transformers import AutoTokenizer, VitsModel
 import os
+import numpy as np
 import noisereduce as nr
+import torch.nn as nn
+from typing import Optional, Iterator
 
 # قراءة التوكن من Secrets
-token = os.getenv("acees-token")
+token = os.getenv("acees-token")  # تأكد أنك سميته بنفس الاسم في Settings → Repository secrets
 
-# تخزين النماذج
+# كائن لتخزين النماذج
 models = {}
 
-# اختيار الجهاز
+# اختيار الجهاز (CUDA لو متوفر، غير كذا CPU)
 device = "cuda" if torch.cuda.is_available() else "cpu"
 
+
 # دالة إزالة الضوضاء
 def remove_noise_nr(audio_data, sr=16000):
     return nr.reduce_noise(y=audio_data, hop_length=256, sr=sr)
 
+
+# دالة inference (streaming / non-streaming)
+def _inference_forward_stream(
+        self,
+        input_ids: Optional[torch.Tensor] = None,
+        attention_mask: Optional[torch.Tensor] = None,
+        speaker_embeddings: Optional[torch.Tensor] = None,
+        chunk_size: int = 32,
+        is_streaming: bool = True
+    ) -> Iterator[torch.Tensor]:
+
+    padding_mask = attention_mask.unsqueeze(-1).float() if attention_mask is not None else torch.ones_like(input_ids).unsqueeze(-1).float()
+    text_encoder_output = self.text_encoder(input_ids=input_ids, padding_mask=padding_mask, attention_mask=attention_mask)
+    hidden_states = text_encoder_output[0].transpose(1, 2)
+    input_padding_mask = padding_mask.transpose(1, 2)
+
+    log_duration = self.duration_predictor(hidden_states, input_padding_mask, speaker_embeddings)
+    length_scale = 1.0 / self.speaking_rate
+    duration = torch.ceil(torch.exp(log_duration) * input_padding_mask * length_scale)
+    predicted_lengths = torch.clamp_min(torch.sum(duration, [1,2]), 1).long()
+
+    indices = torch.arange(predicted_lengths.max(), device=predicted_lengths.device)
+    output_padding_mask = indices.unsqueeze(0) < predicted_lengths.unsqueeze(1)
+    output_padding_mask = output_padding_mask.unsqueeze(1).to(input_padding_mask.dtype)
+
+    attn_mask = torch.unsqueeze(input_padding_mask, 2) * torch.unsqueeze(output_padding_mask, -1)
+    batch_size, _, output_length, input_length = attn_mask.shape
+    cum_duration = torch.cumsum(duration, -1).view(batch_size * input_length, 1)
+    indices = torch.arange(output_length, device=duration.device)
+    valid_indices = indices.unsqueeze(0) < cum_duration
+    valid_indices = valid_indices.to(attn_mask.dtype).view(batch_size, input_length, output_length)
+    padded_indices = valid_indices - nn.functional.pad(valid_indices, [0,0,1,0,0,0])[:, :-1]
+    attn = padded_indices.unsqueeze(1).transpose(2,3) * attn_mask
+
+    prior_means = text_encoder_output[1]
+    prior_log_variances = text_encoder_output[2]
+    prior_latents = prior_means + torch.randn_like(prior_means) * torch.exp(prior_log_variances) * self.noise_scale
+    latents = self.flow(prior_latents, output_padding_mask, speaker_embeddings, reverse=True)
+    spectrogram = latents * output_padding_mask
+
+    if is_streaming:
+        for i in range(0, spectrogram.size(-1), chunk_size):
+            with torch.no_grad():
+                wav = self.decoder(spectrogram[:,:,i:i+chunk_size], speaker_embeddings)
+            yield wav.squeeze().cpu().numpy()
+    else:
+        with torch.no_grad():
+            wav = self.decoder(spectrogram, speaker_embeddings)
+        yield wav.squeeze().cpu().numpy()
+
+
 # تحميل النموذج + التوكن
 def get_model(name_model):
     global models
@@ -35,35 +90,42 @@ def get_model(name_model):
     tokenizer = AutoTokenizer.from_pretrained(name_model, token=token)
     return models[name_model], tokenizer
 
-# نموذج البيانات للـ POST
-class TTSRequest(BaseModel):
-    text: str
-    name_model: str = "wasmdashai/vits-ar-sa-huba-v2"
-    speaking_rate: int = 16000
-
-# إنشاء التطبيق
-app = FastAPI(title="VITS TTS API")
-
-# مسار صحة الخدمة
-@app.get("/")
-def home():
-    return {"message": "FastAPI VITS TTS service is running"}
-
-# مسار تحويل النص إلى كلام
-@app.post("/predict/")
-def modelspeech(req: TTSRequest):
-    try:
-        model, tokenizer = get_model(req.name_model)
-        inputs = tokenizer(req.text, return_tensors="pt").to(device)
-        model.speaking_rate = req.speaking_rate
-        with torch.no_grad():
-            outputs = model(**inputs)
-            waveform = outputs.waveform[0].cpu().numpy()
-        audio = remove_noise_nr(waveform)
-        return {
-            "sampling_rate": model.config.sampling_rate,
-            "audio": audio.tolist()  # تحويل numpy array إلى قائمة
-        }
-    except Exception as e:
-        raise HTTPException(status_code=500, detail=str(e))
+
+# النص الافتراضي
+TXT = "السلا�� عليكم ورحمة الله وبركاته يا هلا وسهلا ومراحب بالغالي"
+
+
+# دالة تحويل النص إلى كلام
+def modelspeech(text=TXT, name_model="wasmdashai/vits-ar-sa-huba-v2", speaking_rate=16000):
+    model, tokenizer = get_model(name_model)
+    inputs = tokenizer(text, return_tensors="pt").to(device)  # يشتغل على CPU أو GPU حسب المتوفر
+    model.speaking_rate = speaking_rate
+    with torch.no_grad():
+        outputs = model(**inputs)
+        waveform = outputs.waveform[0].cpu().numpy()
+    return model.config.sampling_rate, remove_noise_nr(waveform)
+
+
+# واجهة Gradio
+model_choices = gr.Dropdown(
+    choices=[
+        "wasmdashai/vits-ar-sa-huba-v1",
+        "wasmdashai/vits-ar-sa-huba-v2",
+        "wasmdashai/vits-ar-sa-A",
+        "wasmdashai/vits-ar-ye-sa",
+        "wasmdashai/vits-ar-sa-M-v1",
+        "wasmdashai/vits-en-v1"
+    ],
+    label="اختر النموذج",
+    value="wasmdashai/vits-ar-sa-huba-v2"
+)
+
+demo = gr.Interface(
+    fn=modelspeech,
+    inputs=["text", model_choices, gr.Slider(0.1, 1, step=0.1, value=0.8)],
+    outputs=["audio"]
+)
+
+demo.queue()
+demo.launch(server_name="0.0.0.0", server_port=7860)