Update app.py

app.py

@@ -1,7 +1,130 @@
-from fastapi import FastAPI
+from logging import error
+import gradio as gr
+import spaces
+import torch
+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
 
-app = FastAPI()
+# قراءة التوكن من Secrets
+token = os.getenv("acees-token")  # تأكد أنك سميته بنفس الاسم في Settings → Repository secrets
 
-@app.get("/")
-def greet_json():
-    return {"Hello": "World!"}
+# كائن لتخزين النماذج
+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
+    if name_model in models:
+        tokenizer = AutoTokenizer.from_pretrained(name_model, token=token)
+        return models[name_model], tokenizer
+
+    models[name_model] = VitsModel.from_pretrained(name_model, token=token)
+    models[name_model].decoder.apply_weight_norm()
+    for flow in models[name_model].flow.flows:
+        torch.nn.utils.weight_norm(flow.conv_pre)
+        torch.nn.utils.weight_norm(flow.conv_post)
+
+    tokenizer = AutoTokenizer.from_pretrained(name_model, token=token)
+    return models[name_model], tokenizer
+
+
+# النص الافتراضي
+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)