app/services/plate_recognition.py

import base64
import os
import tempfile
import logging
from typing import Any, Dict, List, Optional, Tuple

import cv2
import numpy as np
from fastapi import HTTPException

from app.core.config import MODEL_SERVICE_URL, MODEL_SERVICE_TIMEOUT_SECONDS, HF_SPACE_ID, LOGGER

LOGGER = logging.getLogger("ain_el_aql.plate_recognition")

CHAR_MAP = {
    "aain": ("ع", "E"), "alef": ("أ", "A"), "a": ("أ", "A"), "alf": ("أ", "A"),
    "baa": ("ب", "B"), "daal": ("د", "D"), "dal": ("د", "D"), "e": ("ع", "E"),
    "faa": ("ف", "F"), "geem": ("ج", "G"), "haa": ("هـ", "H"), "kaaf": ("ك", "K"),
    "laam": ("ل", "L"), "meem": ("م", "M"), "noon": ("ن", "N"), "qaf": ("ق", "Q"),
    "raa": ("ر", "R"), "sad": ("ص", "S"), "seen": ("س", "C"), "taa": ("ط", "T"),
    "waaw": ("و", "W"), "waw": ("و", "W"), "yaa": ("ى", "Y"), "zay": ("ز", "Z"),
    "dad": ("ض", "DD"),
    "0": ("0", "0"), "1": ("1", "1"), "2": ("2", "2"), "3": ("3", "3"),
    "4": ("4", "4"), "5": ("5", "5"), "6": ("6", "6"), "7": ("7", "7"),
    "8": ("8", "8"), "9": ("9", "9"),
    "٠": ("0", "0"), "١": ("1", "1"), "٢": ("2", "2"), "٣": ("3", "3"),
    "٤": ("4", "4"), "٥": ("5", "5"), "٦": ("6", "6"), "٧": ("7", "7"),
    "٨": ("8", "8"), "٩": ("9", "9"),
}


def _decode_image_bytes(image_bytes: bytes) -> np.ndarray:
    image_np = np.frombuffer(image_bytes, dtype=np.uint8)
    image_bgr = cv2.imdecode(image_np, cv2.IMREAD_COLOR)
    if image_bgr is None:
        raise HTTPException(status_code=400, detail="Unable to decode image.")
    return image_bgr


def _encode_image_base64(image_bgr: np.ndarray) -> str:
    ok, encoded = cv2.imencode(".jpg", image_bgr)
    if not ok:
        raise HTTPException(status_code=500, detail="Failed to encode image.")
    return base64.b64encode(encoded.tobytes()).decode("utf-8")


def _fit_into_canvas(image_bgr: np.ndarray, target_w: int, target_h: int) -> np.ndarray:
    canvas = np.full((target_h, target_w, 3), 18, dtype=np.uint8)
    if image_bgr.size == 0:
        return canvas
    src_h, src_w = image_bgr.shape[:2]
    scale = min(target_w / max(1, src_w), target_h / max(1, src_h))
    new_w = max(1, int(src_w * scale))
    new_h = max(1, int(src_h * scale))
    resized = cv2.resize(image_bgr, (new_w, new_h), interpolation=cv2.INTER_AREA)
    x_off = (target_w - new_w) // 2
    y_off = (target_h - new_h) // 2
    canvas[y_off: y_off + new_h, x_off: x_off + new_w] = resized
    return canvas


def _compose_user_split_image(plate_focus_bgr: np.ndarray, car_focus_bgr: np.ndarray) -> np.ndarray:
    half_h = max(plate_focus_bgr.shape[0], car_focus_bgr.shape[0], 220)
    half_w = max(plate_focus_bgr.shape[1], car_focus_bgr.shape[1], 320)
    left_half = _fit_into_canvas(plate_focus_bgr, half_w, half_h)
    right_half = _fit_into_canvas(car_focus_bgr, half_w, half_h)
    return np.concatenate([left_half, right_half], axis=1)


def _clamp_bbox(x1: float, y1: float, x2: float, y2: float, width: int, height: int) -> Tuple[int, int, int, int]:
    left = max(0, min(int(x1), width - 1))
    top = max(0, min(int(y1), height - 1))
    right = max(1, min(int(x2), width))
    bottom = max(1, min(int(y2), height))
    if right <= left:
        right = min(width, left + 1)
    if bottom <= top:
        bottom = min(height, top + 1)
    return left, top, right, bottom


