Update app.py

app.py

@@ -12,15 +12,28 @@ from transformers import AutoTokenizer, AutoModel
 import soundfile as sf
 
 # =========================
-# Device & global config
+# Global config (forced per your request)
 # =========================
-DEVICE = "cuda" if torch.cuda.is_available() else "cpu"
-CPU_MODE = (DEVICE != "cuda")
+# نثبّت الإعدادات المطلوبة على CPU
+FORCE_WHISPER_NAME = "large-v3"
+FORCE_COMPUTE_TYPE = "int8"
+FORCE_USE_MARBERT = True
+
+# خيارات تفريغ ثابتة لتقليل الفروقات مع النوتبوك
+ASR_OPTS = dict(
+    word_timestamps=True,
+    vad_filter=True,
+    vad_parameters={"min_silence_duration_ms": 200},
+    beam_size=5,
+    best_of=5,
+    temperature=0.0,          # جعل فك التشفير حتمي قدر الإمكان
+)
 
-# أمان الذاكرة على CPU
-DEFAULT_WHISPER_CPU = "large-v3"
-DEFAULT_COMPUTE_CPU = "int8"
-DEFAULT_USE_MARBERT_CPU = True
+# =========================
+# Device
+# =========================
+DEVICE = "cuda" if torch.cuda.is_available() else "cpu"
+print(f"[INIT] DEVICE={DEVICE}", flush=True)
 
 # =========================
 # Lazy models
@@ -33,28 +46,28 @@ _WHISPER = None
 def load_models(
     sbert_name="sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2",
     marbert_name="UBC-NLP/MARBERT",
-    whisper_name="small",
-    whisper_compute="int8"
+    whisper_name=FORCE_WHISPER_NAME,
+    whisper_compute=FORCE_COMPUTE_TYPE,
+    use_marbert=FORCE_USE_MARBERT
 ):
-    """Load models only once."""
+    """Load models once; forced config respected even on CPU."""
     global _SBERT, _MARBERT_TOK, _MARBERT, _WHISPER
 
-    # حماية على CPU: اجبار نماذج أخف
-    if CPU_MODE:
-        whisper_name = DEFAULT_WHISPER_CPU
-        whisper_compute = DEFAULT_COMPUTE_CPU
-
     if _SBERT is None:
-        _SBERT = SentenceTransformer(sbert_name, device=DEVICE)
+        _SBERT = SentenceTransformer(sbert_name, device=("cuda" if DEVICE=="cuda" else "cpu"))
+        print(f"[LOAD] SBERT: {sbert_name}", flush=True)
 
-    # حمّل MARBERT فقط عند الحاجة (قد يستهلك RAM)
-    if _MARBERT is None and (not CPU_MODE):
+    # مفعّل على CPU حسب رغبتك
+    if _MARBERT is None and use_marbert:
         _MARBERT_TOK = AutoTokenizer.from_pretrained(marbert_name)
-        _MARBERT = AutoModel.from_pretrained(marbert_name).to(DEVICE)
+        _MARBERT = AutoModel.from_pretrained(marbert_name).to(("cuda" if DEVICE=="cuda" else "cpu"))
         _MARBERT.eval()
+        print(f"[LOAD] MARBERT: {marbert_name} (device={DEVICE})", flush=True)
 
     if _WHISPER is None:
-        _WHISPER = WhisperModel(whisper_name, device=DEVICE, compute_type=whisper_compute)
+        _WHISPER = WhisperModel(whisper_name, device=("cuda" if DEVICE=="cuda" else "cpu"),
+                                 compute_type=whisper_compute)
+        print(f"[LOAD] Whisper: {whisper_name} (compute={whisper_compute})", flush=True)
 
 # =========================
 # Normalization / Tokenization / Alignment
@@ -68,7 +81,6 @@ def normalize_ar_orth(text: str) -> str:
     return text
 
 def simple_tokenize(text: str):
-    """يحاول punkt؛ وإن فشل يستخدم تجزئة بسيطة بالمسافات."""
     t = normalize_ar_orth(text)
     try:
         import nltk
@@ -108,7 +120,8 @@ def arabic_soundex(word):
     for ch in w:
         for rep, chars in groups.items():
             if ch in chars:
-                code.append(rep); break
+                code.append(rep)
+                break
     return "".join(code)
 
 def phonetic_similarity(w1, w2):
