app.py

import os
import cv2
import uuid
import base64
import numpy as np
import tensorflow as tf
import mediapipe as mp

from fastapi import FastAPI, File, UploadFile, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel
from typing import Optional
from contextlib import asynccontextmanager


#  PATHS
BASE_DIR = os.path.dirname(os.path.abspath(__file__))

ENGLISH_MODEL_PATH = os.path.join(BASE_DIR, "models", "asl_landmark_cnn_lstm_model.keras")
ARABIC_MODEL_PATH  = os.path.join(BASE_DIR, "models", "arsl_landmark_cnn_lstm_model.keras")


#  CLASSES
ENGLISH_CLASSES = list("ABCDEFGHIJKLMNOPQRSTUVWXYZ") + ["del", "space"]

ARABIC_CLASSES = [
    "Ain", "Al", "Alef", "Beh", "Dad", "Dal", "Feh", "Ghain", "Hah", "Heh",
    "Jeem", "Kaf", "Khah", "Laa", "Lam", "Meem", "Noon", "Qaf", "Reh", "Sad",
    "Seen", "Sheen", "Tah", "Teh", "Teh_Marbuta", "Thal", "Theh", "Waw", "Yeh",
    "Zah", "Zain", "del", "space"
]

ARABIC_MAP = {
    "Alef": "ا", "Beh": "ب", "Teh": "ت", "Theh": "ث", "Jeem": "ج",
    "Hah": "ح", "Khah": "خ", "Dal": "د", "Thal": "ذ", "Reh": "ر",
    "Zain": "ز", "Seen": "س", "Sheen": "ش", "Sad": "ص", "Dad": "ض",
    "Tah": "ط", "Zah": "ظ", "Ain": "ع", "Ghain": "غ", "Feh": "ف",
    "Qaf": "ق", "Kaf": "ك", "Lam": "ل", "Meem": "م", "Noon": "ن",
    "Heh": "ه", "Waw": "و", "Yeh": "ي", "Laa": "لا", "Al": "ال",
    "Teh_Marbuta": "ة"
}


#  CONFIG
TIMESTEPS            = 23
CONFIDENCE_THRESHOLD = 0.8


#  GLOBAL MODELS & MEDIAPIPE
english_model = None
arabic_model  = None
hands_detector = None


@asynccontextmanager
async def lifespan(app: FastAPI):
    global english_model, arabic_model, hands_detector

    print("Loading models...")

    if os.path.exists(ENGLISH_MODEL_PATH):
        english_model = tf.keras.models.load_model(ENGLISH_MODEL_PATH)
        print(" English model loaded")
    else:
        print(f"  English model not found at {ENGLISH_MODEL_PATH}")

    if os.path.exists(ARABIC_MODEL_PATH):
        arabic_model = tf.keras.models.load_model(ARABIC_MODEL_PATH)
        print(" Arabic model loaded")
    else:
        print(f"  Arabic model not found at {ARABIC_MODEL_PATH}")

    mp_hands = mp.solutions.hands
    hands_detector = mp_hands.Hands(
        static_image_mode=True,
        max_num_hands=1,
        min_detection_confidence=0.5
    )
    print(" MediaPipe ready")

    yield

    # cleanup
    if hands_detector:
        hands_detector.close()
    print("Shutting down...")



#  APP
app = FastAPI(
    title="Sign Language Translator API",
    description="Real-time Arabic & English Sign Language recognition using CNN-BiLSTM + MediaPipe",
    version="1.0.0",
    lifespan=lifespan
)

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_methods=["*"],
    allow_headers=["*"],
)



#  SCHEMAS
class PredictResponse(BaseModel):
    language: str
    predicted_label: str          # raw class name  e.g. "Alef"
    predicted_char: str           # display char    e.g. "ا"
    confidence: float
    above_threshold: bool
    message: str



#  HELPERS
def decode_image(data: bytes) -> np.ndarray:
    """Decode uploaded image bytes → BGR numpy array."""
    arr = np.frombuffer(data, dtype=np.uint8)
    img = cv2.imdecode(arr, cv2.IMREAD_COLOR)
    if img is None:
        raise HTTPException(status_code=400, detail="Cannot decode image. Send a valid JPG/PNG.")
    return img