def _build_plate_placeholder(reference_bgr: np.ndarray) -> np.ndarray:
    placeholder = np.full(reference_bgr.shape, 18, dtype=np.uint8)
    h, w = placeholder.shape[:2]
    text = "NO PLATE DETECTED"
    font_scale = 0.8 if w >= 500 else 0.6
    thickness = 2
    text_size, _ = cv2.getTextSize(text, cv2.FONT_HERSHEY_SIMPLEX, font_scale, thickness)
    text_x = max(10, (w - text_size[0]) // 2)
    text_y = max(28, h // 2)
    cv2.putText(placeholder, text, (text_x, text_y), cv2.FONT_HERSHEY_SIMPLEX, font_scale, (0, 220, 220), thickness, cv2.LINE_AA)
    return placeholder


def decode_ocr_result(result: Any) -> Dict[str, Any]:
    if result.boxes is None or len(result.boxes) == 0:
        return {"raw_ordered_labels": [], "characters": [], "arabic": "N/A", "english": "N/A"}
    names = result.names if hasattr(result, "names") else {}
    detections: List[Dict[str, Any]] = []
    for box in result.boxes:
        cls_idx = int(box.cls[0].item()) if box.cls is not None else -1
        raw_label = names.get(cls_idx, str(cls_idx)) if isinstance(names, dict) else str(cls_idx)
        norm_label = str(raw_label).strip().lower()
        xyxy = box.xyxy[0].tolist()
        confidence = float(box.conf[0].item()) if box.conf is not None else 0.0
        center_x = (xyxy[0] + xyxy[2]) / 2.0
        ar_char, en_char = CHAR_MAP.get(norm_label, (str(raw_label), str(raw_label)))
        detections.append({
            "label": str(raw_label),
            "normalized_label": norm_label,
            "arabic": ar_char,
            "english": en_char,
            "is_digit": norm_label.isdigit(),
            "confidence": round(confidence, 4),
            "bbox": [int(xyxy[0]), int(xyxy[1]), int(xyxy[2]), int(xyxy[3])],
            "center_x": center_x,
        })
    letter_detections = [d for d in detections if not d["is_digit"]]
    number_detections = [d for d in detections if d["is_digit"]]
    letter_detections.sort(key=lambda item: item["center_x"], reverse=True)
    number_detections.sort(key=lambda item: item["center_x"], reverse=False)
    ar_letters = [d["arabic"] for d in letter_detections]
    ar_numbers = [d["arabic"] for d in number_detections]
    en_letters = [d["english"] for d in letter_detections]
    en_numbers = [d["english"] for d in number_detections]
    arabic_text = f"{' '.join(ar_letters)} | {' '.join(ar_numbers)}" if (ar_letters or ar_numbers) else "N/A"
    english_text = f"{' '.join(en_letters)} | {''.join(en_numbers)}" if (en_letters or en_numbers) else "N/A"
    clean_chars = [{"label": item["label"], "arabic": item["arabic"], "english": item["english"], "confidence": item["confidence"], "bbox": item["bbox"]} for item in detections]
    return {"raw_ordered_labels": [item["label"] for item in detections], "characters": clean_chars, "arabic": arabic_text, "english": english_text}


def run_pipeline_remote(*, image_bytes: bytes, filename: Optional[str] = None, content_type: Optional[str] = None) -> Dict[str, Any]:
    try:
        from gradio_client import Client, handle_file
    except ImportError:
        raise HTTPException(status_code=500, detail="gradio_client is not installed.")

    # Monkey-patch gradio_client schema parsing bug where boolean additionalProperties throws TypeError
    try:
        import gradio_client.utils
        _orig_parser = gradio_client.utils._json_schema_to_python_type
        if not hasattr(gradio_client.utils, "_safe_parser_applied"):
            def _safe_parser(schema, defs):
                if isinstance(schema, bool) or not isinstance(schema, dict):
                    return "Any"
                return _orig_parser(schema, defs)
            gradio_client.utils._json_schema_to_python_type = _safe_parser
            gradio_client.utils._safe_parser_applied = True
    except Exception:
        pass

    with tempfile.NamedTemporaryFile(delete=False, suffix=".jpg") as tmp:
        tmp.write(image_bytes)
        tmp_path = tmp.name
    try:
        client = Client(HF_SPACE_ID)
        result = client.predict(img=handle_file(tmp_path), api_name="/predict_plate")
        if not isinstance(result, tuple) or len(result) < 3:
            raise HTTPException(status_code=502, detail=f"Unexpected response from HF: {result}")
        annotated_img_path = result[0]
        numbers_ar = str(result[1] or "").strip()
        letters_ar = str(result[2] or "").strip()
        arabic_text = f"{letters_ar} | {numbers_ar}" if (letters_ar or numbers_ar) else "N/A"
        _ar_to_en = {v[0]: v[1] for v in CHAR_MAP.values()}
        letters_en_parts = [_ar_to_en.get(ch, ch) for ch in letters_ar.split(" ") if ch]
        numbers_en_parts = [_ar_to_en.get(ch, ch) for ch in numbers_ar.split(" ") if ch]
        letters_en = " ".join(letters_en_parts)
        numbers_en = "".join(numbers_en_parts)
        english_text = f"{letters_en} | {numbers_en}" if (letters_en or numbers_en) else "N/A"
        annotated_b64 = ""
        if annotated_img_path and os.path.exists(str(annotated_img_path)):
            with open(str(annotated_img_path), "rb") as f:
                annotated_b64 = base64.b64encode(f.read()).decode("utf-8")
        return {
            "plate_info": {"arabic": arabic_text, "english": english_text, "characters": [], "raw_ordered_labels": []},
            "user_page": {},
            "admin_page": {"annotated_image_base64": annotated_b64},
        }
    except HTTPException:
        raise
    except Exception as exc:
        LOGGER.exception("Remote model service failed")
        import traceback
        tb = traceback.format_exc()
        raise HTTPException(status_code=502, detail=f"Remote model service failed: {exc}\nTraceback:\n{tb}") from exc
    finally:
        if os.path.exists(tmp_path):
            try:
                os.remove(tmp_path)
            except OSError:
                pass