Upload 6 files

.gitignore

@@ -0,0 +1,9 @@
+__pycache__/
+*.py[cod]
+*.pth
+*.pt
+*.egg-info/
+.env
+.venv/
+dist/
+build/

Dockerfile

@@ -0,0 +1,28 @@
+# ── Base image ──────────────────────────────────────────────────────────────
+FROM python:3.10-slim
+
+# ── System deps (librosa needs libsndfile) ───────────────────────────────────
+RUN apt-get update && apt-get install -y --no-install-recommends \
+        libsndfile1 \
+        ffmpeg \
+    && rm -rf /var/lib/apt/lists/*
+
+# ── Working directory ────────────────────────────────────────────────────────
+WORKDIR /app
+
+# ── Python deps (cached layer) ───────────────────────────────────────────────
+COPY requirements.txt .
+RUN pip install --no-cache-dir -r requirements.txt
+
+# ── Copy application code ────────────────────────────────────────────────────
+COPY app.py .
+
+# ── Copy model weights (add your .pth file here before building) ─────────────
+# If you push the model file to the Space repo, Docker will pick it up here.
+# COPY entire_model.pth .
+
+# ── HF Spaces listens on port 7860 ──────────────────────────────────────────
+EXPOSE 7860
+
+# ── Launch ───────────────────────────────────────────────────────────────────
+CMD ["uvicorn", "app:app", "--host", "0.0.0.0", "--port", "7860"]

README.md

@@ -1,10 +1,39 @@
 ---
-title: TTS
-emoji: 🐢
-colorFrom: gray
-colorTo: pink
+title: Arabic Speech-to-Text
+emoji: 🎙️
+colorFrom: green
+colorTo: blue
 sdk: docker
 pinned: false
+license: mit
 ---
 
-Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference
+# 🎙️ Arabic Speech-to-Text (Egyptian Dialect)
+
+A custom Transformer-based ASR model trained on Egyptian Arabic speech.
+
+## API Endpoints
+
+| Method | Path | Description |
+|--------|------|-------------|
+| GET | `/` | Health check |
+| GET | `/vocab` | View vocabulary |
+| POST | `/transcribe` | Upload audio → get transcription |
+
+Interactive docs: **`/docs`** (Swagger UI)  
+Alternative docs: **`/redoc`**
+
+## How to use
+
+1. Open `/docs` in your browser.
+2. Click **POST /transcribe → Try it out**.
+3. Upload a WAV/MP3 file.
+4. Click **Execute** and read the transcription.
+
+## Model details
+
+- Architecture: Custom Encoder-Decoder Transformer
+- Encoder input: Mel-Spectrogram (128 mels, 15-sec chunks)
+- Decoder: Character-level greedy decoding
+- Language: Egyptian Arabic
+- Vocabulary size: ~75 characters (Arabic + English + special tokens)

app (1).py

@@ -0,0 +1,448 @@
+import os
+import math
+import gc
+import re
+import string
+import tempfile
+
+import numpy as np
+import torch
+import torch.nn as nn
+import torch.nn.functional as F
+import librosa
+import noisereduce as nr
+from num2words import num2words
+
+from fastapi import FastAPI, File, UploadFile, HTTPException
+from fastapi.responses import JSONResponse
+import uvicorn
+
+# ─────────────────────────────────────────────
+# Constants
+# ─────────────────────────────────────────────
+N_MELS        = 128
+SAMPLE_RATE   = 16000
+HOP_LENGTH    = 512
+N_FFT         = 2048
+CHUNK_LENGTH  = 15
+N_SAMPLES     = SAMPLE_RATE * CHUNK_LENGTH
+N_FRAMES      = math.ceil(N_SAMPLES / HOP_LENGTH)
+MAX_TEXT_LEN  = 70
+MAX_SEQ_LEN   = 70
+NEG_INFTY     = -1e9
+
+MODEL_PATH = "entire_model.pth"   # ضع وزن الموديل هنا
+
+# ─────────────────────────────────────────────
+# Vocabulary
+# ─────────────────────────────────────────────
+special_tokens    = ['<PAD>', '<UNK>', '<SOS>', '<EOS>']
+english_chars     = list(string.ascii_lowercase + ' ')
+arabic_chars      = list("ابتثجحخدذرزسشصضطظعغفقكلمنهويئءىةؤ")
+vocab             = special_tokens + english_chars + arabic_chars
+char2idx          = {c: i for i, c in enumerate(vocab)}
+idx2char          = {i: c for i, c in enumerate(vocab)}
+VOCAB_SIZE        = len(vocab)
+
+# ─────────────────────────────────────────────
+# Model Architecture (same as notebook)
+# ─────────────────────────────────────────────
+class MultiHeadAttention(nn.Module):
+    def __init__(self, d_model, num_heads):
+        super().__init__()
+        assert d_model % num_heads == 0
+        self.d_model   = d_model
+        self.num_heads = num_heads
+        self.head_dim  = d_model // num_heads
+        self.qkv_layer = nn.Linear(d_model, 3 * d_model)
+        self.linear    = nn.Linear(d_model, d_model)
+
+    def forward(self, x, mask=None):
+        B, L, _ = x.shape
+        qkv = self.qkv_layer(x).reshape(B, L, self.num_heads, 3 * self.head_dim)
+        qkv = qkv.permute(0, 2, 1, 3)
+        q, k, v = qkv.chunk(3, dim=-1)
+        attn = torch.matmul(q, k.transpose(-2, -1)) / math.sqrt(self.head_dim)
+        if mask is not None:
+            attn += mask
+        attn = F.softmax(attn, dim=-1)
+        out  = torch.matmul(attn, v).permute(0, 2, 1, 3).reshape(B, L, self.d_model)
+        return self.linear(out)
+
+
+class MultiHeadCrossAttention(nn.Module):
+    def __init__(self, d_model, num_heads):
+        super().__init__()
+        assert d_model % num_heads == 0
+        self.d_model   = d_model
+        self.num_heads = num_heads
+        self.head_dim  = d_model // num_heads
+        self.kv_layer  = nn.Linear(d_model, 2 * d_model)
+        self.q_layer   = nn.Linear(d_model, d_model)
+        self.linear    = nn.Linear(d_model, d_model)
+
+    def forward(self, x, y, mask=None):
+        B, L, _ = x.shape
+        kv = self.kv_layer(y).reshape(B, y.shape[1], self.num_heads, 2 * self.head_dim).permute(0, 2, 1, 3)
+        k, v = kv.chunk(2, dim=-1)
+        q    = self.q_layer(x).reshape(B, L, self.num_heads, self.head_dim).permute(0, 2, 1, 3)
+        attn = torch.matmul(q, k.transpose(-2, -1)) / math.sqrt(self.head_dim)
+        if mask is not None:
+            attn += mask
+        attn = F.softmax(attn, dim=-1)
+        out  = torch.matmul(attn, v).permute(0, 2, 1, 3).reshape(B, L, self.d_model)
+        return self.linear(out)
+
+
+class LayerNormalization(nn.Module):
+    def __init__(self, d_model, eps=1e-5):
+        super().__init__()
+        self.gamma = nn.Parameter(torch.ones(d_model))
+        self.beta  = nn.Parameter(torch.zeros(d_model))
+        self.eps   = eps
+
+    def forward(self, x):
+        mean = x.mean(-1, keepdim=True)
+        var  = ((x - mean) ** 2).mean(-1, keepdim=True)
+        return self.gamma * (x - mean) / torch.sqrt(var + self.eps) + self.beta
+
+
+class PositionwiseFeedForward(nn.Module):
+    def __init__(self, d_model, hidden, drop_prob=0.1):
+        super().__init__()
+        self.net = nn.Sequential(
+            nn.Linear(d_model, hidden), nn.ReLU(), nn.Dropout(drop_prob), nn.Linear(hidden, d_model)
+        )
+    def forward(self, x): return self.net(x)
+
+
+class EncoderLayer(nn.Module):
+    def __init__(self, d_model, ffn_hidden, num_heads, drop_prob):
+        super().__init__()
+        self.attn  = MultiHeadAttention(d_model, num_heads)
+        self.norm1 = LayerNormalization(d_model)
+        self.ff    = PositionwiseFeedForward(d_model, ffn_hidden, drop_prob)
+        self.norm2 = LayerNormalization(d_model)
+        self.drop  = nn.Dropout(drop_prob)
+
+    def forward(self, x, mask=None):
+        x = self.norm1(x + self.drop(self.attn(x, mask)))
+        x = self.norm2(x + self.drop(self.ff(x)))
+        return x
+
+
+class DecoderLayer(nn.Module):
+    def __init__(self, d_model, ffn_hidden, num_heads, drop_prob):
+        super().__init__()
+        self.self_attn  = MultiHeadAttention(d_model, num_heads)
+        self.norm1      = LayerNormalization(d_model)
+        self.cross_attn = MultiHeadCrossAttention(d_model, num_heads)
+        self.norm2      = LayerNormalization(d_model)
+        self.ff         = PositionwiseFeedForward(d_model, ffn_hidden, drop_prob)
+        self.norm3      = LayerNormalization(d_model)
+        self.drop       = nn.Dropout(drop_prob)
+
+    def forward(self, x, enc_out, self_mask=None, cross_mask=None):
+        x = self.norm1(x + self.drop(self.self_attn(x, self_mask)))
+        x = self.norm2(x + self.drop(self.cross_attn(x, enc_out, cross_mask)))
+        x = self.norm3(x + self.drop(self.ff(x)))
+        return x
+
+
+class SequentialEncoder(nn.Sequential):
+    def forward(self, *args):
+        x, mask = args
+        for layer in self:
+            x = layer(x, mask)
+        return x
+
+
+class SequentialDecoder(nn.Sequential):
+    def forward(self, *args):
+        x, enc_out, self_mask, cross_mask = args
+        for layer in self:
+            x = layer(x, enc_out, self_mask, cross_mask)
+        return x
+
+
+class Encoder(nn.Module):
+    def __init__(self, d_model, ffn_hidden, num_heads, drop_prob, num_layers):
+        super().__init__()
+        self.layers = SequentialEncoder(*[EncoderLayer(d_model, ffn_hidden, num_heads, drop_prob) for _ in range(num_layers)])
+
+    def forward(self, x, mask=None):
+        return self.layers(x, mask)
+
+
+class Decoder(nn.Module):
+    def __init__(self, d_model, ffn_hidden, num_heads, drop_prob, num_layers):
+        super().__init__()
+        self.layers = SequentialDecoder(*[DecoderLayer(d_model, ffn_hidden, num_heads, drop_prob) for _ in range(num_layers)])
+
+    def forward(self, x, enc_out, self_mask=None, cross_mask=None):
+        return self.layers(x, enc_out, self_mask, cross_mask)
+
+
+class Transformer(nn.Module):
+    def __init__(self, d_model=512, num_heads=8, num_encoder_layers=6, num_decoder_layers=6, ffn_hidden=2048, drop_prob=0.1):
+        super().__init__()
+        self.encoder = Encoder(d_model, ffn_hidden, num_heads, drop_prob, num_encoder_layers)
+        self.decoder = Decoder(d_model, ffn_hidden, num_heads, drop_prob, num_decoder_layers)
+
+    def forward(self, src, tgt, encoder_self_attention_mask=None, decoder_self_attention_mask=None, decoder_cross_attention_mask=None):
+        src = self.encoder(src, encoder_self_attention_mask)
+        return self.decoder(tgt, src, decoder_self_attention_mask, decoder_cross_attention_mask)
+
+
+class PositionalEncoding(nn.Module):
+    def __init__(self, d_model, max_seq_len):
+        super().__init__()
+        pe  = torch.zeros(max_seq_len, d_model)
+        pos = torch.arange(max_seq_len, dtype=torch.float).unsqueeze(1)
+        div = torch.exp(torch.arange(0, d_model, 2).float() * (-math.log(10000.0) / d_model))
+        pe[:, 0::2] = torch.sin(pos * div)
+        pe[:, 1::2] = torch.cos(pos * div)
+        self.register_buffer('pe', pe.unsqueeze(0))
+
+    def forward(self): return self.pe
+
+
+def get_conv_Lout(L_in, conv):
+    return math.floor((L_in + 2 * conv.padding[0] - conv.dilation[0] * (conv.kernel_size[0] - 1) - 1) / conv.stride[0] + 1)
+
+
+class MMS(nn.Module):
+    def __init__(self, vocab_size, d_model=512, nhead=8, num_encoder_layers=6, num_decoder_layers=6,
+                 dim_feedforward=2048, max_encoder_seq_len=100, max_decoder_seq_len=100, n_mels=N_MELS, dropout=0.1):
+        super().__init__()
+        self.transformer = Transformer(d_model, nhead, num_encoder_layers, num_decoder_layers, dim_feedforward, dropout)
+        self.en_positional_encoding = PositionalEncoding(d_model, max_encoder_seq_len)
+        self.de_positional_encoding = PositionalEncoding(d_model, max_decoder_seq_len)
+        self.conv1     = nn.Conv1d(n_mels, d_model, kernel_size=3, padding=1)
+        self.conv2     = nn.Conv1d(d_model, d_model, kernel_size=3, stride=2, padding=1)
+        self.gelu      = nn.GELU()
+        self.embedding = nn.Embedding(vocab_size, d_model, padding_idx=0)
+        self.d_model   = d_model
+        self.ff        = nn.Linear(d_model, vocab_size)
+
+    def get_encoder_seq_len(self, L_in):
+        return get_conv_Lout(get_conv_Lout(L_in, self.conv1), self.conv2)
+
+    def forward(self, audio, text, enc_mask, dec_mask, cross_mask, device):
+        audio = self.gelu(self.conv1(audio))
+        audio = self.gelu(self.conv2(audio))
+        audio = audio.permute(0, 2, 1)
+        audio += self.en_positional_encoding().to(device)
+        text   = self.embedding(text) + self.de_positional_encoding().to(device)
+        out    = self.transformer(audio, text, enc_mask, dec_mask, cross_mask)
+        return self.ff(out)
+
+# ─────────────────────────────────────────────
+# Helpers
+# ─────────────────────────────────────────────
+def pad_or_trim(array, length=N_SAMPLES, axis=-1):
+    if array.shape[axis] > length:
+        array = array.take(indices=range(length), axis=axis)
+    if array.shape[axis] < length:
+        pad_widths        = [(0, 0)] * array.ndim
+        pad_widths[axis]  = (0, length - array.shape[axis])
+        array             = np.pad(array, pad_widths)
+    return array
+
+
+def preprocess_audio(file_path):
+    audio_data, _ = librosa.load(file_path, sr=SAMPLE_RATE)
+    audio_data    = nr.reduce_noise(y=audio_data, sr=SAMPLE_RATE)
+    duration      = librosa.get_duration(y=audio_data, sr=SAMPLE_RATE)
+    modified      = pad_or_trim(audio_data, padding=False)
+    sgram         = librosa.stft(y=modified, n_fft=N_FFT, hop_length=HOP_LENGTH)
+    sgram_mag, _  = librosa.magphase(sgram)
+    mel           = librosa.feature.melspectrogram(S=sgram_mag, sr=SAMPLE_RATE, n_fft=N_FFT, hop_length=HOP_LENGTH, n_mels=N_MELS)
+    mel_db        = librosa.amplitude_to_db(mel, ref=np.min)
+    orig_len      = mel_db.shape[-1]
+    padded        = np.pad(mel_db, ((0, 0), (0, N_FRAMES - mel_db.shape[-1]))) if orig_len < N_FRAMES else mel_db[:, :N_FRAMES]
+    return padded, orig_len, duration
+
+
+def nlp_preprocessing(sentence):
+    sentence = sentence.lower().replace("\n", " ")
+    sentence = re.sub(r'[إأآ]', 'ا', sentence)
+    sentence = re.sub(r'[^a-zA-Zء-ي\s\d]', '', sentence)
+    sentence = re.sub(r'[\u0617-\u061A\u064B-\u065F]', '', sentence)
+    sentence = re.sub(r'([a-zA-Z])([ء-ي])|([ء-ي])([a-zA-Z])', r'\1\3 \2\4', sentence)
+    sentence = re.sub(r'\s+', ' ', sentence)
+    return re.sub(r'\d+', lambda x: num2words(int(x.group()), lang='ar'), sentence)
+
+
+def tokenize_text(text, max_len=MAX_SEQ_LEN, start_token=True, end_token=True):
+    tokens  = [char2idx.get(c, char2idx['<UNK>']) for c in text]
+    max_len -= (start_token + end_token)
+    text_len = len(tokens)
+    if text_len < max_len:
+        if end_token: tokens += [char2idx['<EOS>']]
+        tokens += [char2idx['<PAD>']] * (max_len - text_len)
+    else:
+        tokens = tokens[:max_len]
+        if end_token: tokens += [char2idx['<EOS>']]
+    if start_token: tokens.insert(0, char2idx['<SOS>'])
+    return tokens
+
+
+def text_decoder(tokens):
+    result = ""
+    for t in tokens.tolist() if hasattr(tokens, 'tolist') else [tokens.item()]:
+        ch = idx2char.get(t, '')
+        if ch in ('<PAD>', '<SOS>', '<EOS>', '<UNK>'): break
+        result += ch
+    return result
+
+
+def generate_padding_masks(transcription, audio_original_len, conv_func, frames=N_FRAMES):
+    batch_size, seq_len = transcription.size()
+    audio_len           = conv_func(frames)
+    look_ahead_mask     = torch.triu(torch.full((seq_len, seq_len), True), diagonal=1)
+    enc_mask            = torch.full([batch_size, audio_len, audio_len], False)
+    dec_self_mask       = torch.full([batch_size, seq_len, seq_len], False)
+    dec_cross_mask      = torch.full([batch_size, seq_len, audio_len], False)
+
+    for i in range(batch_size):
+        new_len = conv_func(audio_original_len[i])
+        enc_mask[i, new_len:, :]  = True
+        enc_mask[i, :, new_len:]  = True
+        dec_cross_mask[i, :, new_len:] = True
+        zero_idx = np.where(transcription[0].cpu().numpy() == 0)[0]
+        if len(zero_idx):
+            idx = zero_idx[0]
+            dec_self_mask[i, idx:, :]  = True
+            dec_self_mask[i, :, idx:]  = True
+            dec_cross_mask[i, idx:, :] = True
+
+    dec_self_attn  = torch.where(look_ahead_mask | dec_self_mask, NEG_INFTY, 0.0)
+    dec_cross_attn = torch.where(dec_cross_mask, NEG_INFTY, 0.0)
+    enc_attn       = torch.where(enc_mask, NEG_INFTY, 0.0)
+    return enc_attn, dec_self_attn, dec_cross_attn
+
+
+# ─────────────────────────────────────────────
+# Load model
+# ─────────────────────────────────────────────
+DEVICE = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
+model  = None
+
+def load_model():
+    global model
+    if not os.path.exists(MODEL_PATH):
+        print(f"⚠️  Model file '{MODEL_PATH}' not found — transcription endpoint will return an error.")
+        return
+    try:
+        model = torch.load(MODEL_PATH, map_location=DEVICE, weights_only=False)
+        model.to(DEVICE)
+        model.eval()
+        print("✅ Model loaded successfully.")
+    except Exception as e:
+        print(f"❌ Failed to load model: {e}")
+
+
+# ─────────────────────────────────────────────
+# Greedy decoder
+# ─────────────────────────────────────────────
+def greedy_decode(audio_tensor, org_len):
+    transcription = ""
+    audio_tensor  = audio_tensor.to(DEVICE)
+    with torch.no_grad():
+        for i in range(MAX_TEXT_LEN):
+            tok = torch.tensor(tokenize_text(transcription, max_len=MAX_TEXT_LEN, end_token=False), dtype=torch.long)
+            tok = tok.unsqueeze(0).to(DEVICE)
+            enc_mask, dec_mask, cross_mask = generate_padding_masks(tok, [org_len], model.get_encoder_seq_len)
+            output     = model(audio_tensor, tok, enc_mask.to(DEVICE), dec_mask.to(DEVICE), cross_mask.to(DEVICE), DEVICE)
+            output     = F.softmax(output, dim=-1)
+            next_token = torch.argmax(output[0, i, :], dim=-1)
+            ch         = idx2char.get(next_token.item(), '')
+            if ch == '<EOS>': break
+            transcription += '' if ch in ('<PAD>', '<SOS>', '<UNK>') else ch
+    return transcription
+
+
+# ─────────────────────────────────────────────
+# FastAPI
+# ─────────────────────────────────────────────
+app = FastAPI(
+    title="Arabic Speech-to-Text API",
+    description="""
+## 🎙️ Arabic Speech-to-Text (Egyptian Dialect)
+
+Upload an audio file and get the Arabic transcription.
+
+### Notes
+- Supports **WAV / MP3 / FLAC / OGG**
+- Best results with **Egyptian Arabic** speech
+- Max audio duration: **15 seconds** (longer files are trimmed)
+- The model performs noise reduction automatically
+""",
+    version="1.0.0",
+)
+
+
+@app.on_event("startup")
+async def startup_event():
+    load_model()
+
+
+@app.get("/", tags=["Health"])
+def root():
+    """Health check endpoint."""
+    return {"status": "ok", "model_loaded": model is not None}
+
+
+@app.get("/vocab", tags=["Info"])
+def get_vocab():
+    """Return the vocabulary used by the model."""
+    return {"vocab_size": VOCAB_SIZE, "vocab": vocab}
+
+
+@app.post("/transcribe", tags=["Transcription"])
+async def transcribe(file: UploadFile = File(..., description="Audio file (wav / mp3 / flac / ogg)")):
+    """
+    Upload an audio file and receive the Arabic transcription.
+
+    - **file**: audio file to transcribe
+    """
+    if model is None:
+        raise HTTPException(status_code=503, detail="Model not loaded. Make sure 'entire_model.pth' is present.")
+
+    allowed = {"audio/wav", "audio/wave", "audio/x-wav", "audio/mpeg",
+               "audio/mp3", "audio/flac", "audio/ogg", "audio/x-flac"}
+    if file.content_type and file.content_type not in allowed:
+        raise HTTPException(status_code=415, detail=f"Unsupported media type: {file.content_type}")
+
+    suffix = os.path.splitext(file.filename)[-1] if file.filename else ".wav"
+    with tempfile.NamedTemporaryFile(delete=False, suffix=suffix) as tmp:
+        tmp.write(await file.read())
+        tmp_path = tmp.name
+
+    try:
+        mel, orig_len, duration = preprocess_audio(tmp_path)
+    except Exception as e:
+        os.unlink(tmp_path)
+        raise HTTPException(status_code=400, detail=f"Audio processing error: {e}")
+    finally:
+        if os.path.exists(tmp_path):
+            os.unlink(tmp_path)
+
+    audio_tensor = torch.tensor(mel, dtype=torch.float32).unsqueeze(0)  # (1, n_mels, n_frames)
+
+    try:
+        result = greedy_decode(audio_tensor, orig_len)
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=f"Inference error: {e}")
+
+    gc.collect()
+    return JSONResponse({
+        "transcription": result,
+        "duration_seconds": round(duration, 2),
+        "original_frames": orig_len,
+    })
+
+
+if __name__ == "__main__":
+    uvicorn.run("app:app", host="0.0.0.0", port=7860, reload=False)

app.py

@@ -0,0 +1,448 @@
+import os
+import math
+import gc
+import re
+import string
+import tempfile
+
+import numpy as np
+import torch
+import torch.nn as nn
+import torch.nn.functional as F
+import librosa
+import noisereduce as nr
+from num2words import num2words
+
+from fastapi import FastAPI, File, UploadFile, HTTPException
+from fastapi.responses import JSONResponse
+import uvicorn
+
+# ─────────────────────────────────────────────
+# Constants
+# ─────────────────────────────────────────────
+N_MELS        = 128
+SAMPLE_RATE   = 16000
+HOP_LENGTH    = 512
+N_FFT         = 2048
+CHUNK_LENGTH  = 15
+N_SAMPLES     = SAMPLE_RATE * CHUNK_LENGTH
+N_FRAMES      = math.ceil(N_SAMPLES / HOP_LENGTH)
+MAX_TEXT_LEN  = 70
+MAX_SEQ_LEN   = 70
+NEG_INFTY     = -1e9
+
+MODEL_PATH = "entire_model.pth"   # ضع وزن الموديل هنا
+
+# ─────────────────────────────────────────────
+# Vocabulary
+# ─────────────────────────────────────────────
+special_tokens    = ['<PAD>', '<UNK>', '<SOS>', '<EOS>']
+english_chars     = list(string.ascii_lowercase + ' ')
+arabic_chars      = list("ابتثجحخدذرزسشصضطظعغفقكلمنهويئءىةؤ")
+vocab             = special_tokens + english_chars + arabic_chars
+char2idx          = {c: i for i, c in enumerate(vocab)}
+idx2char          = {i: c for i, c in enumerate(vocab)}
+VOCAB_SIZE        = len(vocab)
+
+# ─────────────────────────────────────────────
+# Model Architecture (same as notebook)
+# ─────────────────────────────────────────────
+class MultiHeadAttention(nn.Module):
+    def __init__(self, d_model, num_heads):
+        super().__init__()
+        assert d_model % num_heads == 0
+        self.d_model   = d_model
+        self.num_heads = num_heads
+        self.head_dim  = d_model // num_heads
+        self.qkv_layer = nn.Linear(d_model, 3 * d_model)
+        self.linear    = nn.Linear(d_model, d_model)
+
+    def forward(self, x, mask=None):
+        B, L, _ = x.shape
+        qkv = self.qkv_layer(x).reshape(B, L, self.num_heads, 3 * self.head_dim)
+        qkv = qkv.permute(0, 2, 1, 3)
+        q, k, v = qkv.chunk(3, dim=-1)
+        attn = torch.matmul(q, k.transpose(-2, -1)) / math.sqrt(self.head_dim)
+        if mask is not None:
+            attn += mask
+        attn = F.softmax(attn, dim=-1)
+        out  = torch.matmul(attn, v).permute(0, 2, 1, 3).reshape(B, L, self.d_model)
+        return self.linear(out)
+
+
+class MultiHeadCrossAttention(nn.Module):
+    def __init__(self, d_model, num_heads):
+        super().__init__()
+        assert d_model % num_heads == 0
+        self.d_model   = d_model
+        self.num_heads = num_heads
+        self.head_dim  = d_model // num_heads
+        self.kv_layer  = nn.Linear(d_model, 2 * d_model)
+        self.q_layer   = nn.Linear(d_model, d_model)
+        self.linear    = nn.Linear(d_model, d_model)
+
+    def forward(self, x, y, mask=None):
+        B, L, _ = x.shape
+        kv = self.kv_layer(y).reshape(B, y.shape[1], self.num_heads, 2 * self.head_dim).permute(0, 2, 1, 3)
+        k, v = kv.chunk(2, dim=-1)
+        q    = self.q_layer(x).reshape(B, L, self.num_heads, self.head_dim).permute(0, 2, 1, 3)
+        attn = torch.matmul(q, k.transpose(-2, -1)) / math.sqrt(self.head_dim)
+        if mask is not None:
+            attn += mask
+        attn = F.softmax(attn, dim=-1)
+        out  = torch.matmul(attn, v).permute(0, 2, 1, 3).reshape(B, L, self.d_model)
+        return self.linear(out)
+
+
+class LayerNormalization(nn.Module):
+    def __init__(self, d_model, eps=1e-5):
+        super().__init__()
+        self.gamma = nn.Parameter(torch.ones(d_model))
+        self.beta  = nn.Parameter(torch.zeros(d_model))
+        self.eps   = eps
+
+    def forward(self, x):
+        mean = x.mean(-1, keepdim=True)
+        var  = ((x - mean) ** 2).mean(-1, keepdim=True)
+        return self.gamma * (x - mean) / torch.sqrt(var + self.eps) + self.beta
+
+
+class PositionwiseFeedForward(nn.Module):
+    def __init__(self, d_model, hidden, drop_prob=0.1):
+        super().__init__()
+        self.net = nn.Sequential(
+            nn.Linear(d_model, hidden), nn.ReLU(), nn.Dropout(drop_prob), nn.Linear(hidden, d_model)
+        )
+    def forward(self, x): return self.net(x)
+
+
+class EncoderLayer(nn.Module):
+    def __init__(self, d_model, ffn_hidden, num_heads, drop_prob):
+        super().__init__()
+        self.attn  = MultiHeadAttention(d_model, num_heads)
+        self.norm1 = LayerNormalization(d_model)
+        self.ff    = PositionwiseFeedForward(d_model, ffn_hidden, drop_prob)
+        self.norm2 = LayerNormalization(d_model)
+        self.drop  = nn.Dropout(drop_prob)
+
+    def forward(self, x, mask=None):
+        x = self.norm1(x + self.drop(self.attn(x, mask)))
+        x = self.norm2(x + self.drop(self.ff(x)))
+        return x
+
+
+class DecoderLayer(nn.Module):
+    def __init__(self, d_model, ffn_hidden, num_heads, drop_prob):
+        super().__init__()
+        self.self_attn  = MultiHeadAttention(d_model, num_heads)
+        self.norm1      = LayerNormalization(d_model)
+        self.cross_attn = MultiHeadCrossAttention(d_model, num_heads)
+        self.norm2      = LayerNormalization(d_model)
+        self.ff         = PositionwiseFeedForward(d_model, ffn_hidden, drop_prob)
+        self.norm3      = LayerNormalization(d_model)
+        self.drop       = nn.Dropout(drop_prob)
+
+    def forward(self, x, enc_out, self_mask=None, cross_mask=None):
+        x = self.norm1(x + self.drop(self.self_attn(x, self_mask)))
+        x = self.norm2(x + self.drop(self.cross_attn(x, enc_out, cross_mask)))
+        x = self.norm3(x + self.drop(self.ff(x)))
+        return x
+
+
+class SequentialEncoder(nn.Sequential):
+    def forward(self, *args):
+        x, mask = args
+        for layer in self:
+            x = layer(x, mask)
+        return x
+
+
+class SequentialDecoder(nn.Sequential):
+    def forward(self, *args):
+        x, enc_out, self_mask, cross_mask = args
+        for layer in self:
+            x = layer(x, enc_out, self_mask, cross_mask)
+        return x
+
+
+class Encoder(nn.Module):
+    def __init__(self, d_model, ffn_hidden, num_heads, drop_prob, num_layers):
+        super().__init__()
+        self.layers = SequentialEncoder(*[EncoderLayer(d_model, ffn_hidden, num_heads, drop_prob) for _ in range(num_layers)])
+
+    def forward(self, x, mask=None):
+        return self.layers(x, mask)
+
+
+class Decoder(nn.Module):
+    def __init__(self, d_model, ffn_hidden, num_heads, drop_prob, num_layers):
+        super().__init__()
+        self.layers = SequentialDecoder(*[DecoderLayer(d_model, ffn_hidden, num_heads, drop_prob) for _ in range(num_layers)])
+
+    def forward(self, x, enc_out, self_mask=None, cross_mask=None):
+        return self.layers(x, enc_out, self_mask, cross_mask)
+
+
+class Transformer(nn.Module):
+    def __init__(self, d_model=512, num_heads=8, num_encoder_layers=6, num_decoder_layers=6, ffn_hidden=2048, drop_prob=0.1):
+        super().__init__()
+        self.encoder = Encoder(d_model, ffn_hidden, num_heads, drop_prob, num_encoder_layers)
+        self.decoder = Decoder(d_model, ffn_hidden, num_heads, drop_prob, num_decoder_layers)
+
+    def forward(self, src, tgt, encoder_self_attention_mask=None, decoder_self_attention_mask=None, decoder_cross_attention_mask=None):
+        src = self.encoder(src, encoder_self_attention_mask)
+        return self.decoder(tgt, src, decoder_self_attention_mask, decoder_cross_attention_mask)
+
+
+class PositionalEncoding(nn.Module):
+    def __init__(self, d_model, max_seq_len):
+        super().__init__()
+        pe  = torch.zeros(max_seq_len, d_model)
+        pos = torch.arange(max_seq_len, dtype=torch.float).unsqueeze(1)
+        div = torch.exp(torch.arange(0, d_model, 2).float() * (-math.log(10000.0) / d_model))
+        pe[:, 0::2] = torch.sin(pos * div)
+        pe[:, 1::2] = torch.cos(pos * div)
+        self.register_buffer('pe', pe.unsqueeze(0))
+
+    def forward(self): return self.pe
+
+
+def get_conv_Lout(L_in, conv):
+    return math.floor((L_in + 2 * conv.padding[0] - conv.dilation[0] * (conv.kernel_size[0] - 1) - 1) / conv.stride[0] + 1)
+
+
+class MMS(nn.Module):
+    def __init__(self, vocab_size, d_model=512, nhead=8, num_encoder_layers=6, num_decoder_layers=6,
+                 dim_feedforward=2048, max_encoder_seq_len=100, max_decoder_seq_len=100, n_mels=N_MELS, dropout=0.1):
+        super().__init__()
+        self.transformer = Transformer(d_model, nhead, num_encoder_layers, num_decoder_layers, dim_feedforward, dropout)
+        self.en_positional_encoding = PositionalEncoding(d_model, max_encoder_seq_len)
+        self.de_positional_encoding = PositionalEncoding(d_model, max_decoder_seq_len)
+        self.conv1     = nn.Conv1d(n_mels, d_model, kernel_size=3, padding=1)
+        self.conv2     = nn.Conv1d(d_model, d_model, kernel_size=3, stride=2, padding=1)
+        self.gelu      = nn.GELU()
+        self.embedding = nn.Embedding(vocab_size, d_model, padding_idx=0)
+        self.d_model   = d_model
+        self.ff        = nn.Linear(d_model, vocab_size)
+
+    def get_encoder_seq_len(self, L_in):
+        return get_conv_Lout(get_conv_Lout(L_in, self.conv1), self.conv2)
+
+    def forward(self, audio, text, enc_mask, dec_mask, cross_mask, device):
+        audio = self.gelu(self.conv1(audio))
+        audio = self.gelu(self.conv2(audio))
+        audio = audio.permute(0, 2, 1)
+        audio += self.en_positional_encoding().to(device)
+        text   = self.embedding(text) + self.de_positional_encoding().to(device)
+        out    = self.transformer(audio, text, enc_mask, dec_mask, cross_mask)
+        return self.ff(out)
+
+# ─────────────────────────────────────────────
+# Helpers
+# ─────────────────────────────────────────────
+def pad_or_trim(array, length=N_SAMPLES, axis=-1):
+    if array.shape[axis] > length:
+        array = array.take(indices=range(length), axis=axis)
+    if array.shape[axis] < length:
+        pad_widths        = [(0, 0)] * array.ndim
+        pad_widths[axis]  = (0, length - array.shape[axis])
+        array             = np.pad(array, pad_widths)
+    return array
+
+
+def preprocess_audio(file_path):
+    audio_data, _ = librosa.load(file_path, sr=SAMPLE_RATE)
+    audio_data    = nr.reduce_noise(y=audio_data, sr=SAMPLE_RATE)
+    duration      = librosa.get_duration(y=audio_data, sr=SAMPLE_RATE)
+    modified      = pad_or_trim(audio_data, padding=False)
+    sgram         = librosa.stft(y=modified, n_fft=N_FFT, hop_length=HOP_LENGTH)
+    sgram_mag, _  = librosa.magphase(sgram)
+    mel           = librosa.feature.melspectrogram(S=sgram_mag, sr=SAMPLE_RATE, n_fft=N_FFT, hop_length=HOP_LENGTH, n_mels=N_MELS)
+    mel_db        = librosa.amplitude_to_db(mel, ref=np.min)
+    orig_len      = mel_db.shape[-1]
+    padded        = np.pad(mel_db, ((0, 0), (0, N_FRAMES - mel_db.shape[-1]))) if orig_len < N_FRAMES else mel_db[:, :N_FRAMES]
+    return padded, orig_len, duration
+
+
+def nlp_preprocessing(sentence):
+    sentence = sentence.lower().replace("\n", " ")
+    sentence = re.sub(r'[إأآ]', 'ا', sentence)
+    sentence = re.sub(r'[^a-zA-Zء-ي\s\d]', '', sentence)
+    sentence = re.sub(r'[\u0617-\u061A\u064B-\u065F]', '', sentence)
+    sentence = re.sub(r'([a-zA-Z])([ء-ي])|([ء-ي])([a-zA-Z])', r'\1\3 \2\4', sentence)
+    sentence = re.sub(r'\s+', ' ', sentence)
+    return re.sub(r'\d+', lambda x: num2words(int(x.group()), lang='ar'), sentence)
+
+
+def tokenize_text(text, max_len=MAX_SEQ_LEN, start_token=True, end_token=True):
+    tokens  = [char2idx.get(c, char2idx['<UNK>']) for c in text]
+    max_len -= (start_token + end_token)
+    text_len = len(tokens)
+    if text_len < max_len:
+        if end_token: tokens += [char2idx['<EOS>']]
+        tokens += [char2idx['<PAD>']] * (max_len - text_len)
+    else:
+        tokens = tokens[:max_len]
+        if end_token: tokens += [char2idx['<EOS>']]
+    if start_token: tokens.insert(0, char2idx['<SOS>'])
+    return tokens
+
+
+def text_decoder(tokens):
+    result = ""
+    for t in tokens.tolist() if hasattr(tokens, 'tolist') else [tokens.item()]:
+        ch = idx2char.get(t, '')
+        if ch in ('<PAD>', '<SOS>', '<EOS>', '<UNK>'): break
+        result += ch
+    return result
+
+
+def generate_padding_masks(transcription, audio_original_len, conv_func, frames=N_FRAMES):
+    batch_size, seq_len = transcription.size()
+    audio_len           = conv_func(frames)
+    look_ahead_mask     = torch.triu(torch.full((seq_len, seq_len), True), diagonal=1)
+    enc_mask            = torch.full([batch_size, audio_len, audio_len], False)
+    dec_self_mask       = torch.full([batch_size, seq_len, seq_len], False)
+    dec_cross_mask      = torch.full([batch_size, seq_len, audio_len], False)
+
+    for i in range(batch_size):
+        new_len = conv_func(audio_original_len[i])
+        enc_mask[i, new_len:, :]  = True
+        enc_mask[i, :, new_len:]  = True
+        dec_cross_mask[i, :, new_len:] = True
+        zero_idx = np.where(transcription[0].cpu().numpy() == 0)[0]
+        if len(zero_idx):
+            idx = zero_idx[0]
+            dec_self_mask[i, idx:, :]  = True
+            dec_self_mask[i, :, idx:]  = True
+            dec_cross_mask[i, idx:, :] = True
+
+    dec_self_attn  = torch.where(look_ahead_mask | dec_self_mask, NEG_INFTY, 0.0)
+    dec_cross_attn = torch.where(dec_cross_mask, NEG_INFTY, 0.0)
+    enc_attn       = torch.where(enc_mask, NEG_INFTY, 0.0)
+    return enc_attn, dec_self_attn, dec_cross_attn
+
+
+# ─────────────────────────────────────────────
+# Load model
+# ─────────────────────────────────────────────
+DEVICE = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
+model  = None
+
+def load_model():
+    global model
+    if not os.path.exists(MODEL_PATH):
+        print(f"⚠️  Model file '{MODEL_PATH}' not found — transcription endpoint will return an error.")
+        return
+    try:
+        model = torch.load(MODEL_PATH, map_location=DEVICE, weights_only=False)
+        model.to(DEVICE)
+        model.eval()
+        print("✅ Model loaded successfully.")
+    except Exception as e:
+        print(f"❌ Failed to load model: {e}")
+
+
+# ─────────────────────────────────────────────
+# Greedy decoder
+# ─────────────────────────────────────────────
+def greedy_decode(audio_tensor, org_len):
+    transcription = ""
+    audio_tensor  = audio_tensor.to(DEVICE)
+    with torch.no_grad():
+        for i in range(MAX_TEXT_LEN):
+            tok = torch.tensor(tokenize_text(transcription, max_len=MAX_TEXT_LEN, end_token=False), dtype=torch.long)
+            tok = tok.unsqueeze(0).to(DEVICE)
+            enc_mask, dec_mask, cross_mask = generate_padding_masks(tok, [org_len], model.get_encoder_seq_len)
+            output     = model(audio_tensor, tok, enc_mask.to(DEVICE), dec_mask.to(DEVICE), cross_mask.to(DEVICE), DEVICE)
+            output     = F.softmax(output, dim=-1)
+            next_token = torch.argmax(output[0, i, :], dim=-1)
+            ch         = idx2char.get(next_token.item(), '')
+            if ch == '<EOS>': break
+            transcription += '' if ch in ('<PAD>', '<SOS>', '<UNK>') else ch
+    return transcription
+
+
+# ─────────────────────────────────────────────
+# FastAPI
+# ─────────────────────────────────────────────
+app = FastAPI(
+    title="Arabic Speech-to-Text API",
+    description="""
+## 🎙️ Arabic Speech-to-Text (Egyptian Dialect)
+
+Upload an audio file and get the Arabic transcription.
+
+### Notes
+- Supports **WAV / MP3 / FLAC / OGG**
+- Best results with **Egyptian Arabic** speech
+- Max audio duration: **15 seconds** (longer files are trimmed)
+- The model performs noise reduction automatically
+""",
+    version="1.0.0",
+)
+
+
+@app.on_event("startup")
+async def startup_event():
+    load_model()
+
+
+@app.get("/", tags=["Health"])
+def root():
+    """Health check endpoint."""
+    return {"status": "ok", "model_loaded": model is not None}
+
+
+@app.get("/vocab", tags=["Info"])
+def get_vocab():
+    """Return the vocabulary used by the model."""
+    return {"vocab_size": VOCAB_SIZE, "vocab": vocab}
+
+
+@app.post("/transcribe", tags=["Transcription"])
+async def transcribe(file: UploadFile = File(..., description="Audio file (wav / mp3 / flac / ogg)")):
+    """
+    Upload an audio file and receive the Arabic transcription.
+
+    - **file**: audio file to transcribe
+    """
+    if model is None:
+        raise HTTPException(status_code=503, detail="Model not loaded. Make sure 'entire_model.pth' is present.")
+
+    allowed = {"audio/wav", "audio/wave", "audio/x-wav", "audio/mpeg",
+               "audio/mp3", "audio/flac", "audio/ogg", "audio/x-flac"}
+    if file.content_type and file.content_type not in allowed:
+        raise HTTPException(status_code=415, detail=f"Unsupported media type: {file.content_type}")
+
+    suffix = os.path.splitext(file.filename)[-1] if file.filename else ".wav"
+    with tempfile.NamedTemporaryFile(delete=False, suffix=suffix) as tmp:
+        tmp.write(await file.read())
+        tmp_path = tmp.name
+
+    try:
+        mel, orig_len, duration = preprocess_audio(tmp_path)
+    except Exception as e:
+        os.unlink(tmp_path)
+        raise HTTPException(status_code=400, detail=f"Audio processing error: {e}")
+    finally:
+        if os.path.exists(tmp_path):
+            os.unlink(tmp_path)
+
+    audio_tensor = torch.tensor(mel, dtype=torch.float32).unsqueeze(0)  # (1, n_mels, n_frames)
+
+    try:
+        result = greedy_decode(audio_tensor, orig_len)
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=f"Inference error: {e}")
+
+    gc.collect()
+    return JSONResponse({
+        "transcription": result,
+        "duration_seconds": round(duration, 2),
+        "original_frames": orig_len,
+    })
+
+
+if __name__ == "__main__":
+    uvicorn.run("app:app", host="0.0.0.0", port=7860, reload=False)

requirements.txt

@@ -0,0 +1,11 @@
+fastapi==0.111.0
+uvicorn==0.29.0
+python-multipart==0.0.9
+torch==2.2.2
+torchaudio==2.2.2
+librosa==0.10.2
+noisereduce==3.0.2
+num2words==0.5.13
+numpy==1.26.4
+pydub==0.25.1
+soundfile==0.12.1