def extract_landmarks(img_bgr: np.ndarray) -> Optional[np.ndarray]:
    """Run MediaPipe on a BGR image → 63-d landmark vector or None."""
    img_rgb = cv2.cvtColor(img_bgr, cv2.COLOR_BGR2RGB)
    results = hands_detector.process(img_rgb)

    if not results.multi_hand_landmarks:
        return None

    hand = results.multi_hand_landmarks[0]
    coords = []
    for lm in hand.landmark:
        coords.extend([lm.x, lm.y, lm.z])

    lm_arr = np.array(coords, dtype=np.float32)

    # Mirror right hand so model always sees left-hand orientation
    if results.multi_handedness:
        label = results.multi_handedness[0].classification[0].label
        if label == "Right":
            lm_arr[0::3] = 1.0 - lm_arr[0::3]

    return lm_arr


def build_sequence(lm: np.ndarray) -> np.ndarray:
    """Repeat single frame 23× → (1, 23, 63) normalised tensor."""
    x_seq = np.repeat(lm[np.newaxis, :], TIMESTEPS, axis=0)   # (23, 63)
    x_seq = x_seq[np.newaxis, :, :]                            # (1, 23, 63)
    max_val = np.max(np.abs(x_seq))
    if max_val != 0:
        x_seq = x_seq / max_val
    return x_seq


def run_inference(model, x_seq: np.ndarray, classes: list) -> tuple[str, float]:
    """Return (predicted_class_name, confidence)."""
    probs     = model.predict(x_seq, verbose=0)
    idx       = int(np.argmax(probs))
    conf      = float(probs[0][idx])
    label     = classes[idx] if idx < len(classes) else "unknown"
    return label, conf



#  ROUTES
@app.get("/")
def root():
    return {
        "name": "Sign Language Translator API",
        "endpoints": {
            "predict_english": "POST /predict/english",
            "predict_arabic":  "POST /predict/arabic",
            "health":          "GET  /health"
        }
    }


@app.get("/health")
def health():
    return {
        "status": "ok",
        "english_model_loaded": english_model is not None,
        "arabic_model_loaded":  arabic_model  is not None,
    }


@app.post("/predict/english", response_model=PredictResponse)
async def predict_english(file: UploadFile = File(...)):
    if english_model is None:
        raise HTTPException(status_code=503, detail="English model not loaded.")

    img   = decode_image(await file.read())
    lm    = extract_landmarks(img)

    if lm is None:
        raise HTTPException(status_code=422, detail="No hand detected in the image.")

    x_seq        = build_sequence(lm)
    label, conf  = run_inference(english_model, x_seq, ENGLISH_CLASSES)

    return PredictResponse(
        language        = "english",
        predicted_label = label,
        predicted_char  = label,           # same for English
        confidence      = round(conf, 4),
        above_threshold = conf >= CONFIDENCE_THRESHOLD,
        message         = "OK" if conf >= CONFIDENCE_THRESHOLD else f"Low confidence ({conf:.2f})"
    )


@app.post("/predict/arabic", response_model=PredictResponse)
async def predict_arabic(file: UploadFile = File(...)):
    if arabic_model is None:
        raise HTTPException(status_code=503, detail="Arabic model not loaded.")

    img   = decode_image(await file.read())
    lm    = extract_landmarks(img)

    if lm is None:
        raise HTTPException(status_code=422, detail="No hand detected in the image.")

    x_seq        = build_sequence(lm)
    label, conf  = run_inference(arabic_model, x_seq, ARABIC_CLASSES)
    char         = ARABIC_MAP.get(label, label)

    return PredictResponse(
        language        = "arabic",
        predicted_label = label,
        predicted_char  = char,
        confidence      = round(conf, 4),
        above_threshold = conf >= CONFIDENCE_THRESHOLD,
        message         = "OK" if conf >= CONFIDENCE_THRESHOLD else f"Low confidence ({conf:.2f})"
    )