import streamlit as st
import streamlit.components.v1 as components
import numpy as np
import pandas as pd
import pickle, os, re, time
import torch
import librosa
import faiss
import whisper
from sentence_transformers import SentenceTransformer
from sklearn.metrics.pairwise import cosine_similarity
from huggingface_hub import hf_hub_download
# ── Page config ──────────────────────────────────────────────
st.set_page_config(
page_title="HudaAI — Quranic Verse Recognition",
page_icon="🕌",
layout="wide"
)
# ── Constants ────────────────────────────────────────────────
HF_DATASET = "Ahmed062646/WhisperModel_Ai"
TARGET_SR = 16000
DEVICE = "cuda" if torch.cuda.is_available() else "cpu"
EMBED_NAME = "CAMeL-Lab/bert-base-arabic-camelbert-ca"
MIN_SCORE = 0.12
SURAH_AYAH_COUNTS = [
7,286,200,176,120,165,206,75,129,109,123,111,43,52,99,128,111,110,98,
135,112,78,118,64,77,227,93,88,69,60,34,30,73,54,45,83,182,88,75,85,
54,53,89,59,37,35,38,29,18,45,60,49,62,55,78,96,29,22,24,13,14,11,11,
18,12,12,30,52,52,44,28,28,20,56,40,31,50,40,46,42,29,19,36,25,22,17,
19,26,30,20,15,21,11,8,8,19,5,8,8,11,11,8,3,9,5,4,7,3,6,3,5,4,5,6
]
RECITERS = {
"Mishary Alafasy" : "ar.alafasy",
"Abdul Basit" : "ar.abdulbasitmurattal",
"Maher Al Muaiqly" : "ar.mahermuaiqly",
"Minshawi Murattal" : "ar.minshawimujawwad",
}
LANGUAGES = {
"English" : "en",
"Urdu" : "ur",
"Hindi" : "hi",
"French" : "fr",
"Spanish" : "es",
"Bengali" : "bn",
"Portuguese" : "pt",
"Russian" : "ru",
"Mandarin Chinese" : "zh",
"Arabic (Original)" : "ar",
}
RTL_LANGS = {"ur", "ar"}
# ── Helpers ──────────────────────────────────────────────────
def hf_download(filename):
return hf_hub_download(
repo_id=HF_DATASET, filename=filename, repo_type="dataset",
token=os.environ.get("HF_TOKEN")
)
def get_global_ayah_number(surah_id: int, ayah_id: int) -> int:
return sum(SURAH_AYAH_COUNTS[:surah_id - 1]) + ayah_id
def get_ayah_audio_url(surah_id: int, ayah_id: int, reciter_id: str) -> str:
n = get_global_ayah_number(surah_id, ayah_id)
return f"https://cdn.islamic.network/quran/audio/128/{reciter_id}/{n}.mp3"
# ── Load resources ───────────────────────────────────────────
@st.cache_resource(show_spinner=False)
def load_resources():
corpus_path = hf_download("quran_verses_multilingual.csv")
tfidf_vec_path = hf_download("tfidf_vectorizer.pkl")
tfidf_mat_path = hf_download("tfidf_matrix.pkl")
faiss_idx_path = hf_download("faiss_index.bin")
df = pd.read_csv(corpus_path)
with open(tfidf_vec_path, "rb") as f: tfidf_vec = pickle.load(f)
with open(tfidf_mat_path, "rb") as f: tfidf_mat = pickle.load(f)
fidx = faiss.read_index(faiss_idx_path)
embed = SentenceTransformer(EMBED_NAME, device=DEVICE)
wmodel = whisper.load_model("small", device=DEVICE)
return df, tfidf_vec, tfidf_mat, fidx, embed, wmodel
# ── Arabic normalization ─────────────────────────────────────
def normalize_arabic(text):
if not isinstance(text, str): return ""
text = re.sub(r"[\u064B-\u065F\u0670]", "", text)
text = re.sub(r"[أإآٱ]", "ا", text)
text = re.sub(r"ة", "ه", text)
text = re.sub(r"ى", "ي", text)
text = re.sub(r"ـ", "", text)
text = re.sub(r"[^\u0600-\u06FF\s]", "", text)
return re.sub(r"\s+", " ", text).strip()
# ── Hybrid search ────────────────────────────────────────────
def hybrid_search(query_norm, df, tfidf_vec, tfidf_mat, fidx, embed, top_k=3):
if not query_norm.strip(): return []
pool = top_k * 3
qv = tfidf_vec.transform([query_norm])
t_scores = cosine_similarity(qv, tfidf_mat).flatten()
t_top = t_scores.argsort()[::-1][:pool]
qe = embed.encode(
[query_norm], normalize_embeddings=True, convert_to_numpy=True
).astype("float32")
f_scores, f_idx = fidx.search(qe, pool)
score_map = {}
t_min, t_max = t_scores[t_top].min(), t_scores[t_top].max()
t_range = t_max - t_min if t_max != t_min else 1.0
for i in t_top:
k = (int(df.iloc[i]["surah_id"]), int(df.iloc[i]["ayah_id"]))
score_map[k] = {"t": (t_scores[i] - t_min) / t_range, "f": 0.0, "idx": i}
f_arr = f_scores[0]
f_min, f_max = f_arr.min(), f_arr.max()
f_range = f_max - f_min if f_max != f_min else 1.0
for idx, sc in zip(f_idx[0], f_arr):
if idx == -1: continue
k = (int(df.iloc[idx]["surah_id"]), int(df.iloc[idx]["ayah_id"]))
norm = (sc - f_min) / f_range
if k in score_map: score_map[k]["f"] = norm
else: score_map[k] = {"t": 0.0, "f": norm, "idx": idx}
combined = sorted(
[(0.4*v["t"] + 0.6*v["f"], k, v["idx"]) for k, v in score_map.items()],
reverse=True
)
results = []
for score, (sid, aid), idx in combined[:top_k]:
row = df.iloc[idx]
results.append({
"score" : round(float(score), 4),
"surah_id" : sid,
"ayah_id" : aid,
"surah_name_en" : str(row["surah_name_en"]),
"surah_name_ar" : str(row["surah_name_ar"]),
"ayah_ar" : str(row.get("ayah_ar", "")),
"ayah_tr" : str(row.get("ayah_tr", "")),
"ayah_en" : str(row.get("ayah_en", "")),
"ayah_ur" : str(row.get("ayah_ur", "")),
"ayah_hi" : str(row.get("ayah_hi", "")),
"ayah_fr" : str(row.get("ayah_fr", "")),
"ayah_es" : str(row.get("ayah_es", "")),
"ayah_bn" : str(row.get("ayah_bn", "")),
"ayah_pt" : str(row.get("ayah_pt", "")),
"ayah_ru" : str(row.get("ayah_ru", "")),
"ayah_zh" : str(row.get("ayah_zh", "")),
})
return results
# ── Global CSS ───────────────────────────────────────────────
st.markdown("""
""", unsafe_allow_html=True)
# ── Header ───────────────────────────────────────────────────
st.markdown("""
""", unsafe_allow_html=True)
# ── Load models ───────────────────────────────────────────────
with st.spinner("⏳ Loading AI models — please wait (~2 min first time)..."):
df, tfidf_vec, tfidf_mat, fidx, embed, wmodel = load_resources()
st.success("✅ Models ready!")
# ── Session state init ────────────────────────────────────────
if "rec_audio_bytes" not in st.session_state:
st.session_state["rec_audio_bytes"] = None
if "rec_ready" not in st.session_state:
st.session_state["rec_ready"] = False
# ── Sidebar ──────────────────────────────────────────────────
with st.sidebar:
st.markdown("### ⚙️ Settings")
lang_name = st.selectbox("🌐 Translation Language", list(LANGUAGES.keys()), index=0)
lang_code = LANGUAGES[lang_name]
reciter_name = st.selectbox("🎵 Verification Reciter", list(RECITERS.keys()), index=0)
reciter_id = RECITERS[reciter_name]
top_k = st.slider("Matches to show", 1, 5, 3)
min_score = st.slider("Min confidence threshold", 0.0, 0.5, 0.12, 0.01)
show_trans = st.checkbox("Show transliteration", value=True)
show_audio = st.checkbox("Show ayah audio player", value=True)
st.markdown("---")
st.markdown("### 📖 About")
st.markdown("""
HudaAI identifies Quranic verses using
**Whisper ASR** + **FAISS semantic search**.
**Course:** AI2002 · Artificial Intelligence
**University:** FAST-NUCES Lahore
**Semester:** Spring 2026
""")
# ── Input section ─────────────────────────────────────────────
col_l, col_r = st.columns([3, 2])
audio_bytes = None
input_source = None
with col_l:
st.markdown("### 🎙️ Input Recitation")
tab_up, tab_rec = st.tabs(["📁 Upload Audio", "🎙️ Record Audio"])
with tab_up:
uploaded = st.file_uploader(
"Upload Quranic recitation (MP3, WAV, M4A, OGG)",
type=["mp3", "wav", "m4a", "ogg"]
)
if uploaded:
audio_bytes = uploaded.read()
input_source = "upload"
st.audio(audio_bytes)
with tab_rec:
# ── Recorder: JS sends base64 WAV to a hidden text_area, Python reads it ──
# We write the b64 payload to a temp file path stored in session_state key
# and trigger a rerun via a checkbox toggle bridge.
# Hidden textarea bridge — JS will inject base64 audio here
b64_bridge = st.text_area(
"rec_bridge", value="", key="rec_b64_bridge",
label_visibility="collapsed", height=1,
help="internal"
)
# Hide the textarea with CSS
st.markdown("""
""", unsafe_allow_html=True)
# Process incoming base64 audio from JS
if b64_bridge and b64_bridge.startswith("data:audio"):
import base64 as _b64
header, b64data = b64_bridge.split(",", 1)
raw = _b64.b64decode(b64data)
st.session_state["rec_audio_bytes"] = raw
st.session_state["rec_ready"] = True
# Show recorder component
components.html("""
● Record
✅ Use This Recording
⏳ Saving to Streamlit...
This will save your recording instantly — no download needed.
""", height=310, scrolling=False)
# Show recorded audio if available in session state
if st.session_state.get("rec_ready") and st.session_state.get("rec_audio_bytes"):
st.success("✅ Recording saved! Click **Identify Ayah** below.")
st.audio(st.session_state["rec_audio_bytes"], format="audio/ogg")
if st.button("🗑️ Clear Recording", key="clear_rec"):
st.session_state["rec_ready"] = False
st.session_state["rec_audio_bytes"] = None
st.rerun()
with col_r:
st.markdown("### 🌐 Settings Summary")
st.info(
f"**Translation:** {lang_name} \n"
f"**Verification reciter:** {reciter_name}"
)
if audio_bytes:
st.success(
f"✅ Audio ready \n"
f"Source: {'Recording' if input_source == 'record' else 'Upload'}"
)
# ── Pull recorded audio from session_state if no upload ──────
if audio_bytes is None and st.session_state.get("rec_ready") and st.session_state.get("rec_audio_bytes"):
audio_bytes = st.session_state["rec_audio_bytes"]
input_source = "record"
# ── Identify button ───────────────────────────────────────────
if audio_bytes:
st.markdown("---")
if st.button("🔍 Identify Ayah", type="primary", use_container_width=True):
tmp = "/tmp/huda_in"
with open(tmp, "wb") as f:
f.write(audio_bytes)
with st.spinner("🔄 Preprocessing..."):
audio, _ = librosa.load(tmp, sr=TARGET_SR, mono=True)
audio, _ = librosa.effects.trim(audio, top_db=20)
audio = (audio / (np.abs(audio).max() + 1e-8) * 0.95).astype(np.float32)
with st.spinner("🎙️ Whisper transcribing..."):
t0 = time.time()
result = wmodel.transcribe(
audio, language="ar", task="transcribe",
fp16=(DEVICE == "cuda"),
condition_on_previous_text=False,
no_speech_threshold=0.4, verbose=False
)
whisper_ms = round((time.time() - t0) * 1000)
raw_text = result["text"].strip()
norm_text = normalize_arabic(raw_text)
no_sp_prob = result["segments"][0]["no_speech_prob"] if result.get("segments") else 1.0
detected_lg = result.get("language", "ar")
src_lbl = "🎙️ Recorded" if input_source == "record" else "📁 Uploaded"
st.markdown(f"""
🎙️ Whisper Output
{src_lbl}
Raw: {raw_text}
Normalized: {norm_text}
Language: {detected_lg} |
No-speech: {round(no_sp_prob,3)} |
Time: {whisper_ms}ms
""", unsafe_allow_html=True)
# Validation
invalid, reason = False, ""
if no_sp_prob > 0.60:
invalid = True
reason = f"Audio appears silent (no-speech prob: {round(no_sp_prob,2)})"
elif not norm_text:
invalid = True
reason = "No Arabic text could be transcribed"
elif detected_lg not in ("ar", "arabic"):
invalid = True
reason = f"Audio is not in Arabic (detected: {detected_lg})"
elif len(norm_text.split()) < 2:
invalid = True
reason = "Transcription too short to be an ayah"
if invalid:
st.markdown(f"""
⛔ Invalid Input
❌ Audio does not contain a Quranic ayah.
Reason: {reason}
""", unsafe_allow_html=True)
else:
with st.spinner("🔍 Searching Quran..."):
t1 = time.time()
matches = hybrid_search(norm_text, df, tfidf_vec, tfidf_mat, fidx, embed, top_k)
search_ms = round((time.time() - t1) * 1000)
if not matches or matches[0]["score"] < min_score:
best = matches[0]["score"] if matches else 0.0
st.markdown(f"""
⛔ No Quranic Ayah Found
❌ Audio does not appear to be a Quranic recitation.
Reason: Best score ({best:.4f}) below threshold ({min_score})
""", unsafe_allow_html=True)
else:
st.markdown(f"### 🎯 Results · **{lang_name}**")
for i, m in enumerate(matches):
is_top = (i == 0)
box_cls = "top-match" if is_top else "candidate"
badge = (
'✅ Top Match '
if is_top else
f'Candidate #{i+1} '
)
translation = m["ayah_ar"] if lang_code == "ar" else m.get(f"ayah_{lang_code}", "")
if not translation or translation == "nan":
translation = "[Translation not available]"
score_pct = min(int(m["score"] / 0.8 * 100), 100)
bar_color = "#2d6a4f" if is_top else "#adb5bd"
rtl_cls = "translation-rtl" if lang_code in RTL_LANGS else ""
tr_row = ""
if show_trans and m.get("ayah_tr", "") not in ("", "nan"):
tr_row = (
f''
f'📝 {m["ayah_tr"]}
'
)
st.markdown(f"""
{badge}
📖 Surah {m["surah_id"]}
{m["surah_name_en"]} — {m["surah_name_ar"]}
Ayah {m["ayah_id"]}
{m["ayah_ar"]}
🌐 {lang_name}: {translation}
{tr_row}
""", unsafe_allow_html=True)
# ── Ayah audio verification ───────────────────────────
if show_audio:
audio_url = get_ayah_audio_url(m["surah_id"], m["ayah_id"], reciter_id)
st.markdown(f"""
🔊 Verify — {reciter_name}
· Surah {m["surah_id"]}
· Ayah {m["ayah_id"]}
""", unsafe_allow_html=True)
# Fetch audio bytes server-side so Streamlit serves reliably
try:
import urllib.request
with urllib.request.urlopen(audio_url, timeout=10) as resp:
ayah_audio_bytes = resp.read()
st.audio(ayah_audio_bytes, format="audio/mp3")
except Exception:
st.warning(
f"⚠️ Could not load audio automatically. "
f"[▶ Listen directly]({audio_url})"
)
st.markdown(
f'⏱ Whisper: {whisper_ms}ms | '
f'Search: {search_ms}ms | '
f'Total: {whisper_ms+search_ms}ms
',
unsafe_allow_html=True
)
# ── Footer ───────────────────────────────────────────────────
st.markdown("---")
st.markdown(
""
"HudaAI · AI2002 Artificial Intelligence · "
"FAST-NUCES Lahore · Spring 2026"
"
",
unsafe_allow_html=True
)