@@ -168,8 +181,8 @@ def marbert_cls_similarity(a: str, b: str) -> float:
     if _MARBERT is None:
         return 0.0
     with torch.no_grad():
-        ta = _MARBERT_TOK(a, return_tensors='pt', truncation=True, padding=True).to(DEVICE)
-        tb = _MARBERT_TOK(b, return_tensors='pt', truncation=True, padding=True).to(DEVICE)
+        ta = _MARBERT_TOK(a, return_tensors='pt', truncation=True, padding=True).to(("cuda" if DEVICE=="cuda" else "cpu"))
+        tb = _MARBERT_TOK(b, return_tensors='pt', truncation=True, padding=True).to(("cuda" if DEVICE=="cuda" else "cpu"))
         ea = _MARBERT(**ta).last_hidden_state[:,0,:]
         eb = _MARBERT(**tb).last_hidden_state[:,0,:]
         sim = util.cos_sim(ea, eb).item()
@@ -390,7 +403,7 @@ def literal_similarity(original, recited):
     return {"levenshtein": round(lev,3), "word_overlap": round(word_overlap,3),
             "bleu1": round(bleu1,3), "literal_score": round(final_score,3)}
 
-def semantic_similarity(original, recited, use_marbert=True):
+def semantic_similarity(original, recited, use_marbert=FORCE_USE_MARBERT):
     sbert_sim = float(util.pytorch_cos_sim(_SBERT.encode(original, convert_to_tensor=True),
                                            _SBERT.encode(recited,  convert_to_tensor=True)))
     marbert_sim = marbert_cls_similarity(original, recited) if use_marbert else 0.0
@@ -398,10 +411,9 @@ def semantic_similarity(original, recited, use_marbert=True):
             "semantic_score": round(max(sbert_sim, marbert_sim),3)}
 
 # =========================
-# Audio input helper
+# Audio helper
 # =========================
 def ensure_audio_path(audio):
-    """Accepts filepath (str) OR (numpy_array, sr). Returns a valid filepath."""
     if isinstance(audio, str):
         if not os.path.exists(audio):
             raise FileNotFoundError(f"Audio path not found: {audio}")
@@ -415,7 +427,7 @@ def ensure_audio_path(audio):
     raise ValueError("Unsupported audio input format")
 
 # =========================
-# Pipeline (with robust error reporting)
+# Pipeline (robust errors + logs)
 # =========================
 def transcribe_and_evaluate(audio, original_text, whisper_size=None,
                             compute_type=None, vad=True, use_marbert=True):
