Deploy Quran reciter Gradio app

.gitignore

@@ -0,0 +1,12 @@
+__pycache__/
+*.py[cod]
+*.egg-info/
+
+.venv/
+data/
+runs/
+pretrained_models/
+test/
+
+.env
+.DS_Store

README.md

@@ -0,0 +1,124 @@
+---
+title: Quran Reciter Identification
+emoji: 🎙️
+colorFrom: green
+colorTo: blue
+sdk: gradio
+sdk_version: 5.0.0
+app_file: app.py
+python_version: "3.11"
+suggested_hardware: cpu-basic
+---
+
+# Quran Reciter Identification
+
+Identify Quran reciters from short unknown audio clips using a strong pretrained
+speaker-recognition backbone.
+
+## Approach
+
+This project uses `speechbrain/spkrec-ecapa-voxceleb` as the default embedding
+model. ECAPA-TDNN is a proven speaker-recognition architecture, and the
+SpeechBrain checkpoint is trained on VoxCeleb1+VoxCeleb2. The local classifier
+is trained on Quran reciter embeddings, so training is fast enough for a
+moderate laptop dataset while still using a deep speaker model.
+
+Recommended dataset:
+
+- `Buraaq/quran-md-ayahs` on Hugging Face: 187,080 verse-level clips, 30
+  reciters, complete Quran coverage, about 450+ hours.
+
+The code also supports any folder dataset shaped like:
+
+```text
+data/raw/
+  alafasy/
+    clip001.mp3
+    clip002.wav
+  abdul_basit/
+    clip001.mp3
+```
+
+## Setup
+
+```powershell
+python -m venv .venv
+.\.venv\Scripts\Activate.ps1
+pip install -r requirements.txt
+pip install -e .
+```
+
+Install FFmpeg if your audio files are MP3 and `torchaudio` cannot decode them
+on your machine.
+
+## Build a Moderate Quran-MD Dataset
+
+This exports a balanced subset from Hugging Face into local WAV clips and a
+manifest. The default keeps all 30 reciters and about 500 ayahs per reciter,
+spread across train/validation/test surah ranges.
+
+```powershell
+python -m quran_reciter_id.export_quran_md `
+  --output-dir data/quran_md `
+  --max-samples-per-reciter 500
+```
+
+For a larger run, increase `--max-samples-per-reciter`; use `0` for the full
+dataset.
+
+## Build a Manifest from Local Folders
+
+```powershell
+python -m quran_reciter_id.build_manifest `
+  --audio-root data/raw `
+  --output data/manifests/local.jsonl
+```
+
+## Train
+
+```powershell
+python -m quran_reciter_id.train `
+  --manifest data/quran_md/manifest.jsonl `
+  --output-dir runs/ecapa_quran_md
+```
+
+Training saves:
+
+- `model.pt`: classifier head and normalization stats
+- `labels.json`: reciter labels
+- `metrics.json`: validation/test metrics
+- `embeddings/*.npz`: cached ECAPA embeddings
+
+## Predict an Unknown Reciter
+
+```powershell
+python -m quran_reciter_id.predict `
+  --run-dir runs/ecapa_quran_md `
+  --audio-file path\to\unknown_recitation.mp3 `
+  --top-k 5
+```
+
+The predictor returns top candidates, probabilities, and a centroid similarity
+score. If confidence is below `--unknown-threshold`, it reports the clip as
+unknown/out-of-distribution.
+
+## Test in the Browser
+
+Launch the Gradio interface to upload audio or record from a microphone:
+
+```powershell
+python gradio_app.py
+```
+
+Choose the checkpoint directory and device in the interface. Browser microphone
+access requires localhost or an HTTPS connection.
+
+## Notes
+
+- Split by surah when possible. That prevents the model from memorizing a
+  specific verse recording pattern instead of the reciter voice.
+- A 30-reciter Quran-MD subset is the best practical starting point today. If a
+  directly downloadable Tadabur release becomes available, use
+  `build_manifest.py` after arranging audio by reciter folder.
+- For production, use longer clips when possible. A 10-30 second recitation
+  generally gives more stable speaker embeddings than a very short ayah.

app.py

@@ -0,0 +1,15 @@
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+
+
+# Hugging Face Spaces runs the repository directly rather than installing the
+# src-layout package first.
+sys.path.insert(0, str(Path(__file__).resolve().parent / "src"))
+
+from gradio_app import demo  # noqa: E402
+
+
+if __name__ == "__main__":
+    demo.launch()

configs/ecapa_quran_md.yaml

@@ -0,0 +1,22 @@
+seed: 1337
+sample_rate: 16000
+embedding_model: speechbrain/spkrec-ecapa-voxceleb
+device: auto
+
+dataset:
+  min_duration_sec: 1.0
+  max_duration_sec: 45.0
+  split_strategy: per_reciter_order
+
+classifier:
+  hidden_dim: 256
+  dropout: 0.2
+  batch_size: 128
+  epochs: 80
+  learning_rate: 0.001
+  weight_decay: 0.0001
+  patience: 10
+
+inference:
+  unknown_threshold: 0.45
+  min_centroid_similarity: 0.25

gradio_app.py

@@ -0,0 +1,106 @@
+from __future__ import annotations
+
+import os
+from pathlib import Path
+
+import gradio as gr
+
+from quran_reciter_id.predict import predict
+
+
+def _default_checkpoint_dir() -> str:
+    configured = os.getenv("CHECKPOINT_DIR")
+    if configured:
+        return configured
+    if Path("checkpoint/model.pt").is_file():
+        return "checkpoint"
+    local_runs = sorted(Path("runs").glob("*/model.pt"))
+    return str(local_runs[-1].parent) if local_runs else "checkpoint"
+
+
+def identify_reciter(
+    audio_file: str | None,
+    video_file: str | None,
+    run_dir_text: str,
+    device: str,
+    top_k: int,
+) -> tuple[str, list[list[object]]]:
+    media_file = audio_file or video_file
+    if not media_file:
+        raise gr.Error("Record audio, upload audio, or upload an MP4 video first.")
+
+    run_dir = Path(run_dir_text).expanduser()
+    checkpoint_path = run_dir / "model.pt"
+    if not checkpoint_path.is_file():
+        raise gr.Error(f"Checkpoint not found: {checkpoint_path}")
+
+    try:
+        result = predict(
+            run_dir=run_dir,
+            audio_file=Path(media_file),
+            top_k=int(top_k),
+            device=device.strip() or "auto",
+        )
+    except Exception as exc:
+        raise gr.Error(f"Prediction failed: {exc}") from exc
+
+    if result["is_unknown"]:
+        verdict = "### Unknown speaker\nNo known reciter passed the acceptance thresholds."
+    else:
+        verdict = f"### Prediction: `{result['prediction']}`"
+
+    candidates = [
+        [
+            candidate["reciter_id"],
+            round(candidate["probability"] * 100, 2),
+            round(candidate["centroid_similarity"], 4),
+        ]
+        for candidate in result["top_candidates"]
+    ]
+    return verdict, candidates
+
+
+with gr.Blocks(title="Quran Reciter ID") as demo:
+    gr.Markdown(
+        "# Quran Reciter Identification\n"
+        "Record a recitation or upload audio to test a trained checkpoint."
+    )
+
+    with gr.Row():
+        with gr.Column(scale=2):
+            audio = gr.Audio(
+                label="Audio",
+                sources=["microphone", "upload"],
+                type="filepath",
+            )
+            video = gr.Video(
+                label="MP4 video (the audio track will be analyzed)",
+                sources=["upload"],
+                format="mp4",
+            )
+        with gr.Column(scale=1):
+            run_dir = gr.Textbox(
+                label="Checkpoint directory",
+                value=_default_checkpoint_dir(),
+            )
+            device = gr.Textbox(label="Device", value="auto")
+            top_k = gr.Slider(1, 10, value=5, step=1, label="Top candidates")
+            identify = gr.Button("Identify reciter", variant="primary")
+
+    verdict = gr.Markdown()
+    candidates = gr.Dataframe(
+        headers=["Reciter", "Probability (%)", "Centroid similarity"],
+        datatype=["str", "number", "number"],
+        interactive=False,
+        label="Top candidates",
+    )
+
+    identify.click(
+        fn=identify_reciter,
+        inputs=[audio, video, run_dir, device, top_k],
+        outputs=[verdict, candidates],
+    )
+
+
+if __name__ == "__main__":
+    demo.launch(server_name="0.0.0.0")

pyproject.toml

@@ -0,0 +1,25 @@
+[build-system]
+requires = ["setuptools>=68", "wheel"]
+build-backend = "setuptools.build_meta"
+
+[project]
+name = "quran-reciter-id"
+version = "0.1.0"
+description = "Quran reciter identification with deep speaker embeddings."
+requires-python = ">=3.10"
+dependencies = [
+  "datasets>=2.20.0",
+  "huggingface-hub>=0.24.0",
+  "gradio>=5.0.0",
+  "numpy>=1.24",
+  "PyYAML>=6.0",
+  "scikit-learn>=1.4",
+  "soundfile>=0.12",
+  "speechbrain>=1.0.0",
+  "torch>=2.2",
+  "torchaudio>=2.2",
+  "tqdm>=4.66",
+]
+
+[tool.setuptools.packages.find]
+where = ["src"]

requirements.txt

@@ -0,0 +1,11 @@
+datasets>=2.20.0
+huggingface-hub>=0.24.0
+gradio>=5.0.0
+numpy>=1.24
+PyYAML>=6.0
+scikit-learn>=1.4
+soundfile>=0.12
+speechbrain>=1.0.0
+torch>=2.2
+torchaudio>=2.2
+tqdm>=4.66

src/quran_reciter_id/__init__.py

@@ -0,0 +1,3 @@
+"""Quran reciter identification package."""
+
+__version__ = "0.1.0"

src/quran_reciter_id/audio.py

@@ -0,0 +1,89 @@
+from __future__ import annotations
+
+import subprocess
+from pathlib import Path
+
+import torch
+import torchaudio
+
+
+def _load_audio_with_ffmpeg(
+    path: str | Path,
+    sample_rate: int,
+    max_duration_sec: float | None,
+) -> torch.Tensor:
+    command = [
+        "ffmpeg",
+        "-v",
+        "error",
+        "-i",
+        str(path),
+        "-vn",
+    ]
+    if max_duration_sec is not None:
+        command.extend(["-t", str(max_duration_sec)])
+    command.extend(
+        ["-ac", "1", "-ar", str(sample_rate), "-f", "f32le", "pipe:1"]
+    )
+    completed = subprocess.run(command, capture_output=True, check=False)
+    if completed.returncode != 0 or not completed.stdout:
+        error = completed.stderr.decode("utf-8", errors="replace").strip()
+        raise RuntimeError(f"FFmpeg could not decode {path}: {error}")
+
+    waveform = torch.frombuffer(bytearray(completed.stdout), dtype=torch.float32)
+    return waveform.unsqueeze(0).clamp(-1.0, 1.0)
+
+
+def audio_duration_sec(path: str | Path) -> float | None:
+    try:
+        info = torchaudio.info(str(path))
+    except Exception:
+        return None
+    if info.sample_rate <= 0:
+        return None
+    return float(info.num_frames) / float(info.sample_rate)
+
+
+def load_audio(
+    path: str | Path,
+    sample_rate: int = 16000,
+    max_duration_sec: float | None = None,
+) -> torch.Tensor:
+    try:
+        waveform, source_rate = torchaudio.load(str(path), backend="ffmpeg")
+    except Exception:
+        # The TorchAudio wheel may not include its optional FFmpeg backend even
+        # when the system ffmpeg executable supports M4A/MP4. Decode through the
+        # executable as a reliable fallback.
+        return _load_audio_with_ffmpeg(path, sample_rate, max_duration_sec)
+    if waveform.ndim == 2 and waveform.shape[0] > 1:
+        waveform = waveform.mean(dim=0, keepdim=True)
+    elif waveform.ndim == 1:
+        waveform = waveform.unsqueeze(0)
+
+    if source_rate != sample_rate:
+        waveform = torchaudio.functional.resample(waveform, source_rate, sample_rate)
+
+    if max_duration_sec is not None:
+        max_frames = int(sample_rate * max_duration_sec)
+        if waveform.shape[-1] > max_frames:
+            waveform = waveform[..., :max_frames]
+
+    waveform = waveform.clamp(-1.0, 1.0)
+    return waveform
+
+
+def normalize_array_audio(array, source_rate: int, target_rate: int = 16000) -> torch.Tensor:
+    waveform = torch.as_tensor(array).float()
+    if waveform.ndim == 1:
+        waveform = waveform.unsqueeze(0)
+    elif waveform.ndim == 2:
+        if waveform.shape[0] > waveform.shape[1]:
+            waveform = waveform.transpose(0, 1)
+        if waveform.shape[0] > 1:
+            waveform = waveform.mean(dim=0, keepdim=True)
+
+    if source_rate != target_rate:
+        waveform = torchaudio.functional.resample(waveform, source_rate, target_rate)
+
+    return waveform.clamp(-1.0, 1.0)

src/quran_reciter_id/build_manifest.py

@@ -0,0 +1,136 @@
+from __future__ import annotations
+
+import argparse
+import json
+from collections import defaultdict
+from math import ceil
+from pathlib import Path
+
+from tqdm import tqdm
+
+from .audio import audio_duration_sec
+from .config import DEFAULT_CONFIG_PATH, load_config
+from .manifest import iter_audio_files, summarize_manifest, write_jsonl
+
+
+def _assign_per_reciter_order_splits(rows: list[dict]) -> None:
+    rows_by_reciter: dict[str, list[dict]] = defaultdict(list)
+    for row in rows:
+        rows_by_reciter[row["reciter_id"]].append(row)
+
+    for reciter_rows in rows_by_reciter.values():
+        sample_count = len(reciter_rows)
+        if sample_count < 3:
+            val_count = test_count = 0
+        else:
+            val_count = max(1, ceil(sample_count * 0.10))
+            test_count = max(1, ceil(sample_count * 0.10))
+        train_count = sample_count - val_count - test_count
+
+        for index, row in enumerate(reciter_rows):
+            if index < train_count:
+                row["split"] = "train"
+            elif index < train_count + val_count:
+                row["split"] = "val"
+            else:
+                row["split"] = "test"
+
+
+def _assign_surah_splits(
+    rows: list[dict],
+    train_surah_max: int,
+    val_surah_max: int,
+) -> None:
+    for row in rows:
+        surah_id = row.get("surah_id")
+        if surah_id is None:
+            raise ValueError(
+                f"Cannot use the surah split strategy because {row['path']} "
+                "does not have a six-digit SSSAAA filename."
+            )
+        if surah_id <= train_surah_max:
+            row["split"] = "train"
+        elif surah_id <= val_surah_max:
+            row["split"] = "val"
+        else:
+            row["split"] = "test"
+
+
+def build_manifest(
+    audio_root: Path,
+    output: Path,
+    min_duration_sec: float = 1.0,
+    max_duration_sec: float = 60.0,
+    split_strategy: str = "per_reciter_order",
+    train_surah_max: int = 95,
+    val_surah_max: int = 105,
+) -> dict:
+    rows: list[dict] = []
+    root = audio_root.resolve()
+
+    audio_paths = list(iter_audio_files(root))
+    for audio_path in tqdm(audio_paths, desc="Building manifest", unit="audio"):
+        try:
+            reciter_id = audio_path.relative_to(root).parts[0]
+        except IndexError:
+            continue
+
+        duration = audio_duration_sec(audio_path)
+        if duration is None:
+            continue
+        if duration < min_duration_sec or duration > max_duration_sec:
+            continue
+
+        row = {
+            "path": str(audio_path.resolve()),
+            "reciter_id": reciter_id,
+            "reciter_name": reciter_id,
+            "duration_sec": duration,
+        }
+        if len(audio_path.stem) == 6 and audio_path.stem.isdigit():
+            row["surah_id"] = int(audio_path.stem[:3])
+            row["ayah_id"] = int(audio_path.stem[3:])
+        rows.append(row)
+
+    if split_strategy == "per_reciter_order":
+        _assign_per_reciter_order_splits(rows)
+    elif split_strategy == "surah":
+        _assign_surah_splits(rows, train_surah_max, val_surah_max)
+    else:
+        raise ValueError(
+            f"Unknown split strategy {split_strategy!r}; expected "
+            "'per_reciter_order' or 'surah'."
+        )
+    write_jsonl(output, rows)
+    summary = summarize_manifest(rows)
+    summary_path = output.with_name("summary.json")
+    summary_path.write_text(json.dumps(summary, indent=2), encoding="utf-8")
+    return summary
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description="Build a reciter manifest from audio folders.")
+    parser.add_argument("--audio-root", required=True, type=Path)
+    parser.add_argument("--output", required=True, type=Path)
+    parser.add_argument("--min-duration-sec", default=1.0, type=float)
+    parser.add_argument("--max-duration-sec", default=60.0, type=float)
+    parser.add_argument("--config", default=DEFAULT_CONFIG_PATH, type=Path)
+    args = parser.parse_args()
+
+    config = load_config(args.config)
+    dataset_config = config.get("dataset", {})
+
+    summary = build_manifest(
+        audio_root=args.audio_root,
+        output=args.output,
+        min_duration_sec=args.min_duration_sec,
+        max_duration_sec=args.max_duration_sec,
+        split_strategy=str(dataset_config.get("split_strategy", "per_reciter_order")),
+        train_surah_max=int(dataset_config.get("train_surah_max", 95)),
+        val_surah_max=int(dataset_config.get("val_surah_max", 105)),
+    )
+    print(json.dumps(summary, indent=2))
+
+
+if __name__ == "__main__":
+    main()

src/quran_reciter_id/config.py

@@ -0,0 +1,72 @@
+from __future__ import annotations
+
+from pathlib import Path
+from typing import Any
+
+import yaml
+
+
+DEFAULT_CONFIG_PATH = Path("configs/ecapa_quran_md.yaml")
+
+
+def load_config(path: str | Path | None = None) -> dict[str, Any]:
+    config_path = Path(path) if path else DEFAULT_CONFIG_PATH
+    with config_path.open("r", encoding="utf-8") as handle:
+        return yaml.safe_load(handle)
+
+
+def resolve_device(device: str) -> str:
+    if not device.startswith("cuda") and device != "auto":
+        return device
+
+    import torch
+    import torch.nn.functional as F
+
+    if not torch.cuda.is_available():
+        if device == "auto":
+            return "cpu"
+        raise RuntimeError("CUDA was requested, but PyTorch cannot access a CUDA GPU")
+
+    if device == "auto":
+        candidates = list(range(torch.cuda.device_count()))
+        candidates.sort(key=lambda index: torch.cuda.mem_get_info(index)[0], reverse=True)
+    else:
+        try:
+            requested_index = int(device.split(":", 1)[1]) if ":" in device else 0
+        except ValueError as exc:
+            raise ValueError(f"Invalid CUDA device: {device}") from exc
+        if requested_index >= torch.cuda.device_count():
+            raise RuntimeError(
+                f"{device} was requested, but only {torch.cuda.device_count()} CUDA "
+                "device(s) are visible. Use 'device: auto' or a valid logical index."
+            )
+        candidates = [requested_index]
+
+    failures: list[str] = []
+    for index in candidates:
+        candidate = f"cuda:{index}"
+        try:
+            x = torch.zeros(1, 1, 32, device=candidate)
+            weight = torch.zeros(1, 1, 3, device=candidate)
+            F.conv1d(x, weight)
+            torch.cuda.synchronize(index)
+            return candidate
+        except Exception as exc:
+            failures.append(f"{candidate}: {exc}")
+
+    # Keep training on the GPU if only the optional cuDNN acceleration layer is
+    # broken or incompatible with the installed CUDA/PyTorch runtime.
+    torch.backends.cudnn.enabled = False
+    for index in candidates:
+        candidate = f"cuda:{index}"
+        try:
+            x = torch.zeros(1, 1, 32, device=candidate)
+            weight = torch.zeros(1, 1, 3, device=candidate)
+            F.conv1d(x, weight)
+            torch.cuda.synchronize(index)
+            print(f"Warning: cuDNN initialization failed; using {candidate} without cuDNN.")
+            return candidate
+        except Exception as exc:
+            failures.append(f"{candidate} without cuDNN: {exc}")
+
+    raise RuntimeError("No working CUDA device was found. " + " | ".join(failures))

src/quran_reciter_id/embeddings.py

@@ -0,0 +1,67 @@
+from __future__ import annotations
+
+from pathlib import Path
+
+import numpy as np
+import torch
+from speechbrain.inference.speaker import EncoderClassifier
+from tqdm import tqdm
+
+from .audio import load_audio
+
+
+class EcapaEmbedder:
+    def __init__(
+        self,
+        source: str = "speechbrain/spkrec-ecapa-voxceleb",
+        device: str = "cpu",
+        savedir: str | Path = "pretrained_models/spkrec-ecapa-voxceleb",
+        sample_rate: int = 16000,
+    ) -> None:
+        self.sample_rate = sample_rate
+        self.device = device
+        self.classifier = EncoderClassifier.from_hparams(
+            source=source,
+            savedir=str(savedir),
+            run_opts={"device": device},
+        )
+
+    @torch.inference_mode()
+    def encode_file(self, path: str | Path, max_duration_sec: float | None = 45.0) -> np.ndarray:
+        waveform = load_audio(path, sample_rate=self.sample_rate, max_duration_sec=max_duration_sec)
+        waveform = waveform.to(self.device)
+        embedding = self.classifier.encode_batch(waveform)
+        return embedding.squeeze().detach().cpu().numpy().astype(np.float32)
+
+
+def extract_embeddings(
+    rows: list[dict],
+    embedder: EcapaEmbedder,
+    label_to_idx: dict[str, int],
+    max_duration_sec: float | None = 45.0,
+) -> tuple[np.ndarray, np.ndarray, np.ndarray]:
+    embeddings: list[np.ndarray] = []
+    labels: list[int] = []
+    keep_indices: list[int] = []
+
+    for index, row in enumerate(tqdm(rows, desc="Extracting ECAPA embeddings")):
+        try:
+            embedding = embedder.encode_file(row["path"], max_duration_sec=max_duration_sec)
+        except Exception as exc:
+            error_message = str(exc)
+            if "CUDA" in error_message.upper() or "CUDNN" in error_message.upper():
+                raise RuntimeError(
+                    f"GPU embedding inference failed on {embedder.device}: {exc}. "
+                    "Set 'device: cpu' in the config, or select a working CUDA device."
+                ) from exc
+            print(f"Skipping {row['path']}: {exc}")
+            continue
+        embeddings.append(embedding)
+        labels.append(label_to_idx[row["reciter_id"]])
+        keep_indices.append(index)
+
+    return (
+        np.stack(embeddings).astype(np.float32),
+        np.asarray(labels, dtype=np.int64),
+        np.asarray(keep_indices, dtype=np.int64),
+    )

src/quran_reciter_id/export_quran_md.py

@@ -0,0 +1,197 @@
+from __future__ import annotations
+
+import argparse
+import io
+import json
+from collections import Counter
+from math import ceil
+from pathlib import Path
+
+import soundfile as sf
+import torchaudio
+from datasets import Audio, load_dataset
+from tqdm import tqdm
+
+from .audio import normalize_array_audio
+from .manifest import split_from_surah, summarize_manifest, write_jsonl
+
+
+def _decode_audio(audio: dict) -> tuple[object, int]:
+    """Decode a raw Hugging Face audio record with FFmpeg when available."""
+    if audio.get("bytes") is not None:
+        source = io.BytesIO(audio["bytes"])
+    elif audio.get("path"):
+        source = audio["path"]
+    else:
+        raise ValueError("Audio record contains neither bytes nor a path")
+
+    # Explicitly avoid the SoundFile backend for MP3. Its mpg123 decoder emits
+    # errors directly to stderr for damaged frames and rejects files that FFmpeg
+    # can often recover partially.
+    backends = torchaudio.list_audio_backends()
+    backend = "ffmpeg" if "ffmpeg" in backends else None
+    waveform, sampling_rate = torchaudio.load(source, backend=backend)
+    return waveform, sampling_rate
+
+
+def export_quran_md(
+    output_dir: Path,
+    max_samples_per_reciter: int | None = 10,
+    sample_rate: int = 16000,
+    streaming: bool = True,
+    reciters: set[str] | None = None,
+    verbose_skips: bool = False,
+    expected_reciters: int = 30,
+) -> dict:
+    if max_samples_per_reciter is not None and max_samples_per_reciter <= 0:
+        max_samples_per_reciter = None
+
+    target_by_split: dict[str, int] | None = None
+    if max_samples_per_reciter is not None:
+        if max_samples_per_reciter < 3:
+            val_target = 0
+            test_target = 0
+        else:
+            val_target = max(1, ceil(max_samples_per_reciter * 0.10))
+            test_target = max(1, ceil(max_samples_per_reciter * 0.10))
+        train_target = max(1, max_samples_per_reciter - val_target - test_target)
+        target_by_split = {"train": train_target, "val": val_target, "test": test_target}
+
+    output_dir.mkdir(parents=True, exist_ok=True)
+    clips_dir = output_dir / "clips"
+    manifest_path = output_dir / "manifest.jsonl"
+    summary_path = output_dir / "summary.json"
+
+    dataset = load_dataset("Buraaq/quran-md-ayahs", split="train", streaming=streaming)
+    # Some records in this dataset cannot be decoded by libsndfile. Keeping the
+    # encoded bytes lets us handle failures per record instead of crashing while
+    # IterableDataset is constructing the next row.
+    dataset = dataset.cast_column("audio", Audio(decode=False))
+    counts: Counter[str] = Counter()
+    split_counts: Counter[tuple[str, str]] = Counter()
+    rows: list[dict] = []
+    skipped_audio = 0
+
+    dataset_iterator = iter(dataset)
+    progress = tqdm(dataset_iterator, desc="Exporting Quran-MD", unit="audio")
+    for row in progress:
+        reciter_id = row["reciter_id"]
+        if reciters and reciter_id not in reciters:
+            continue
+
+        if max_samples_per_reciter is not None and counts[reciter_id] >= max_samples_per_reciter:
+            continue
+
+        surah_id = int(row["surah_id"])
+        ayah_id = int(row["ayah_id"])
+        if target_by_split is None:
+            split = split_from_surah(surah_id)
+        else:
+            # Assign consecutive valid clips to the requested 80/10/10 quota.
+            # This avoids scanning most of the 35 GB dataset just to reach the
+            # late Surahs used by split_from_surah.
+            reciter_position = counts[reciter_id]
+            if reciter_position < target_by_split["train"]:
+                split = "train"
+            elif reciter_position < target_by_split["train"] + target_by_split["val"]:
+                split = "val"
+            else:
+                split = "test"
+
+        try:
+            audio_array, source_rate = _decode_audio(row["audio"])
+        except Exception as exc:
+            skipped_audio += 1
+            if verbose_skips:
+                tqdm.write(
+                    f"Skipping unreadable audio for {reciter_id} "
+                    f"{surah_id:03d}:{ayah_id:03d}: {exc}"
+                )
+            continue
+
+        waveform = normalize_array_audio(audio_array, source_rate, sample_rate)
+        duration = waveform.shape[-1] / sample_rate
+        if duration < 1.0:
+            continue
+
+        rel_path = Path("clips") / reciter_id / f"{surah_id:03d}{ayah_id:03d}.wav"
+        audio_path = output_dir / rel_path
+        audio_path.parent.mkdir(parents=True, exist_ok=True)
+        sf.write(str(audio_path), waveform.squeeze(0).numpy(), sample_rate)
+
+        counts[reciter_id] += 1
+        split_counts[(reciter_id, split)] += 1
+        rows.append(
+            {
+                "path": str(audio_path.resolve()),
+                "reciter_id": reciter_id,
+                "reciter_name": row["reciter_name"],
+                "surah_id": surah_id,
+                "ayah_id": ayah_id,
+                "duration_sec": duration,
+                "split": split,
+            }
+        )
+        progress.set_postfix(
+            reciters=len(counts),
+            exported=len(rows),
+            skipped=skipped_audio,
+        )
+
+        target_reciter_count = len(reciters) if reciters else expected_reciters
+        if (
+            max_samples_per_reciter is not None
+            and len(counts) >= target_reciter_count
+            and all(count >= max_samples_per_reciter for count in counts.values())
+        ):
+            break
+
+    # Breaking a streaming Hugging Face iterator can leave its download worker
+    # alive until interpreter shutdown, which may trigger a PyGILState fatal
+    # error. Close both wrappers while Python is still fully initialized.
+    progress.close()
+    close_iterator = getattr(dataset_iterator, "close", None)
+    if close_iterator is not None:
+        close_iterator()
+
+    write_jsonl(manifest_path, rows)
+    summary = summarize_manifest(rows)
+    summary["skipped_unreadable_audio"] = skipped_audio
+    summary_path.write_text(json.dumps(summary, indent=2, ensure_ascii=False), encoding="utf-8")
+    return summary
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description="Export a balanced Quran-MD subset.")
+    parser.add_argument("--output-dir", default=Path("data/quran_md"), type=Path)
+    parser.add_argument("--max-samples-per-reciter", default=10, type=int)
+    parser.add_argument("--sample-rate", default=16000, type=int)
+    parser.add_argument("--no-streaming", action="store_true")
+    parser.add_argument("--reciters", nargs="*", default=None)
+    parser.add_argument(
+        "--expected-reciters",
+        default=30,
+        type=int,
+        help="Stop after this many reciters reach the sample limit (default: 30).",
+    )
+    parser.add_argument(
+        "--verbose-skips",
+        action="store_true",
+        help="Print one message for every unreadable dataset record.",
+    )
+    args = parser.parse_args()
+
+    summary = export_quran_md(
+        output_dir=args.output_dir,
+        max_samples_per_reciter=args.max_samples_per_reciter,
+        sample_rate=args.sample_rate,
+        streaming=not args.no_streaming,
+        reciters=set(args.reciters) if args.reciters else None,
+        verbose_skips=args.verbose_skips,
+        expected_reciters=args.expected_reciters,
+    )
+    print(json.dumps(summary, indent=2, ensure_ascii=False))
+
+
+if __name__ == "__main__":
+    main()

src/quran_reciter_id/manifest.py

@@ -0,0 +1,67 @@
+from __future__ import annotations
+
+import hashlib
+import json
+from collections import Counter
+from pathlib import Path
+from typing import Iterable
+
+
+AUDIO_EXTENSIONS = {".flac", ".m4a", ".mp3", ".ogg", ".wav"}
+
+
+def read_jsonl(path: str | Path) -> list[dict]:
+    rows: list[dict] = []
+    with Path(path).open("r", encoding="utf-8") as handle:
+        for line in handle:
+            line = line.strip()
+            if line:
+                rows.append(json.loads(line))
+    return rows
+
+
+def write_jsonl(path: str | Path, rows: Iterable[dict]) -> None:
+    output = Path(path)
+    output.parent.mkdir(parents=True, exist_ok=True)
+    with output.open("w", encoding="utf-8") as handle:
+        for row in rows:
+            handle.write(json.dumps(row, ensure_ascii=False) + "\n")
+
+
+def iter_audio_files(audio_root: str | Path) -> Iterable[Path]:
+    root = Path(audio_root)
+    for path in sorted(root.rglob("*")):
+        if path.is_file() and path.suffix.lower() in AUDIO_EXTENSIONS:
+            yield path
+
+
+def split_from_surah(surah_id: int | None) -> str:
+    if surah_id is None:
+        return "train"
+    if surah_id <= 95:
+        return "train"
+    if surah_id <= 105:
+        return "val"
+    return "test"
+
+
+def stable_hash_split(value: str, train_pct: int = 80, val_pct: int = 10) -> str:
+    digest = hashlib.sha1(value.encode("utf-8")).hexdigest()
+    bucket = int(digest[:8], 16) % 100
+    if bucket < train_pct:
+        return "train"
+    if bucket < train_pct + val_pct:
+        return "val"
+    return "test"
+
+
+def summarize_manifest(rows: Iterable[dict]) -> dict:
+    rows = list(rows)
+    split_counts = Counter(row.get("split", "train") for row in rows)
+    reciter_counts = Counter(row["reciter_id"] for row in rows)
+    return {
+        "num_samples": len(rows),
+        "num_reciters": len(reciter_counts),
+        "splits": dict(sorted(split_counts.items())),
+        "samples_per_reciter": dict(sorted(reciter_counts.items())),
+    }

src/quran_reciter_id/model.py

@@ -0,0 +1,28 @@
+from __future__ import annotations
+
+import torch
+from torch import nn
+
+
+class EmbeddingClassifier(nn.Module):
+    def __init__(
+        self,
+        input_dim: int,
+        num_classes: int,
+        hidden_dim: int = 256,
+        dropout: float = 0.2,
+    ) -> None:
+        super().__init__()
+        self.net = nn.Sequential(
+            nn.Linear(input_dim, hidden_dim),
+            nn.ReLU(),
+            nn.Dropout(dropout),
+            nn.Linear(hidden_dim, num_classes),
+        )
+
+    def forward(self, embeddings: torch.Tensor) -> torch.Tensor:
+        return self.net(embeddings)
+
+
+def l2_normalize(x: torch.Tensor, eps: float = 1e-12) -> torch.Tensor:
+    return x / x.norm(dim=-1, keepdim=True).clamp_min(eps)

src/quran_reciter_id/predict.py

@@ -0,0 +1,97 @@
+from __future__ import annotations
+
+import argparse
+import json
+from pathlib import Path
+
+import torch
+
+from .config import resolve_device
+from .embeddings import EcapaEmbedder
+from .model import EmbeddingClassifier, l2_normalize
+
+
+def load_checkpoint(run_dir: Path, device: str) -> dict:
+    try:
+        return torch.load(run_dir / "model.pt", map_location=device, weights_only=False)
+    except TypeError:
+        return torch.load(run_dir / "model.pt", map_location=device)
+
+
+@torch.inference_mode()
+def predict(run_dir: Path, audio_file: Path, top_k: int = 5, device: str = "auto") -> dict:
+    resolved_device = resolve_device(device)
+    checkpoint = load_checkpoint(run_dir, resolved_device)
+    labels = checkpoint["labels"]
+
+    embedder = EcapaEmbedder(
+        source=checkpoint["embedding_model"],
+        device=resolved_device,
+        sample_rate=int(checkpoint["sample_rate"]),
+    )
+    embedding = (
+        torch.from_numpy(embedder.encode_file(audio_file))
+        .float()
+        .unsqueeze(0)
+        .to(resolved_device)
+    )
+    mean = checkpoint["mean"].to(resolved_device)
+    std = checkpoint["std"].to(resolved_device)
+    embedding = (embedding - mean) / std
+
+    model = EmbeddingClassifier(
+        input_dim=int(checkpoint["input_dim"]),
+        num_classes=len(labels),
+        hidden_dim=int(checkpoint["classifier"]["hidden_dim"]),
+        dropout=float(checkpoint["classifier"]["dropout"]),
+    ).to(resolved_device)
+    model.load_state_dict(checkpoint["state_dict"])
+    model.eval()
+
+    probs = torch.softmax(model(embedding), dim=-1).squeeze(0).cpu()
+    normalized = l2_normalize(embedding.cpu())
+    centroid_scores = (normalized @ checkpoint["centroids"].cpu().T).squeeze(0)
+
+    k = min(top_k, len(labels))
+    top_probs, top_indices = probs.topk(k)
+    candidates = []
+    for probability, index in zip(top_probs.tolist(), top_indices.tolist(), strict=True):
+        candidates.append(
+            {
+                "reciter_id": labels[index],
+                "probability": float(probability),
+                "centroid_similarity": float(centroid_scores[index].item()),
+            }
+        )
+
+    inference_cfg = checkpoint.get("inference", {})
+    unknown_threshold = float(inference_cfg.get("unknown_threshold", 0.45))
+    min_centroid_similarity = float(inference_cfg.get("min_centroid_similarity", 0.25))
+    best = candidates[0]
+    is_unknown = (
+        best["probability"] < unknown_threshold
+        or best["centroid_similarity"] < min_centroid_similarity
+    )
+
+    return {
+        "audio_file": str(audio_file.resolve()),
+        "prediction": None if is_unknown else best["reciter_id"],
+        "is_unknown": is_unknown,
+        "top_candidates": candidates,
+    }
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description="Predict the reciter for an audio file.")
+    parser.add_argument("--run-dir", required=True, type=Path)
+    parser.add_argument("--audio-file", required=True, type=Path)
+    parser.add_argument("--top-k", default=5, type=int)
+    parser.add_argument("--device", default="auto")
+    args = parser.parse_args()
+
+    result = predict(args.run_dir, args.audio_file, top_k=args.top_k, device=args.device)
+    print(json.dumps(result, indent=2, ensure_ascii=False))
+
+
+if __name__ == "__main__":
+    main()

src/quran_reciter_id/train.py

@@ -0,0 +1,246 @@
+from __future__ import annotations
+
+import argparse
+import json
+import random
+from pathlib import Path
+
+import numpy as np
+import torch
+from sklearn.metrics import accuracy_score, classification_report
+from torch.utils.data import DataLoader, TensorDataset
+
+from .config import load_config, resolve_device
+from .embeddings import EcapaEmbedder, extract_embeddings
+from .manifest import read_jsonl
+from .model import EmbeddingClassifier, l2_normalize
+
+
+def set_seed(seed: int) -> None:
+    random.seed(seed)
+    np.random.seed(seed)
+    torch.manual_seed(seed)
+    if torch.cuda.is_available():
+        torch.cuda.manual_seed_all(seed)
+
+
+def split_masks(splits: np.ndarray) -> dict[str, np.ndarray]:
+    return {name: splits == name for name in ["train", "val", "test"]}
+
+
+def make_loaders(
+    x: np.ndarray,
+    y: np.ndarray,
+    train_mask: np.ndarray,
+    val_mask: np.ndarray,
+    batch_size: int,
+) -> tuple[DataLoader, DataLoader | None]:
+    train_ds = TensorDataset(torch.from_numpy(x[train_mask]), torch.from_numpy(y[train_mask]))
+    train_loader = DataLoader(train_ds, batch_size=batch_size, shuffle=True)
+
+    if val_mask.any():
+        val_ds = TensorDataset(torch.from_numpy(x[val_mask]), torch.from_numpy(y[val_mask]))
+        val_loader = DataLoader(val_ds, batch_size=batch_size, shuffle=False)
+    else:
+        val_loader = None
+
+    return train_loader, val_loader
+
+
+@torch.inference_mode()
+def predict_logits(model: torch.nn.Module, x: np.ndarray, device: str, batch_size: int = 256) -> np.ndarray:
+    model.eval()
+    preds: list[np.ndarray] = []
+    for start in range(0, len(x), batch_size):
+        batch = torch.from_numpy(x[start : start + batch_size]).to(device)
+        preds.append(model(batch).detach().cpu().numpy())
+    return np.concatenate(preds, axis=0)
+
+
+def evaluate_split(
+    model: torch.nn.Module,
+    x: np.ndarray,
+    y: np.ndarray,
+    mask: np.ndarray,
+    labels: list[str],
+    device: str,
+) -> dict:
+    if not mask.any():
+        return {"samples": 0}
+    logits = predict_logits(model, x[mask], device)
+    pred = logits.argmax(axis=1)
+    result = {
+        "samples": int(mask.sum()),
+        "accuracy": float(accuracy_score(y[mask], pred)),
+    }
+    if len(labels) <= 50:
+        result["classification_report"] = classification_report(
+            y[mask],
+            pred,
+            labels=list(range(len(labels))),
+            target_names=labels,
+            zero_division=0,
+            output_dict=True,
+        )
+    return result
+
+
+def train_classifier(
+    x: np.ndarray,
+    y: np.ndarray,
+    splits: np.ndarray,
+    labels: list[str],
+    config: dict,
+    output_dir: Path,
+    device: str,
+) -> dict:
+    classifier_cfg = config["classifier"]
+    masks = split_masks(splits)
+    train_mask = masks["train"]
+    val_mask = masks["val"]
+
+    mean = x[train_mask].mean(axis=0, keepdims=True).astype(np.float32)
+    std = (x[train_mask].std(axis=0, keepdims=True) + 1e-6).astype(np.float32)
+    x_norm = ((x - mean) / std).astype(np.float32)
+
+    train_loader, val_loader = make_loaders(
+        x_norm,
+        y,
+        train_mask,
+        val_mask,
+        batch_size=int(classifier_cfg["batch_size"]),
+    )
+
+    model = EmbeddingClassifier(
+        input_dim=x.shape[1],
+        num_classes=len(labels),
+        hidden_dim=int(classifier_cfg["hidden_dim"]),
+        dropout=float(classifier_cfg["dropout"]),
+    ).to(device)
+
+    optimizer = torch.optim.AdamW(
+        model.parameters(),
+        lr=float(classifier_cfg["learning_rate"]),
+        weight_decay=float(classifier_cfg["weight_decay"]),
+    )
+    loss_fn = torch.nn.CrossEntropyLoss()
+
+    best_state = None
+    best_val = -1.0
+    patience_left = int(classifier_cfg["patience"])
+
+    for epoch in range(1, int(classifier_cfg["epochs"]) + 1):
+        model.train()
+        total_loss = 0.0
+        for batch_x, batch_y in train_loader:
+            batch_x = batch_x.to(device)
+            batch_y = batch_y.to(device)
+            optimizer.zero_grad(set_to_none=True)
+            loss = loss_fn(model(batch_x), batch_y)
+            loss.backward()
+            optimizer.step()
+            total_loss += float(loss.item()) * batch_x.shape[0]
+
+        if val_loader is not None:
+            val_logits = predict_logits(model, x_norm[val_mask], device)
+            val_score = float(accuracy_score(y[val_mask], val_logits.argmax(axis=1)))
+        else:
+            train_logits = predict_logits(model, x_norm[train_mask], device)
+            val_score = float(accuracy_score(y[train_mask], train_logits.argmax(axis=1)))
+
+        avg_loss = total_loss / max(1, int(train_mask.sum()))
+        print(f"epoch={epoch:03d} loss={avg_loss:.4f} score={val_score:.4f}")
+
+        if val_score > best_val:
+            best_val = val_score
+            best_state = {key: value.detach().cpu().clone() for key, value in model.state_dict().items()}
+            patience_left = int(classifier_cfg["patience"])
+        else:
+            patience_left -= 1
+            if patience_left <= 0:
+                break
+
+    if best_state is not None:
+        model.load_state_dict(best_state)
+
+    x_train_tensor = torch.from_numpy(x_norm[train_mask])
+    y_train_tensor = torch.from_numpy(y[train_mask])
+    centroids = []
+    for class_index in range(len(labels)):
+        class_embeddings = x_train_tensor[y_train_tensor == class_index]
+        centroid = class_embeddings.mean(dim=0)
+        centroids.append(centroid)
+    centroids_tensor = l2_normalize(torch.stack(centroids))
+
+    metrics = {
+        "train": evaluate_split(model, x_norm, y, train_mask, labels, device),
+        "val": evaluate_split(model, x_norm, y, masks["val"], labels, device),
+        "test": evaluate_split(model, x_norm, y, masks["test"], labels, device),
+        "best_validation_score": best_val,
+    }
+
+    checkpoint = {
+        "state_dict": model.state_dict(),
+        "labels": labels,
+        "mean": torch.from_numpy(mean),
+        "std": torch.from_numpy(std),
+        "centroids": centroids_tensor,
+        "input_dim": x.shape[1],
+        "classifier": classifier_cfg,
+        "embedding_model": config["embedding_model"],
+        "sample_rate": config["sample_rate"],
+        "inference": config.get("inference", {}),
+    }
+    torch.save(checkpoint, output_dir / "model.pt")
+    (output_dir / "labels.json").write_text(json.dumps(labels, indent=2), encoding="utf-8")
+    (output_dir / "metrics.json").write_text(json.dumps(metrics, indent=2), encoding="utf-8")
+    return metrics
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description="Train a Quran reciter identification model.")
+    parser.add_argument("--manifest", required=True, type=Path)
+    parser.add_argument("--output-dir", default=Path("runs/ecapa_quran"), type=Path)
+    parser.add_argument("--config", default=Path("configs/ecapa_quran_md.yaml"), type=Path)
+    parser.add_argument("--recompute-embeddings", action="store_true")
+    args = parser.parse_args()
+
+    config = load_config(args.config)
+    set_seed(int(config["seed"]))
+    device = resolve_device(str(config.get("device", "auto")))
+    args.output_dir.mkdir(parents=True, exist_ok=True)
+    embeddings_dir = args.output_dir / "embeddings"
+    embeddings_dir.mkdir(parents=True, exist_ok=True)
+    cache_path = embeddings_dir / "all_embeddings.npz"
+
+    rows = read_jsonl(args.manifest)
+    labels = sorted({row["reciter_id"] for row in rows})
+    label_to_idx = {label: index for index, label in enumerate(labels)}
+
+    if cache_path.exists() and not args.recompute_embeddings:
+        cached = np.load(cache_path, allow_pickle=True)
+        x = cached["embeddings"]
+        y = cached["labels"]
+        keep_indices = cached["keep_indices"]
+    else:
+        embedder = EcapaEmbedder(
+            source=config["embedding_model"],
+            device=device,
+            sample_rate=int(config["sample_rate"]),
+        )
+        x, y, keep_indices = extract_embeddings(
+            rows,
+            embedder,
+            label_to_idx,
+            max_duration_sec=float(config["dataset"]["max_duration_sec"]),
+        )
+        np.savez_compressed(cache_path, embeddings=x, labels=y, keep_indices=keep_indices)
+
+    kept_rows = [rows[int(index)] for index in keep_indices]
+    splits = np.asarray([row.get("split", "train") for row in kept_rows])
+    metrics = train_classifier(x, y, splits, labels, config, args.output_dir, device)
+    print(json.dumps(metrics, indent=2))
+
+
+if __name__ == "__main__":
+    main()