import base64 from datetime import datetime, timezone from typing import Any, Dict, List, Optional, Tuple from uuid import uuid4 from fastapi import APIRouter, File, Form, Header, HTTPException, UploadFile import requests from app.core.config import ( SUPABASE_URL, SUPABASE_SERVICE_ROLE_KEY, SUPABASE_TIMEOUT_SECONDS, SUPABASE_RAW_BUCKET, SUPABASE_PROCESSED_BUCKET, MODEL_SERVICE_HEADER_NAME, supabase_configured, LOGGER, ) from app.security.auth import require_authenticated_user, require_developer_user, _supabase_headers, _normalize_spaces from app.services.plate_recognition import run_pipeline_remote, _encode_image_base64 from app.services.supabase_client import upload_to_supabase_storage, delete_from_supabase_storage from app.api.helpers import ( _normalize_parking_location, _normalize_camera_source, _normalize_event_type, _build_plate_payload, _is_plate_detected, _split_arabic_plate, _split_english_plate, _extract_created_by_user_id, ) TZ_UTC = timezone.utc router = APIRouter(tags=["predict"]) def _normalize_plate_text(text: Optional[str]) -> str: if not text: return "" result = str(text).strip().lower() for prefix in ["car", "vehicle", "license", "plate", "lp", "ar", "en"]: if result.startswith(prefix): result = result[len(prefix):].strip() result = "".join(c for c in result if c.isalnum() or c in "|-") _ARABIC_INDIC = '\u0660\u0661\u0662\u0663\u0664\u0665\u0666\u0667\u0668\u0669' _WESTERN = '0123456789' for ar, en in zip(_ARABIC_INDIC, _WESTERN): result = result.replace(ar, en) return result.strip().upper() def _resolve_vehicle_record_by_plate(plate_text_ar: str, plate_text_en: str, user_id: Optional[str]) -> Optional[Dict[str, Any]]: if not supabase_configured(): return None plate_letters_ar, plate_numbers_ar = _split_arabic_plate(plate_text_ar) plate_letters_en, plate_numbers_en = _split_english_plate(plate_text_en) def _number_variants(num_str: Optional[str]) -> List[str]: if not num_str: return [] raw = num_str.strip() if not raw: return [] no_space = "".join(raw.split()) western = no_space arabic_indic = no_space _AI = '\u0660\u0661\u0662\u0663\u0664\u0665\u0666\u0667\u0668\u0669' _WE = '0123456789' for ar, en in zip(_AI, _WE): western = western.replace(ar, en) arabic_indic = arabic_indic.replace(en, ar) candidates = {raw, no_space, western, arabic_indic} if len(western) > 1: candidates.add(" ".join(list(western))) if len(arabic_indic) > 1: candidates.add(" ".join(list(arabic_indic))) rev_w = western[::-1] rev_a = arabic_indic[::-1] if rev_w != western: candidates.add(rev_w) if rev_a != arabic_indic: candidates.add(rev_a) if len(rev_w) > 1 and rev_w != western: candidates.add(" ".join(list(rev_w))) if len(rev_a) > 1 and rev_a != arabic_indic: candidates.add(" ".join(list(rev_a))) return [v for v in candidates if v] def normalize_for_cmp(text: Optional[str]) -> str: if not text: return "" t = "".join(text.split()).lower() t = t.replace('\u0627', '\u0623').replace('\u0625', '\u0623').replace('\u0622', '\u0623') t = t.replace('\u0629', '\u0647').replace('\u064a', '\u0649') _AI = '\u0660\u0661\u0662\u0663\u0664\u0665\u0666\u0667\u0668\u0669' _WE = '0123456789' for ar, en in zip(_AI, _WE): t = t.replace(ar, en) return t matched_vehicles: List[Tuple[Dict[str, Any], str]] = [] def _fetch_and_filter(ocr_numbers: Optional[str], ocr_letters: Optional[str], lang: str): if not ocr_numbers or not ocr_letters: return variants = _number_variants(ocr_numbers) if not variants: return col = "plate_numbers_ar" if lang == "ar" else "plate_numbers_en" conditions = [f'{col}.eq."{(v).replace(chr(34), "")}"' for v in variants] or_filter = "(" + ",".join(conditions) + ")" params = { "select": "id,owner_id,plate_letters_ar,plate_numbers_ar,plate_letters_en,plate_numbers_en,car_model,car_color,is_active", "is_active": "eq.true", "or": or_filter, } response = requests.get( f"{SUPABASE_URL}/rest/v1/vehicles", params=params, headers=_supabase_headers(api_key=SUPABASE_SERVICE_ROLE_KEY, bearer=SUPABASE_SERVICE_ROLE_KEY), timeout=SUPABASE_TIMEOUT_SECONDS, ) if response.status_code == 200: rows = response.json() if isinstance(rows, list): ocr_num_norm = normalize_for_cmp(ocr_numbers) ocr_let_norm = normalize_for_cmp(ocr_letters) let_col = "plate_letters_ar" if lang == "ar" else "plate_letters_en" for row in rows: db_num_norm = normalize_for_cmp(row.get(col, "")) db_let_norm = normalize_for_cmp(row.get(let_col, "")) is_match = (db_num_norm == ocr_num_norm and db_let_norm == ocr_let_norm) if not is_match and db_let_norm == ocr_let_norm: is_match = sorted(db_num_norm) == sorted(ocr_num_norm) if is_match: matched_vehicles.append((row, lang)) _fetch_and_filter(plate_numbers_ar, plate_letters_ar, "ar") if not matched_vehicles: _fetch_and_filter(plate_numbers_en, plate_letters_en, "en") if not matched_vehicles: return None if user_id: for v, lang in matched_vehicles: if str(v.get("owner_id")) == str(user_id): return v return matched_vehicles[0][0] def _resolve_or_create_open_parking_session(*, vehicle_id: Optional[str], created_by: Optional[str], parking_location: Optional[str], plate_arabic: Optional[str] = None, plate_english: Optional[str] = None, create_if_missing: bool = True) -> Optional[str]: if not supabase_configured(): return None params: Dict[str, str] = {"select": "id,parking_location,status,check_out_at", "check_out_at": "is.null", "status": "not.in.(exited,left_without_payment)", "order": "check_in_at.desc", "limit": "1"} if vehicle_id: params["vehicle_id"] = f"eq.{vehicle_id}" elif plate_arabic or plate_english: is_ar_valid = plate_arabic and plate_arabic.upper() not in {"", "N/A", "NONE", "NULL", "-"} is_en_valid = plate_english and plate_english.upper() not in {"", "N/A", "NONE", "NULL", "-"} if is_ar_valid or is_en_valid: params["vehicle_id"] = "is.null" plate_filters = [] if is_ar_valid: plate_filters.append(f"plate_arabic.eq.{plate_arabic}") if is_en_valid: plate_filters.append(f"plate_english.eq.{plate_english}") if plate_filters: params["or"] = f"({','.join(plate_filters)})" else: return None else: return None open_response = requests.get( f"{SUPABASE_URL}/rest/v1/parking_sessions", params=params, headers=_supabase_headers(api_key=SUPABASE_SERVICE_ROLE_KEY, bearer=SUPABASE_SERVICE_ROLE_KEY), timeout=SUPABASE_TIMEOUT_SECONDS, ) if open_response.status_code != 200: raise RuntimeError(f"Failed to query parking_sessions ({open_response.status_code})") rows = open_response.json() if isinstance(rows, list) and rows: session_id = rows[0].get("id") if not session_id: return None current_location = rows[0].get("parking_location") if parking_location and parking_location != current_location: requests.patch( f"{SUPABASE_URL}/rest/v1/parking_sessions", params={"id": f"eq.{session_id}"}, json={"parking_location": parking_location}, headers=_supabase_headers(api_key=SUPABASE_SERVICE_ROLE_KEY, bearer=SUPABASE_SERVICE_ROLE_KEY, content_type="application/json"), timeout=SUPABASE_TIMEOUT_SECONDS, ) return str(session_id) if not create_if_missing: return None payload: Dict[str, Any] = {"vehicle_id": vehicle_id, "status": "entered", "plate_arabic": plate_arabic, "plate_english": plate_english, "check_in_at": datetime.now(TZ_UTC).isoformat()} if created_by: payload["created_by"] = created_by if parking_location: payload["parking_location"] = parking_location create_response = requests.post( f"{SUPABASE_URL}/rest/v1/parking_sessions", json=payload, headers=_supabase_headers(api_key=SUPABASE_SERVICE_ROLE_KEY, bearer=SUPABASE_SERVICE_ROLE_KEY, content_type="application/json", prefer="return=representation"), timeout=SUPABASE_TIMEOUT_SECONDS, ) if create_response.status_code not in {200, 201}: raise RuntimeError(f"Failed to create parking_session ({create_response.status_code})") created_rows = create_response.json() if isinstance(created_rows, list) and created_rows: sid = created_rows[0].get("id") return str(sid) if sid else None return None def _fetch_parking_session_by_id(session_id: Optional[str]) -> Optional[Dict[str, Any]]: if not supabase_configured() or not session_id: return None response = requests.get( f"{SUPABASE_URL}/rest/v1/parking_sessions", params={"select": "*", "id": f"eq.{session_id}", "limit": "1"}, headers=_supabase_headers(api_key=SUPABASE_SERVICE_ROLE_KEY, bearer=SUPABASE_SERVICE_ROLE_KEY), timeout=SUPABASE_TIMEOUT_SECONDS, ) if response.status_code != 200: return None rows = response.json() if isinstance(rows, list) and rows and isinstance(rows[0], dict): return rows[0] return None def _insert_car_event_record(*, event_type: str, created_by: Optional[str], vehicle_id: Optional[str], session_id: Optional[str], raw_image_path: str, user_split_image_path: str, admin_annotated_image_path: str, ocr_arabic: str, ocr_english: str, ocr_raw_labels: List[str], ocr_confidence: Optional[float], parking_location: Optional[str], event_metadata: Dict[str, Any]) -> Optional[str]: payload: Dict[str, Any] = {"event_type": event_type, "raw_image_path": raw_image_path, "user_split_image_path": user_split_image_path, "admin_annotated_image_path": admin_annotated_image_path, "ocr_arabic": ocr_arabic, "ocr_english": ocr_english, "ocr_raw_labels": ocr_raw_labels, "event_metadata": event_metadata} if created_by: payload["created_by"] = created_by if vehicle_id: payload["vehicle_id"] = vehicle_id if session_id: payload["session_id"] = session_id if ocr_confidence is not None: payload["ocr_confidence"] = ocr_confidence if parking_location: payload["parking_location"] = parking_location response = requests.post( f"{SUPABASE_URL}/rest/v1/car_events", json=payload, headers=_supabase_headers(api_key=SUPABASE_SERVICE_ROLE_KEY, bearer=SUPABASE_SERVICE_ROLE_KEY, content_type="application/json", prefer="return=representation"), timeout=SUPABASE_TIMEOUT_SECONDS, ) if response.status_code not in {200, 201}: raise RuntimeError(f"Failed to insert car_events ({response.status_code})") rows = response.json() if isinstance(rows, list) and rows: return rows[0].get("id") return None def _persist_prediction_to_supabase(*, original_image_bytes: bytes, response_payload: Dict[str, Any], request_user: Optional[Dict[str, Any]], parking_location: Optional[str], event_type: str, camera_source: Optional[str]) -> Dict[str, Any]: default_alert = {"is_new_car": None, "notify_roles": ["admin", "security"], "action": "lookup_unavailable", "reason": "supabase_unavailable", "message": "Could not verify registration status.", "vehicle_id": None, "owner_id": None, "plate_arabic": None, "plate_english": None} if not supabase_configured(): return {"enabled": False, "saved": False, "reason": "Supabase env vars are not configured."} user_folder = request_user["id"] if request_user else "anonymous" object_base = f"{user_folder}/{datetime.now(TZ_UTC):%Y/%m/%d}/{uuid4().hex}" raw_path = f"{object_base}_raw.jpg" user_split_path = f"{object_base}_user.jpg" admin_annotated_path = f"{object_base}_admin.jpg" new_car_alert = dict(default_alert) uploaded_objects: List[Tuple[str, str]] = [] try: plate_info = response_payload.get("plate_info", {}) plate_detected = _is_plate_detected(plate_info) request_user_id = _extract_created_by_user_id(request_user) vehicle_record = _resolve_vehicle_record_by_plate(plate_info.get("arabic", ""), plate_info.get("english", ""), request_user_id) vehicle_id = str(vehicle_record.get("id")) if isinstance(vehicle_record, dict) and vehicle_record.get("id") else None owner_id = str(vehicle_record.get("owner_id")) if isinstance(vehicle_record, dict) and vehicle_record.get("owner_id") else None if not plate_detected: new_car_alert.update({"reason": "plate_not_detected", "message": "Could not detect a plate."}) elif vehicle_id: new_car_alert.update({"reason": "already_registered", "message": "Car is already registered."}) else: new_car_alert = {"is_new_car": True, "notify_roles": ["admin", "security"], "action": "vendor_outreach_recommended", "reason": "car_not_registered_by_any_user", "message": "New car detected.", "vehicle_id": None, "owner_id": None, "plate_arabic": plate_info.get("arabic"), "plate_english": plate_info.get("english")} session_id = _resolve_or_create_open_parking_session( vehicle_id=vehicle_id, created_by=request_user_id, parking_location=parking_location, plate_arabic=plate_info.get("arabic"), plate_english=plate_info.get("english"), create_if_missing=(event_type == "entry") ) if event_type == "exit" and session_id: requests.patch( f"{SUPABASE_URL}/rest/v1/parking_sessions", params={"id": f"eq.{session_id}"}, json={"status": "exited", "check_out_at": datetime.now(TZ_UTC).isoformat()}, headers=_supabase_headers(api_key=SUPABASE_SERVICE_ROLE_KEY, bearer=SUPABASE_SERVICE_ROLE_KEY, content_type="application/json"), timeout=SUPABASE_TIMEOUT_SECONDS, ) # Trigger gate open automatically on any valid prediction event (entry or exit) try: from app.api.gate import _gate_pending_commands, _gate_command_lock cmd = { "action": "open", "parking_location": parking_location or "OPERA", "duration_seconds": 10, "plate": plate_info.get("arabic") or "", "queued_at": datetime.now(TZ_UTC).isoformat(), } with _gate_command_lock: _gate_pending_commands.append(cmd) if len(_gate_pending_commands) > 20: _gate_pending_commands.pop(0) LOGGER.info("Gate triggered automatically for %s", event_type) except Exception: LOGGER.exception("Failed to trigger gate automatically") upload_to_supabase_storage(SUPABASE_RAW_BUCKET, raw_path, original_image_bytes) uploaded_objects.append((SUPABASE_RAW_BUCKET, raw_path)) user_page = response_payload.get("user_page", {}) if isinstance(response_payload.get("user_page"), dict) else {} admin_page = response_payload.get("admin_page", {}) if isinstance(response_payload.get("admin_page"), dict) else {} user_split_b64 = user_page.get("split_image_base64", "") admin_annotated_b64 = admin_page.get("annotated_image_base64", "") if user_split_b64: upload_to_supabase_storage(SUPABASE_PROCESSED_BUCKET, user_split_path, base64.b64decode(user_split_b64)) uploaded_objects.append((SUPABASE_PROCESSED_BUCKET, user_split_path)) if admin_annotated_b64: upload_to_supabase_storage(SUPABASE_PROCESSED_BUCKET, admin_annotated_path, base64.b64decode(admin_annotated_b64)) uploaded_objects.append((SUPABASE_PROCESSED_BUCKET, admin_annotated_path)) event_id = _insert_car_event_record(event_type=event_type, created_by=request_user_id, vehicle_id=vehicle_id, session_id=session_id, raw_image_path=f"{SUPABASE_RAW_BUCKET}/{raw_path}", user_split_image_path=f"{SUPABASE_PROCESSED_BUCKET}/{user_split_path}", admin_annotated_image_path=f"{SUPABASE_PROCESSED_BUCKET}/{admin_annotated_path}", ocr_arabic=plate_info.get("arabic", "N/A"), ocr_english=plate_info.get("english", "N/A"), ocr_raw_labels=plate_info.get("raw_ordered_labels", []), ocr_confidence=None, parking_location=parking_location, event_metadata={"event_type": event_type, "parking_location": parking_location, "camera_source": camera_source, "new_car_alert": new_car_alert, "app_registration": {"is_registered": vehicle_id is not None, "vehicle_id": vehicle_id, "owner_id": owner_id}}) return {"enabled": True, "saved": True, "event_id": event_id, "session_id": session_id, "app_registration": {"is_registered": vehicle_id is not None, "vehicle_id": vehicle_id, "owner_id": owner_id}, "new_car_alert": new_car_alert} except Exception as exc: LOGGER.exception("Supabase persistence failed") for bucket, obj_path in reversed(uploaded_objects): try: delete_from_supabase_storage(bucket, obj_path) except Exception: pass return {"enabled": True, "saved": False, "error": str(exc)} @router.post("/predict") async def predict(image: UploadFile = File(...), event_type: Optional[str] = Form(default="entry"), parking_location: Optional[str] = Form(default=None), camera_source: Optional[str] = Form(default=None), authorization: Optional[str] = Header(default=None)) -> Dict[str, Any]: request_user = require_authenticated_user(authorization) normalized_parking_location = _normalize_parking_location(parking_location) normalized_camera_source = _normalize_camera_source(camera_source) if image.content_type and not image.content_type.startswith("image/") and image.content_type != "application/octet-stream": raise HTTPException(status_code=400, detail="Only image uploads are supported.") image_bytes = await image.read() if not image_bytes: raise HTTPException(status_code=400, detail="Uploaded file is empty.") response = run_pipeline_remote(image_bytes=image_bytes, filename=image.filename, content_type=image.content_type) if not isinstance(response, dict): raise HTTPException(status_code=502, detail="Inference pipeline returned an invalid payload.") response["inference_provider"] = "remote" response["filename"] = image.filename normalized_event_type = _normalize_event_type(event_type) if normalized_event_type not in {"entry", "exit"}: # Determine event type automatically based on Supabase open session state plate_info = response.get("plate_info", {}) request_user_id = _extract_created_by_user_id(request_user) vehicle_record = _resolve_vehicle_record_by_plate(plate_info.get("arabic", ""), plate_info.get("english", ""), request_user_id) vehicle_id = str(vehicle_record.get("id")) if isinstance(vehicle_record, dict) and vehicle_record.get("id") else None open_session_id = _resolve_or_create_open_parking_session( vehicle_id=vehicle_id, created_by=request_user_id, parking_location=normalized_parking_location, plate_arabic=plate_info.get("arabic"), plate_english=plate_info.get("english"), create_if_missing=False ) if open_session_id: normalized_event_type = "exit" else: normalized_event_type = "entry" response["event_type"] = normalized_event_type response["parking_location"] = normalized_parking_location response["camera_source"] = normalized_camera_source plate_payload = _build_plate_payload(response.get("plate_info", {})) response["plate_payload"] = plate_payload response["plate"] = plate_payload.get("arabic_text") or "N/A" response["plate_text_ar"] = plate_payload.get("arabic_text") or "N/A" response["plate_text_en"] = plate_payload.get("english_text") or "N/A" _ar = plate_payload.get("arabic", {}) if isinstance(plate_payload.get("arabic"), dict) else {} _en = plate_payload.get("english", {}) if isinstance(plate_payload.get("english"), dict) else {} response["plate_letters_ar"] = _ar.get("letters") or "" response["plate_numbers_ar"] = _ar.get("numbers") or "" response["plate_letters_en"] = _en.get("letters") or "" response["plate_numbers_en"] = _en.get("numbers") or "" original_b64 = base64.b64encode(image_bytes).decode("utf-8") response["processed_image_base64"] = str((response.get("user_page", {}) or {}).get("split_image_base64") or original_b64) response["annotated_image_base64"] = str((response.get("admin_page", {}) or {}).get("annotated_image_base64") or original_b64) if request_user: response["request_user"] = {"id": request_user.get("id"), "email": request_user.get("email")} supabase_result = _persist_prediction_to_supabase(original_image_bytes=image_bytes, response_payload=response, request_user=request_user, parking_location=normalized_parking_location, event_type=normalized_event_type, camera_source=normalized_camera_source) response["supabase"] = supabase_result response["app_registration"] = supabase_result.get("app_registration", {"is_registered": False, "vehicle_id": None, "owner_id": None}) response["new_car_alert"] = supabase_result.get("new_car_alert", {"is_new_car": None, "notify_roles": ["admin", "security"], "action": "lookup_unavailable", "reason": "supabase_unavailable", "message": "Could not verify registration status.", "vehicle_id": None, "owner_id": None, "plate_arabic": None, "plate_english": None}) response["session_id"] = supabase_result.get("session_id") response["event_id"] = supabase_result.get("event_id") # Fetch and attach entry time for exit event calculations if response.get("session_id"): session_row = _fetch_parking_session_by_id(response["session_id"]) if session_row: response["entry_time"] = session_row.get("check_in_at") return response @router.post("/model/infer") async def model_infer(image: UploadFile = File(...), model_service_secret: Optional[str] = Header(default=None, alias=MODEL_SERVICE_HEADER_NAME)) -> Dict[str, Any]: if image.content_type and not image.content_type.startswith("image/") and image.content_type != "application/octet-stream": raise HTTPException(status_code=400, detail="Only image uploads are supported.") image_bytes = await image.read() if not image_bytes: raise HTTPException(status_code=400, detail="Uploaded file is empty.") response = run_pipeline_remote(image_bytes=image_bytes, filename=image.filename, content_type=image.content_type) response["filename"] = image.filename response["inference_provider"] = "remote" return response @router.post("/developer/predict") async def developer_predict(image: UploadFile = File(...), event_type: Optional[str] = Form(default="entry"), parking_location: Optional[str] = Form(default=None), camera_source: Optional[str] = Form(default=None), authorization: Optional[str] = Header(default=None)) -> Dict[str, Any]: request_user, requester_role = require_developer_user(authorization) return await predict(image=image, event_type=event_type, parking_location=parking_location, camera_source=camera_source, authorization=authorization)