@@ -423,29 +435,29 @@ def transcribe_and_evaluate(audio, original_text, whisper_size=None,
         if not original_text or not original_text.strip():
             raise ValueError("Original text is empty.")
 
-        # Defaults per device
-        if CPU_MODE:
-            whisper_size = DEFAULT_WHISPER_CPU
-            compute_type = DEFAULT_COMPUTE_CPU
-            use_marbert = DEFAULT_USE_MARBERT_CPU
-        else:
-            whisper_size = whisper_size or "large-v3"
-            compute_type = compute_type or "float16"
+        # نُهمل اختيارات الواجهة ونفرض إعداداتك
+        whisper_size = FORCE_WHISPER_NAME
+        compute_type = FORCE_COMPUTE_TYPE
+        use_marbert = FORCE_USE_MARBERT
+
+        print(f"[RUN] whisper={whisper_size}, compute={compute_type}, marbert={use_marbert}", flush=True)
 
-        load_models(whisper_name=whisper_size, whisper_compute=compute_type)
+        load_models(whisper_name=whisper_size, whisper_compute=compute_type, use_marbert=use_marbert)
 
         audio_path = ensure_audio_path(audio)
-        segments, info = _WHISPER.transcribe(
-            audio_path, word_timestamps=True,
-            vad_filter=vad, vad_parameters={"min_silence_duration_ms": 200}
-        )
+        print(f"[AUDIO] path={audio_path}", flush=True)
+
+        segments, info = _WHISPER.transcribe(audio_path, **ASR_OPTS)
         segments = list(segments)
+        print(f"[ASR] segments={len(segments)}", flush=True)
 
+        # Build ASR text from words (more control)
         words = []
         for seg in segments:
             for w in (seg.words or []):
                 tok = clean_ar_token(w.word)
-                if tok: words.append(tok)
+                if tok:
+                    words.append(tok)
         asr_text = " ".join(words)
 
         ref_tokens = simple_tokenize(original_text)
@@ -454,6 +466,7 @@ def transcribe_and_evaluate(audio, original_text, whisper_size=None,
 
         df_words = extract_word_conf_table(segments)
         asr_token_conf, low_t, high_t = build_asr_token_conf(df_words, hyp_tokens)
+        print(f"[CONF] low_t={low_t:.3f}, high_t={high_t:.3f}", flush=True)
 
         results, corrected_text = classify_alignment_optimized(
             aligned, ref_tokens, hyp_tokens,
@@ -462,33 +475,31 @@ def transcribe_and_evaluate(audio, original_text, whisper_size=None,
         )
 
         lit = literal_similarity(original_text, corrected_text)
-        sem = semantic_similarity(original_text, corrected_text, use_marbert=(use_marbert and not CPU_MODE))
+        sem = semantic_similarity(original_text, corrected_text, use_marbert=use_marbert)
 
         df = pd.DataFrame(results)
 
         report = {
-            "whisper_model": whisper_size,
-            "compute_type": compute_type,
+            "requested": {"whisper_model": whisper_size, "compute_type": compute_type, "use_marbert": use_marbert},
+            "effective": {"whisper_model": whisper_size, "compute_type": compute_type, "use_marbert": use_marbert},
             "original_text": original_text,
             "asr_text": asr_text,
             "corrected_text": corrected_text,
             "literal": lit,
             "semantic": sem,
-            "low_t": low_t, "high_t": high_t,
+            "low_t": float(low_t), "high_t": float(high_t),
         }
         return corrected_text, asr_text, json.dumps(report, ensure_ascii=False, indent=2), df
 
     except Exception as e:
         tb = traceback.format_exc()
         print("ERROR in transcribe_and_evaluate:\n", tb, flush=True)
-        # نرجع JSON بالخطأ بدل ما نفجّر الواجهة
         empty_df = pd.DataFrame([{"ASR_word":"","GT_word":"","status":"ERROR","reason":str(e),"used":""}])
         err_json = json.dumps({"error": str(e), "traceback": tb}, ensure_ascii=False, indent=2)
         gr.Warning(str(e))
         return "", "", err_json, empty_df
 
 def api_predict(audio, original_text, whisper_size=None, compute_type=None, vad=True, use_marbert=True):
-    # نفس الدالة لكن ترجع JSON فقط
     corrected_text, asr_text, report_json, df = transcribe_and_evaluate(
         audio, original_text, whisper_size, compute_type, vad, use_marbert
     )
@@ -505,23 +516,15 @@ def build_ui():
         gr.Markdown("## Samaali — ASR Post-Processing (Whisper + Alignment + Confidence + Semantics)")
 
         with gr.Row():
-            # filepath أسلم للـ Spaces
             audio = gr.Audio(sources=["microphone","upload"], type="filepath", label="Audio")
             original = gr.Textbox(lines=8, label="Original Text (Ground Truth)")
 
         with gr.Row():
-            whisper_size = gr.Dropdown(
-                choices=["tiny","base","small","medium","large-v3"],
-                value=("large-v3" if not CPU_MODE else DEFAULT_WHISPER_CPU),
-                label="Whisper model size"
-            )
-            compute_type = gr.Dropdown(
-                choices=["int8", "int8_float16", "float16", "float32"],
-                value=("float16" if not CPU_MODE else DEFAULT_COMPUTE_CPU),
-                label="compute_type"
-            )
+            # واجهة ثابتة حسب طلبك (تُهمل في الدالة لكن نعرضها)
+            whisper_size = gr.Dropdown(choices=["large-v3"], value="large-v3", label="Whisper model size (forced)")
+            compute_type = gr.Dropdown(choices=["int8"], value="int8", label="compute_type (forced)")
             vad = gr.Checkbox(value=True, label="VAD filter")
-            use_marbert = gr.Checkbox(value=(not CPU_MODE), label="Use MARBERT (semantic)")
+            use_marbert = gr.Checkbox(value=True, label="Use MARBERT (forced)")
 
         btn = gr.Button("Transcribe & Evaluate", variant="primary")