Upload whisper_asr.py

whisper_asr.py

@@ -0,0 +1,121 @@
+# whisper_asr.py
+import os, re, json, math, tempfile, traceback
+import numpy as np
+import pandas as pd
+import torch
+import soundfile as sf
+import textdistance
+
+# هذه المتغيرات مطلوبة في هذا الملف، وستُستورد إلى ملف المعالجة اللاحقة
+# لتجنب تكرار التعريف.
+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,
+)
+
+# =========================
+# Device
+# =========================
+DEVICE = "cuda" if torch.cuda.is_available() else "cpu"
+print(f"[INIT] DEVICE={DEVICE}", flush=True)
+
+# =========================
+# Lazy models (Whisper only)
+# =========================
+_WHISPER = None
+
+def load_whisper_model(
+    whisper_name=FORCE_WHISPER_NAME,
+    whisper_compute=FORCE_COMPUTE_TYPE,
+):
+    """Load Whisper model once; forced config respected even on CPU."""
+    global _WHISPER
+    from faster_whisper import WhisperModel
+
+    if _WHISPER is None:
+        _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)
+    return _WHISPER
+
+# =========================
+# Faster-Whisper helpers
+# =========================
+def normalize_ar_orth(text: str) -> str:
+    # تطبيع عام للمحاذاة
+    text = re.sub(r"[ًٌٍَُِّْـ]", "", text)
+    text = re.sub(r"[“”\"',:؛؟.!()\[\]{}،\-–—_]", " ", text)
+    text = re.sub(r"\s+", " ", text).strip()
+    return text
+
+def clean_ar_token(t: str) -> str:
+    t = t.strip()
+    t = re.sub(r'^[^\w\u0600-\u06FF]+|[^\w\u0600-\u06FF]+$', '', t)
+    t = normalize_ar_orth(t)
+    return t
+
+def extract_word_conf_table(segments):
+    rows = []
+    for seg in segments:
+        for w in (seg.words or []):
+            rows.append({
+                "seg_start": float(seg.start),
+                "seg_end": float(seg.end),
+                "word_start": float(w.start),
+                "word_end": float(w.end),
+                "word": clean_ar_token(w.word),
+                "prob": float(w.probability),
+            })
+    return pd.DataFrame(rows)
+
+def build_asr_token_conf(df_words: pd.DataFrame, hyp_tokens: list):
+    toks_probs, toks_durs = [], []
+    for _, row in df_words.iterrows():
+        prob = row["prob"]
+        dur  = (row["word_end"] - row["word_start"]) * 1000.0
+        toks_probs.append(prob)
+        toks_durs.append(dur)
+
+    L = len(hyp_tokens)
+    if len(toks_probs) >= L:
+        toks_probs = toks_probs[:L]
+        toks_durs  = toks_durs[:L]
+    else:
+        pad = L - len(toks_probs)
+        toks_probs += [None]*pad
+        toks_durs  += [None]*pad
+
+    arr = np.array([p for p in toks_probs if p is not None])
+    if arr.size:
+        low_t  = float(np.quantile(arr, 0.15))
+        high_t = float(np.quantile(arr, 0.70))
+    else:
+        low_t, high_t = 0.5, 0.85
+
+    asr_token_conf = {i: {"prob": toks_probs[i], "duration_ms": toks_durs[i]} for i in range(L)}
+    return asr_token_conf, low_t, high_t
+
+# =========================
+# Audio helper
+# =========================
+def ensure_audio_path(audio):
+    if isinstance(audio, str):
+        if not os.path.exists(audio):
+            raise FileNotFoundError(f"Audio path not found: {audio}")
+        return audio
+    if isinstance(audio, tuple) and len(audio) == 2:
+        data, sr = audio
+        if isinstance(data, np.ndarray):
+            tmp = tempfile.NamedTemporaryFile(suffix=".wav", delete=False)
+            sf.write(tmp.name, data, sr)
+            return tmp.name
+    raise ValueError("Unsupported audio input format")