Upload 48 files

.gitattributes

@@ -1,35 +1 @@
-*.7z filter=lfs diff=lfs merge=lfs -text
-*.arrow filter=lfs diff=lfs merge=lfs -text
-*.bin filter=lfs diff=lfs merge=lfs -text
-*.bz2 filter=lfs diff=lfs merge=lfs -text
-*.ckpt filter=lfs diff=lfs merge=lfs -text
-*.ftz filter=lfs diff=lfs merge=lfs -text
-*.gz filter=lfs diff=lfs merge=lfs -text
-*.h5 filter=lfs diff=lfs merge=lfs -text
-*.joblib filter=lfs diff=lfs merge=lfs -text
-*.lfs.* filter=lfs diff=lfs merge=lfs -text
-*.mlmodel filter=lfs diff=lfs merge=lfs -text
-*.model filter=lfs diff=lfs merge=lfs -text
-*.msgpack filter=lfs diff=lfs merge=lfs -text
-*.npy filter=lfs diff=lfs merge=lfs -text
-*.npz filter=lfs diff=lfs merge=lfs -text
-*.onnx filter=lfs diff=lfs merge=lfs -text
-*.ot filter=lfs diff=lfs merge=lfs -text
-*.parquet filter=lfs diff=lfs merge=lfs -text
-*.pb filter=lfs diff=lfs merge=lfs -text
-*.pickle filter=lfs diff=lfs merge=lfs -text
-*.pkl filter=lfs diff=lfs merge=lfs -text
-*.pt filter=lfs diff=lfs merge=lfs -text
-*.pth filter=lfs diff=lfs merge=lfs -text
-*.rar filter=lfs diff=lfs merge=lfs -text
-*.safetensors filter=lfs diff=lfs merge=lfs -text
-saved_model/**/* filter=lfs diff=lfs merge=lfs -text
-*.tar.* filter=lfs diff=lfs merge=lfs -text
-*.tar filter=lfs diff=lfs merge=lfs -text
-*.tflite filter=lfs diff=lfs merge=lfs -text
-*.tgz filter=lfs diff=lfs merge=lfs -text
-*.wasm filter=lfs diff=lfs merge=lfs -text
-*.xz filter=lfs diff=lfs merge=lfs -text
-*.zip filter=lfs diff=lfs merge=lfs -text
-*.zst filter=lfs diff=lfs merge=lfs -text
-*tfevents* filter=lfs diff=lfs merge=lfs -text
+*.pt filter=lfs diff=lfs merge=lfs -text

app/config.py

@@ -0,0 +1,130 @@
+"""
+Project configuration file.
+All hardcoded values are extracted from the notebook and centralized here.
+"""
+
+import os
+from pathlib import Path
+from dotenv import load_dotenv, find_dotenv
+
+# Load environment variables from .env file (skipped on HF Spaces demo)
+GRADIO_SPACE = os.getenv("GRADIO_SPACE", "").lower() in ("1", "true", "yes")
+if not GRADIO_SPACE:
+    ENV_FILE = find_dotenv(filename=".env", usecwd=True)
+    if ENV_FILE:
+        load_dotenv(ENV_FILE, override=False, interpolate=True)
+
+# ======================
+# PROJECT PATHS
+# ======================
+DEFAULT_PROJECT_ROOT = Path(__file__).resolve().parents[1]
+PROJECT_ROOT = Path(os.getenv("PROJECT_ROOT", DEFAULT_PROJECT_ROOT)).resolve()
+BASE_DIR = str(PROJECT_ROOT)
+APP_DIR = Path(__file__).parent.resolve()
+
+# ======================
+# MODEL PATHS & HUGGING FACE CONFIGURATION
+# ======================
+HF_TOKEN = os.getenv("HF_TOKEN", "")
+USE_HUGGING_FACE = (
+    os.getenv("USE_HUGGING_FACE", "").lower() in ("1", "true", "yes")
+    or bool(HF_TOKEN)
+    or bool(
+        os.getenv("HF_PLATE_DETECTOR_REPO")
+        and os.getenv("HF_CHAR_OCR_REPO")
+    )
+)
+
+# Hugging Face model IDs
+HF_PLATE_DETECTOR_REPO = os.getenv("HF_PLATE_DETECTOR_REPO", "")
+HF_CHAR_OCR_REPO = os.getenv("HF_CHAR_OCR_REPO", "")
+HF_CAR_CLASSIFIER_REPO = os.getenv("HF_CAR_CLASSIFIER_REPO", "")
+
+# Local fallback paths (used if HF token not provided)
+MODELS_DIR = PROJECT_ROOT / "models"
+PLATE_DETECTOR_MODEL_PATH = str(os.getenv("PLATE_DETECTOR_MODEL_PATH", 
+                                           MODELS_DIR / "plate_detector" / "best.pt"))
+CHAR_OCR_MODEL_PATH = str(os.getenv("CHAR_OCR_MODEL_PATH", 
+                                     MODELS_DIR / "char_ocr" / "best.pt"))
+CAR_CLASSIFIER_MODEL_PATH = str(os.getenv("CAR_CLASSIFIER_MODEL_PATH", 
+                                          MODELS_DIR / "car_classifier" / "best.pt"))
+
+# Debug: Print resolved paths
+import logging
+logger = logging.getLogger(__name__)
+logger.info(f"Config - PROJECT_ROOT: {PROJECT_ROOT}")
+logger.info(f"Config - MODELS_DIR: {MODELS_DIR}")
+logger.info(f"Config - PLATE_DETECTOR_MODEL_PATH: {PLATE_DETECTOR_MODEL_PATH}")
+
+# ======================
+# MODEL INFERENCE SETTINGS
+# ======================
+CONF_THRESHOLD = 0.25         # Confidence threshold for YOLO detections
+VARIANCE_THRESHOLD = 100.0    # Minimum variance to consider frame usable
+
+# ======================
+# IMAGE PREPROCESSING SETTINGS
+# ======================
+# Enhancer removed: EDSR settings deprecated
+
+# ======================
+# SCORING WEIGHTS
+# ======================
+SEGMENT_SCORE_WEIGHT = 0.65   # Weight for segment score in final calculation
+PLATE_SCORE_WEIGHT = 0.35     # Weight for plate score in final calculation
+
+# ======================
+# DATABASE CONFIGURATION (Supabase)
+# ======================
+
+# Supabase configuration (NEW)
+SUPABASE_URL = os.getenv("SUPABASE_URL", "https://yoenbdxxdwknizqthkfp.supabase.co")
+SUPABASE_SERVICE_KEY = os.getenv("SUPABASE_SERVICE_KEY", "")
+'''
+SUPABASE_URL         = os.getenv("SUPABASE_URL",         "")
+SUPABASE_SERVICE_KEY = os.getenv("SUPABASE_SERVICE_KEY", "")
+'''
+
+if not SUPABASE_SERVICE_KEY and not GRADIO_SPACE:
+    raise ValueError("SUPABASE_SERVICE_KEY not found in environment variables")
+
+# ======================
+# PLATE SCORING RULES
+# ======================
+PLATE_SCORING_RULES = {
+    "digit_count_weight": 0.3,      # Lower digit count → higher score
+    "repeating_weight": 0.2,        # Bonus for repeating patterns
+    "sequential_weight": 0.2,       # Bonus for sequential patterns
+    "low_number_weight": 0.3,       # Bonus for numbers < 100
+}
+
+# ======================
+# VIDEO CAPTURE (pipeline VideoCapturer)
+# ======================
+CAPTURE_OUTPUT_DIR = PROJECT_ROOT / "src" / "capture" / "captured_frames"
+CAPTURE_OUTPUT_DIR.mkdir(parents=True, exist_ok=True)
+
+CAPTURE_PROCESS_EVERY_N_FRAMES = int(os.getenv("CAPTURE_PROCESS_EVERY_N_FRAMES", "3"))
+CAPTURE_FRAME_BUFFER_SIZE = int(os.getenv("CAPTURE_FRAME_BUFFER_SIZE", "20"))
+CAPTURE_MOTION_THRESHOLD = int(os.getenv("CAPTURE_MOTION_THRESHOLD", "1500"))
+CAPTURE_MOG2_HISTORY = int(os.getenv("CAPTURE_MOG2_HISTORY", "500"))
+CAPTURE_MOG2_VAR_THRESHOLD = int(os.getenv("CAPTURE_MOG2_VAR_THRESHOLD", "50"))
+CAPTURE_COOLDOWN_SECONDS = float(os.getenv("CAPTURE_COOLDOWN_SECONDS", "2"))
+CAPTURE_TRIGGER_MODE = os.getenv("CAPTURE_TRIGGER_MODE", "yolo")  # yolo | motion
+CAPTURE_VEHICLE_MODEL = os.getenv("CAPTURE_VEHICLE_MODEL", "yolo11n.pt")
+CAPTURE_VEHICLE_CONF = float(os.getenv("CAPTURE_VEHICLE_CONF", "0.40"))
+CAPTURE_VEHICLE_MIN_BOX_AREA = int(os.getenv("CAPTURE_VEHICLE_MIN_BOX_AREA", "3000"))
+CAPTURE_VEHICLE_CROP_PADDING = int(os.getenv("CAPTURE_VEHICLE_CROP_PADDING", "50"))
+CAPTURE_VEHICLE_FILTER_ENABLED = os.getenv("CAPTURE_VEHICLE_FILTER_ENABLED", "true").lower() == "true"
+CAPTURE_MAX_VEHICLES_PER_TRIGGER = int(os.getenv("CAPTURE_MAX_VEHICLES_PER_TRIGGER", "1"))
+CAPTURE_MIN_SHARPNESS_SCORE = float(os.getenv("CAPTURE_MIN_SHARPNESS_SCORE", "5.0"))
+CAPTURE_MIN_BRIGHTNESS = int(os.getenv("CAPTURE_MIN_BRIGHTNESS", "30"))
+CAPTURE_MAX_BRIGHTNESS = int(os.getenv("CAPTURE_MAX_BRIGHTNESS", "220"))
+CAPTURE_DEDUP_IOU = float(os.getenv("CAPTURE_DEDUP_IOU", "0.45"))
+# Safety cap for POST /api/detect video uploads (None = no cap)
+CAPTURE_MAX_TRIGGERS = int(os.getenv("CAPTURE_MAX_TRIGGERS", "30")) or None
+
+# ======================
+# ENSURE REQUIRED PATHS
+# ======================
+MODELS_DIR.mkdir(parents=True, exist_ok=True)

app/database/connection.py

@@ -0,0 +1,83 @@
+"""
+Supabase Database Connection Module
+
+Handles connection to Supabase PostgreSQL database using the Supabase REST API.
+Credentials are read from config.py environment variables.
+"""
+
+import requests
+from ..config import SUPABASE_URL, SUPABASE_SERVICE_KEY
+
+
+class Database:
+    """
+    Supabase database connection manager.
+    
+    Uses the Supabase REST API directly via requests.
+    Service key provides admin-level access from backend.
+    """
+    
+    def __init__(self, url: str = SUPABASE_URL, key: str = SUPABASE_SERVICE_KEY):
+        """
+        Initialize Supabase connection.
+        
+        Args:
+            url: Supabase project URL.
+            key: Supabase service key (for server-side admin access).
+        
+        Raises:
+            ValueError: If credentials are missing.
+        """
+        if not url or not key:
+            raise ValueError("SUPABASE_URL and SUPABASE_SERVICE_KEY are required")
+
+        self.base_url = url.rstrip("/")
+        self.headers = {
+            "apikey": key,
+            "Authorization": f"Bearer {key}",
+            "Content-Type": "application/json"
+        }
+        self.session = requests.Session()
+
+        print(f"Supabase connection configured: {self.base_url}")
+
+    def close(self):
+        """
+        Close the underlying HTTP session.
+        """
+        self.session.close()
+
+    def close_all(self):
+        """
+        Alias for backward compatibility with shutdown cleanup.
+        """
+        self.close()
+
+    def insert(self, table: str, data: dict) -> list:
+        """
+        Insert a record into a Supabase table and return the inserted row.
+        """
+        url = f"{self.base_url}/rest/v1/{table}"
+        headers = self.headers.copy()
+        headers["Prefer"] = "return=representation"
+
+        response = self.session.post(url, json=data, headers=headers)
+        if response.status_code not in (200, 201):
+            raise RuntimeError(f"Supabase insert failed ({response.status_code}): {response.text}")
+
+        return response.json()
+
+    def query(self, table: str, select: str = "*", params: dict = None) -> list:
+        """
+        Run a simple select query against a Supabase table.
+        """
+        url = f"{self.base_url}/rest/v1/{table}"
+        query_params = {"select": select}
+        if params:
+            query_params.update(params)
+
+        response = self.session.get(url, headers=self.headers, params=query_params)
+        if response.status_code != 200:
+            raise RuntimeError(f"Supabase query failed ({response.status_code}): {response.text}")
+
+        return response.json()

app/database/logger.py

@@ -0,0 +1,113 @@
+"""
+Visit Logger Module
+
+Logs detection results and scores to the database.
+Records include: plate detection, character recognition, and scoring information.
+"""
+
+from datetime import datetime
+import hashlib
+
+from .connection import Database
+
+
+class VisitLogger:
+    """
+    Logs visit records to the database.
+
+    Records are written to multiple tables:
+    - visits: Main visit record
+    - plate_scores: License plate prestige score
+    - segment_scores: Vehicle segment/class score
+    - purchasing_power: Final purchasing power metric
+    """
+
+    def __init__(self, database: Database):
+        self.db = database
+
+    def log(self, plate_digits: str, plate_letters: str, plate_score: float,
+            segment: str, segment_score: float, purchasing_power: float,
+            car_brand: str = None, car_model: str = None,
+            timestamp: datetime = None,
+            camera_id: str = None) -> str:
+
+        if not plate_digits:
+            raise ValueError("plate_digits cannot be empty")
+        if not camera_id:
+            raise ValueError(
+                "camera_id is required — pass a valid camera UUID from the cameras table."
+            )
+        if not (0.0 <= plate_score <= 1.0):
+            raise ValueError(f"plate_score must be in [0.0, 1.0], got {plate_score}")
+        if not (0.0 <= segment_score <= 1.0):
+            raise ValueError(f"segment_score must be in [0.0, 1.0], got {segment_score}")
+        if not (0.0 <= purchasing_power <= 1.0):
+            raise ValueError(f"purchasing_power must be in [0.0, 1.0], got {purchasing_power}")
+
+        timestamp = timestamp or datetime.now()
+
+        try:
+            visit_id = self._insert_visit(timestamp, plate_digits, plate_letters, camera_id)
+            self._insert_plate_score(visit_id, plate_score)
+            self._insert_segment_score(visit_id, segment, segment_score)
+            self._insert_purchasing_power_score(visit_id, purchasing_power)
+            return visit_id
+        except Exception as e:
+            raise RuntimeError(f"Failed to log visit: {e}")
+
+    def _insert_visit(self, timestamp: datetime, plate_digits: str,
+                      plate_letters: str, camera_id: str) -> str:
+
+        plate_hash = hashlib.sha256(
+            f"{plate_letters}{plate_digits}".encode("utf-8")
+        ).hexdigest()
+
+        data = {
+            "camera_id":    camera_id,
+            "plate_hash":   plate_hash,
+            "plate_digits": plate_digits,
+            "plate_letters": plate_letters,
+            "visited_at":   timestamp.isoformat()
+        }
+
+        response = self.db.insert("visits", data)
+        if not response or len(response) == 0:
+            raise RuntimeError("Failed to insert visit: no data returned")
+
+        return response[0]["id"]
+
+    def _insert_plate_score(self, visit_id: str, score: float) -> None:
+        data = {
+            "visit_id":           visit_id,
+            "digit_length_score": score,
+            "repeat_score":       0,
+            "sequential_score":   0,
+            "low_number_score":   0,
+            "total_score":        score
+        }
+        response = self.db.insert("plate_scores", data)
+        if not response or len(response) == 0:
+            raise RuntimeError("Failed to insert plate score")
+
+    def _insert_segment_score(self, visit_id: str, segment: str,
+                              score: float) -> None:
+        data = {
+            "visit_id":      visit_id,
+            "segment":       segment,
+            "confidence":    score,
+            "segment_score": score
+        }
+        response = self.db.insert("segment_scores", data)
+        if not response or len(response) == 0:
+            raise RuntimeError("Failed to insert segment score")
+
+    def _insert_purchasing_power_score(self, visit_id: str, score: float) -> None:
+        data = {
+            "visit_id":       visit_id,
+            "plate_weight":   0,
+            "segment_weight": 0,
+            "final_score":    score
+        }
+        response = self.db.insert("purchasing_power", data)
+        if not response or len(response) == 0:
+            raise RuntimeError("Failed to insert purchasing power score")

app/database/supabase/config.toml

@@ -0,0 +1 @@
+project_id = "yoenbdxxdwknizqthkfp"

app/database/supabase/migrations/20260508175106_d61705db-2487-451d-ad0c-3940fa1fb5b0.sql

@@ -0,0 +1,61 @@
+
+CREATE TABLE public.cameras (
+  id UUID NOT NULL DEFAULT gen_random_uuid() PRIMARY KEY,
+  location_name TEXT NOT NULL,
+  location_type TEXT NOT NULL,
+  installed_at TIMESTAMPTZ NOT NULL DEFAULT now()
+);
+
+CREATE TABLE public.visits (
+  id UUID NOT NULL DEFAULT gen_random_uuid() PRIMARY KEY,
+  camera_id UUID NOT NULL REFERENCES public.cameras(id) ON DELETE CASCADE,
+  plate_hash TEXT NOT NULL,
+  plate_digits TEXT NOT NULL,
+  plate_letters TEXT NOT NULL,
+  visited_at TIMESTAMPTZ NOT NULL DEFAULT now()
+);
+CREATE INDEX idx_visits_camera ON public.visits(camera_id);
+CREATE INDEX idx_visits_visited_at ON public.visits(visited_at DESC);
+CREATE INDEX idx_visits_plate_hash ON public.visits(plate_hash);
+
+CREATE TABLE public.plate_scores (
+  id UUID NOT NULL DEFAULT gen_random_uuid() PRIMARY KEY,
+  visit_id UUID NOT NULL REFERENCES public.visits(id) ON DELETE CASCADE,
+  digit_length_score NUMERIC NOT NULL DEFAULT 0,
+  repeat_score NUMERIC NOT NULL DEFAULT 0,
+  sequential_score NUMERIC NOT NULL DEFAULT 0,
+  low_number_score NUMERIC NOT NULL DEFAULT 0,
+  total_score NUMERIC NOT NULL DEFAULT 0
+);
+CREATE INDEX idx_plate_scores_visit ON public.plate_scores(visit_id);
+
+CREATE TABLE public.segment_scores (
+  id UUID NOT NULL DEFAULT gen_random_uuid() PRIMARY KEY,
+  visit_id UUID NOT NULL REFERENCES public.visits(id) ON DELETE CASCADE,
+  segment TEXT NOT NULL,
+  confidence NUMERIC NOT NULL DEFAULT 0,
+  segment_score NUMERIC NOT NULL DEFAULT 0
+);
+CREATE INDEX idx_segment_scores_visit ON public.segment_scores(visit_id);
+
+CREATE TABLE public.purchasing_power (
+  id UUID NOT NULL DEFAULT gen_random_uuid() PRIMARY KEY,
+  visit_id UUID NOT NULL REFERENCES public.visits(id) ON DELETE CASCADE,
+  plate_weight NUMERIC NOT NULL DEFAULT 0,
+  segment_weight NUMERIC NOT NULL DEFAULT 0,
+  final_score NUMERIC NOT NULL DEFAULT 0
+);
+CREATE INDEX idx_pp_visit ON public.purchasing_power(visit_id);
+CREATE INDEX idx_pp_final_score ON public.purchasing_power(final_score DESC);
+
+ALTER TABLE public.cameras ENABLE ROW LEVEL SECURITY;
+ALTER TABLE public.visits ENABLE ROW LEVEL SECURITY;
+ALTER TABLE public.plate_scores ENABLE ROW LEVEL SECURITY;
+ALTER TABLE public.segment_scores ENABLE ROW LEVEL SECURITY;
+ALTER TABLE public.purchasing_power ENABLE ROW LEVEL SECURITY;
+
+CREATE POLICY "Public read cameras" ON public.cameras FOR SELECT USING (true);
+CREATE POLICY "Public read visits" ON public.visits FOR SELECT USING (true);
+CREATE POLICY "Public read plate_scores" ON public.plate_scores FOR SELECT USING (true);
+CREATE POLICY "Public read segment_scores" ON public.segment_scores FOR SELECT USING (true);
+CREATE POLICY "Public read purchasing_power" ON public.purchasing_power FOR SELECT USING (true);

app/main.py

@@ -0,0 +1,113 @@
+"""
+FastAPI application entry point.
+
+Start with:
+    uvicorn app.main:app --host 0.0.0.0 --port 8000 --reload
+"""
+
+import logging
+from contextlib import asynccontextmanager
+
+from fastapi import FastAPI
+from fastapi.middleware.cors import CORSMiddleware
+
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s - %(levelname)s - %(message)s",
+)
+logger = logging.getLogger(__name__)
+
+
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    """Initialize resources on startup, clean up on shutdown."""
+
+    # ── Startup ───────────────────────────────────────────────────────────────
+    logger.info("Starting up...")
+
+    try:
+        from .pipeline.coordinator import LPRPipeline
+        app.state.pipeline = LPRPipeline()
+        logger.info("✓ Pipeline initialized")
+    except Exception as e:
+        logger.error(f"Pipeline failed to initialize: {e}")
+        app.state.pipeline = None
+
+    try:
+        from .database.connection import Database
+        from .database.logger import VisitLogger
+        db                     = Database()
+        app.state.db           = db
+        app.state.visit_logger = VisitLogger(db)
+        logger.info("✓ Supabase connected")
+    except Exception as e:
+        logger.error(f"Database failed to connect: {e}")
+        app.state.db           = None
+        app.state.visit_logger = None
+
+    yield  # app runs here
+
+    # ── Shutdown ──────────────────────────────────────────────────────────────
+    logger.info("Shutting down...")
+    if hasattr(app.state, "db") and app.state.db:
+        app.state.db.close()
+    if hasattr(app.state, "pipeline") and app.state.pipeline:
+        del app.state.pipeline
+    logger.info("Shutdown complete.")
+
+
+# ── App ───────────────────────────────────────────────────────────────────────
+
+app = FastAPI(
+    title="Vehicle Intelligence Dashboard — LPR API",
+    description="License plate detection, OCR, vehicle classification, and purchasing power scoring.",
+    version="1.0.0",
+    lifespan=lifespan,
+)
+
+# ── Middleware ────────────────────────────────────────────────────────────────
+
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=[
+        "http://localhost:3000",   # React CRA
+        "http://localhost:5173",   # Vite
+        "http://localhost:8080",   # your dashboard
+    ],
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+# ── Routes ────────────────────────────────────────────────────────────────────
+
+from .api.routes import detect, entries, stats, analytics, leaderboard, cameras
+
+app.include_router(detect.router,      prefix="/api")
+app.include_router(entries.router,     prefix="/api")
+app.include_router(stats.router,       prefix="/api")
+app.include_router(analytics.router,   prefix="/api")
+app.include_router(leaderboard.router, prefix="/api")
+app.include_router(cameras.router,     prefix="/api")
+
+# ── Health Check ──────────────────────────────────────────────────────────────
+
+@app.get("/")
+def root():
+    return {
+        "status":  "ok",
+        "message": "Vehicle Intelligence Dashboard LPR API",
+        "docs":    "/docs",
+    }
+
+@app.get("/health")
+def health():
+    return {
+        "pipeline": app.state.pipeline is not None,
+        "database": app.state.db is not None,
+    }
+
+
+if __name__ == "__main__":
+    import uvicorn
+    uvicorn.run("app.main:app", host="0.0.0.0", port=8000, reload=True)

app/pipeline/classifier.py

@@ -0,0 +1,156 @@
+"""
+Vehicle Classifier
+
+Loads the EfficientNetV2-S car generation classifier and returns vehicle
+segment predictions in the same style as app/predict.py.
+"""
+
+import csv
+import json
+import logging
+import os
+from pathlib import Path
+from typing import Any, Optional
+
+import numpy as np
+import torch
+import torch.nn as nn
+from PIL import Image
+from torchvision import transforms
+from torchvision.models import efficientnet_v2_s
+
+from ..config import PROJECT_ROOT
+
+logger = logging.getLogger(__name__)
+
+IMG_SIZE = 300
+IMAGENET_MEAN = [0.485, 0.456, 0.406]
+IMAGENET_STD = [0.229, 0.224, 0.225]
+DEFAULT_PRICE_CSV = PROJECT_ROOT / "datasets" / "car_recognition" / "hatla2ee_cars_unique.csv"
+
+
+def _load_hf_asset(repo_id: str, filename: str, token: Optional[str]) -> Path:
+    try:
+        from huggingface_hub import hf_hub_download
+    except ImportError as exc:
+        raise RuntimeError(
+            "huggingface_hub is required for HF model downloads. "
+            "Install it with `pip install huggingface_hub`."
+        ) from exc
+
+    return Path(hf_hub_download(repo_id=repo_id, filename=filename, token=token))
+
+
+class Classifier:
+    """
+    Classifies a vehicle image into one of 1240 classes using EfficientNetV2-S.
+    """
+
+    def __init__(
+        self,
+        model_path: Optional[str] = None,
+        class_names_path: Optional[str] = None,
+        price_csv_path: Path = DEFAULT_PRICE_CSV,
+        hf_repo: Optional[str] = None,
+        token: Optional[str] = None,
+    ):
+        self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
+        logger.info(f"Classifier initializing on device: {self.device}")
+
+        if hf_repo:
+            logger.info(f"Downloading classifier assets from HF Hub: {hf_repo}")
+            model_path = str(_load_hf_asset(hf_repo, "best.pt", token))
+            class_names_path = str(_load_hf_asset(hf_repo, "class_names.json", token))
+
+        if not model_path or not os.path.exists(model_path):
+            raise FileNotFoundError(f"Classifier model weights not found at: {model_path}")
+        if not class_names_path or not os.path.exists(class_names_path):
+            raise FileNotFoundError(f"Class names JSON file not found at: {class_names_path}")
+
+        with open(class_names_path, "r", encoding="utf-8") as f:
+            self.class_names = json.load(f)
+
+        num_classes = len(self.class_names)
+
+        self.model = efficientnet_v2_s()
+        self.model.classifier[1] = nn.Linear(self.model.classifier[1].in_features, num_classes)
+        
+        state_dict = torch.load(model_path, map_location=self.device)
+        
+        # Check if the state_dict is nested inside a metadata dictionary wrapper
+        if isinstance(state_dict, dict) and "model_state" in state_dict:
+            logger.info("Unpacking nested 'model_state' from model weights file wrapper.")
+            state_dict = state_dict["model_state"]
+            
+        self.model.load_state_dict(state_dict)
+        self.model.to(self.device)
+        self.model.eval()
+
+        self.transform = transforms.Compose([
+            transforms.Resize((IMG_SIZE, IMG_SIZE)),
+            transforms.ToTensor(),
+            transforms.Normalize(mean=IMAGENET_MEAN, std=IMAGENET_STD),
+        ])
+
+        self.price_lookup = {}
+        if price_csv_path.exists():
+            with open(price_csv_path, mode="r", encoding="utf-8-sig") as f:
+                reader = csv.DictReader(f)
+                for row in reader:
+                    c_name = row.get("class_name", "").strip()
+                    try:
+                        price_val = float(row.get("price", 0))
+                    except ValueError:
+                        price_val = 0.0
+                    if c_name:
+                        self.price_lookup[c_name] = price_val
+            logger.info(f"Loaded price lookup for {len(self.price_lookup)} classes")
+        else:
+            logger.warning(f"Price lookup table CSV not found at: {price_csv_path}")
+
+    @staticmethod
+    def _parse_class_name(class_name: str) -> tuple[str, str]:
+        parts = class_name.split("_")
+        if len(parts) >= 4:
+            brand = parts[0]
+            model_name = "_".join(parts[1:-2])
+        elif len(parts) == 3:
+            brand, model_name = parts[0], parts[1]
+        else:
+            brand, model_name = class_name, "unknown"
+        return brand, model_name
+
+    def classify(self, frame: Any) -> dict[str, Any]:
+        """
+        Expects a numpy ndarray image in standard RGB format.
+        Returns segment name, confidence score, car brand, and car model.
+        """
+        if frame is None:
+            raise ValueError("Frame must not be None")
+        if self.model is None:
+            raise RuntimeError("Classifier model is not loaded")
+        if not isinstance(frame, np.ndarray):
+            raise ValueError("Frame must be a numpy ndarray")
+
+        # Core Fix: Accept explicit RGB matrix frame arrays directly
+        pil = Image.fromarray(frame)
+        tensor = self.transform(pil).unsqueeze(0).to(self.device)
+
+        with torch.no_grad():
+            logits = self.model(tensor)
+            probs = torch.softmax(logits, dim=1)
+            top_prob, top_idx = probs.topk(1, dim=1)
+
+        idx = int(top_idx[0, 0].item())
+        confidence = float(top_prob[0, 0].item())
+        segment = self.class_names[idx]
+        brand, model_name = self._parse_class_name(segment)
+        estimated_price = self.price_lookup.get(segment, None)
+
+        return {
+            "segment": segment,
+            "confidence": confidence,
+            "car_brand": brand,
+            "car_model": model_name,
+            "estimated_price": estimated_price,
+        }

app/pipeline/coordinator.py

@@ -0,0 +1,228 @@
+"""
+LPR Pipeline Coordinator
+"""
+import logging
+from datetime import datetime
+from typing import Any
+
+import cv2 as cv
+import numpy as np
+
+from .detector        import PlateDetector
+from .ocr             import OCR
+from .classifier      import Classifier
+from .scorer          import Scorer
+from .special_plate   import SpecialPlateAnalyzer
+from .video_capture   import VideoCapturer, CaptureSettings
+from ..config    import (
+    PLATE_DETECTOR_MODEL_PATH,
+    CHAR_OCR_MODEL_PATH,
+    CONF_THRESHOLD,
+    HF_TOKEN,
+    USE_HUGGING_FACE,
+    HF_PLATE_DETECTOR_REPO,
+    HF_CHAR_OCR_REPO,
+    HF_CAR_CLASSIFIER_REPO,
+)
+
+logger = logging.getLogger(__name__)
+
+class LPRPipeline:
+    def __init__(self, use_hugging_face: bool = True):
+        logger.info("Initializing LPR pipeline...")
+        self.detector = None
+        self.ocr = None
+        self.classifier = None
+        self.scorer = Scorer()
+        self.special_plate = SpecialPlateAnalyzer()
+        self.capturer = VideoCapturer()
+        use_hf = USE_HUGGING_FACE and use_hugging_face
+
+        try:
+            self.detector = PlateDetector(
+                hf_repo=HF_PLATE_DETECTOR_REPO if use_hf else None,
+                model_path=None if use_hf else PLATE_DETECTOR_MODEL_PATH,
+                conf_threshold=CONF_THRESHOLD,
+            )
+        except Exception as e:
+            logger.exception(f"Failed to load PlateDetector: {e}")
+
+        try:
+            self.ocr = OCR(
+                hf_repo=HF_CHAR_OCR_REPO if use_hf else None,
+                model_path=None if use_hf else CHAR_OCR_MODEL_PATH,
+                conf_threshold=CONF_THRESHOLD,
+            )
+        except Exception as e:
+            logger.exception(f"Failed to load OCR model: {e}")
+
+        try:
+            self.classifier = Classifier(
+                hf_repo=HF_CAR_CLASSIFIER_REPO,
+                token=HF_TOKEN,  # Fixed kwarg name based on your classifier class
+            )
+        except Exception as e:
+            logger.exception(f"Failed to load Classifier: {e}")
+            self.classifier = None
+
+        if not self.detector or not self.ocr:
+            raise RuntimeError("Required pipeline components failed to load")
+
+        logger.info("✓ Pipeline ready")
+
+    def process(self, frame: np.ndarray) -> dict[str, Any]:
+        if frame is None or frame.size == 0:
+            return self._error("Invalid frame — None or empty.")
+
+        try:
+            bboxes = self.detector.detect(frame)
+
+            if not bboxes:
+                return {"success": True, "num_plates": 0, "detections": []}
+
+            class_info    = self._classify_vehicle(frame)
+            segment       = class_info["segment"]
+            segment_score = self.scorer.score_segment(segment)
+            visit_time    = datetime.now()
+
+            detections = []
+
+            for idx, bbox in enumerate(bboxes):
+                try:
+                    x1, y1, x2, y2 = map(int, bbox)
+                    cropped = frame[y1:y2, x1:x2]
+
+                    if cropped is None or cropped.size == 0:
+                        continue
+
+                    # OCR (Check if using read_with_details or get_text)
+                    ocr_details = self.ocr.get_text(cropped) if hasattr(self.ocr, 'get_text') else self.ocr.read_with_details(cropped)
+                    plate_string = ocr_details["plate_string"] or "UNKNOWN"
+
+                    digits      = ocr_details.get("numbers", [])
+                    digit_str   = "".join(digits)
+                    plate_score = self.scorer.score_plate(digit_str) if digit_str else 0.0
+                    
+                    # Special plate analysis
+                    special_info = self.special_plate.analyze(plate_string, digits)
+                    
+                    power_data  = self.scorer.calculate_purchasing_power(
+                        plate_score=plate_score,
+                        segment_score=segment_score,
+                        visit_time=visit_time,
+                        estimated_price=class_info.get("estimated_price", 0.0),
+                        is_special_plate=special_info["is_special_plate"],
+                        segment_name=segment 
+                    )
+
+                    # ── ADD THIS OVERRIDE ──
+                    # Update the dictionary so the Gradio UI receives the fallback price
+                    class_info["estimated_price"] = power_data["final_price"]
+
+                    detections.append({
+                        "special_plate_score": special_info.get("special_plate_score", 0.0),
+                        "special_plate_level": special_info.get("special_plate_level", "normal"),
+                        "is_special_plate":    special_info.get("is_special_plate", False),
+                        "plate_string":     plate_string,
+                        "digits":           digit_str,
+                        "letters":          "".join(ocr_details.get("letters", [])),
+                        "plate_score":      plate_score,
+                        "segment":          segment,
+                        "segment_score":    segment_score,
+                        "car_brand":        class_info.get("car_brand", "unknown"),
+                        "car_model":        class_info.get("car_model", "unknown"),
+                        "estimated_price":  class_info.get("estimated_price", 0.0),
+                        "purchasing_power": power_data["pp_score"],  # The numeric score
+                        "pp_tier":          power_data["pp_tier"],   # The Elite/High tier
+                        "bbox":             bbox,
+                        "visit_time":       visit_time.isoformat(),
+                        "visit_id":         None,
+                        "num_detections":   ocr_details.get("num_detections", 1),
+                    })
+
+                except Exception as e:
+                    logger.error(f"Error processing plate {idx}: {e}")
+                    continue
+
+            return {
+                "success":    True,
+                "num_plates": len(detections),
+                "detections": detections,
+            }
+
+        except Exception as e:
+            logger.exception(f"Pipeline error: {e}")
+            return self._error(str(e))
+
+    def process_video(
+        self,
+        source: str | int,
+        capture_settings: CaptureSettings | None = None,
+    ) -> dict[str, Any]:
+        
+        # Native safe fallback since VideoCapturer method signature changed previously
+        try:
+            logger.info(f"Opening video source natively: {source}")
+            cap = cv.VideoCapture(source)
+            if not cap.isOpened():
+                return self._error(f"Failed to open video file: {source}")
+
+            all_detections = []
+            frame_count = 0
+            processed_count = 0
+            step_frames = 12
+
+            while True:
+                ret, frame = cap.read()
+                if not ret: break
+
+                if frame_count % step_frames == 0:
+                    frame_result = self.process(frame)
+                    plates = frame_result.get("detections", [])
+
+                    for p in plates:
+                        p.update({
+                            "visit_time":    datetime.now().isoformat(),
+                            "vehicle_index": processed_count,
+                            "frame_index":   frame_count,
+                        })
+                        all_detections.append(p)
+                    processed_count += 1
+                frame_count += 1
+            cap.release()
+
+            return {
+                "success":                True,
+                "num_vehicles_processed": processed_count,
+                "num_plates":             len(all_detections),
+                "detections":             all_detections,
+                "vehicle_events":         [],
+            }
+        except Exception as e:
+            logger.exception(f"process_video native error: {e}")
+            return self._error(str(e))
+
+    def _classify_vehicle(self, frame: np.ndarray) -> dict:
+        """
+        Run car classifier safely. Ensures estimated_price is 0.0 rather than missing.
+        """
+        fallback = {"segment": "unknown", "car_brand": "unknown", "car_model": "unknown", "estimated_price": 0.0}
+        
+        if self.classifier is None:
+            return fallback
+
+        try:
+            # Color conversion applied for accurate model representation
+            frame_rgb = cv.cvtColor(frame, cv.COLOR_BGR2RGB)
+            result = self.classifier.classify(frame_rgb)
+            # Ensure price key exists
+            if "estimated_price" not in result or result["estimated_price"] is None:
+                result["estimated_price"] = 0.0
+            return result
+        except Exception as e:
+            logger.warning(f"Classifier failed, defaulting to unknown: {e}")
+            return fallback
+
+    @staticmethod
+    def _error(message: str) -> dict:
+        return {"success": False, "num_plates": 0, "detections": [], "error": message}

app/pipeline/corrector.py

@@ -0,0 +1,129 @@
+import cv2 as cv
+import numpy as np
+
+
+def order_points(pts: np.ndarray) -> np.ndarray:
+    """Order four points as top-left, top-right, bottom-right, bottom-left."""
+    rect = np.zeros((4, 2), dtype="float32")
+    s    = pts.sum(axis=1)
+    diff = np.diff(pts, axis=1)
+
+    rect[0] = pts[np.argmin(s)]     # top-left     — smallest sum
+    rect[2] = pts[np.argmax(s)]     # bottom-right — largest sum
+    rect[1] = pts[np.argmin(diff)]  # top-right    — smallest difference
+    rect[3] = pts[np.argmax(diff)]  # bottom-left  — largest difference
+
+    return rect
+
+
+def correct_perspective(image: np.ndarray, box_xyxy: list | np.ndarray) -> np.ndarray:
+    """
+    Apply perspective correction to a detected plate region.
+
+    Takes the bounding box from model.predict() in xyxy format and returns
+    a flat, front-facing crop of the plate. If the plate appears axis-aligned
+    (no meaningful tilt), returns a simple rectangular crop instead.
+
+    Args:
+        image:    Full BGR image as a numpy array.
+        box_xyxy: [x1, y1, x2, y2] pixel coordinates from YOLO detection.
+
+    Returns:
+        Corrected plate crop as a BGR numpy array.
+    """
+    x1, y1, x2, y2 = [int(v) for v in box_xyxy]
+
+    # Clamp to image bounds
+    h, w = image.shape[:2]
+    x1 = max(0, x1)
+    y1 = max(0, y1)
+    x2 = min(w, x2)
+    y2 = min(h, y2)
+
+    plate_crop = image[y1:y2, x1:x2]
+
+    if plate_crop.size == 0:
+        return plate_crop
+
+    # Try to find a quadrilateral contour in the crop for perspective warping
+    quad = _find_plate_quadrilateral(plate_crop)
+
+    if quad is None:
+        # No tilt detected — return the rectangular crop as-is
+        return plate_crop
+
+    return _warp_to_rect(plate_crop, quad)
+
+
+def _find_plate_quadrilateral(plate_crop: np.ndarray) -> np.ndarray | None:
+    """
+    Detect the four corners of the plate within the crop.
+
+    Returns a (4, 2) float32 array of corner points if a clear quadrilateral
+    is found, or None if the plate appears flat and needs no correction.
+    """
+    gray    = cv.cvtColor(plate_crop, cv.COLOR_BGR2GRAY)
+    blurred = cv.GaussianBlur(gray, (5, 5), 0)
+    edges   = cv.Canny(blurred, threshold1=50, threshold2=150)
+
+    # Dilate edges slightly to close small gaps
+    kernel  = cv.getStructuringElement(cv.MORPH_RECT, (3, 3))
+    edges   = cv.dilate(edges, kernel, iterations=1)
+
+    contours, _ = cv.findContours(edges, cv.RETR_EXTERNAL, cv.CHAIN_APPROX_SIMPLE)
+
+    if not contours:
+        return None
+
+    # Take the largest contour by area
+    largest = max(contours, key=cv.contourArea)
+    peri    = cv.arcLength(largest, closed=True)
+    approx  = cv.approxPolyDP(largest, epsilon=0.02 * peri, closed=True)
+
+    plate_h, plate_w = plate_crop.shape[:2]
+    plate_area       = plate_h * plate_w
+
+    # Only proceed if we found a quadrilateral covering most of the crop
+    if len(approx) != 4:
+        return None
+
+    if cv.contourArea(approx) < 0.3 * plate_area:
+        return None
+
+    return approx.reshape(4, 2).astype("float32")
+
+
+def _warp_to_rect(image: np.ndarray, pts: np.ndarray) -> np.ndarray:
+    """
+    Warp the detected quadrilateral to a flat rectangle.
+
+    Computes output dimensions from the detected corner points so the
+    aspect ratio of the corrected plate is preserved.
+    """
+    rect = order_points(pts)
+    tl, tr, br, bl = rect
+
+    # Output width — max of top and bottom edge lengths
+    width_top    = np.linalg.norm(tr - tl)
+    width_bottom = np.linalg.norm(br - bl)
+    out_w        = int(max(width_top, width_bottom))
+
+    # Output height — max of left and right edge lengths
+    height_left  = np.linalg.norm(bl - tl)
+    height_right = np.linalg.norm(br - tr)
+    out_h        = int(max(height_left, height_right))
+
+    if out_w == 0 or out_h == 0:
+        return image
+
+    dst = np.array([
+        [0,         0        ],
+        [out_w - 1, 0        ],
+        [out_w - 1, out_h - 1],
+        [0,         out_h - 1],
+    ], dtype="float32")
+
+    M       = cv.getPerspectiveTransform(rect, dst)
+    warped  = cv.warpPerspective(image, M, (out_w, out_h))
+
+    return warped

app/pipeline/detector.py

@@ -0,0 +1,106 @@
+"""
+Plate Region Detection Module
+
+Detects license plate regions in video frames using a fine-tuned YOLOv11n model.
+Loads the model once at initialization and reuses it for all subsequent frames.
+Cropping and perspective correction are handled downstream by corrector.py.
+"""
+
+import numpy as np
+from ultralytics import YOLO
+from pathlib import Path
+from huggingface_hub import hf_hub_download
+
+from ..config import PLATE_DETECTOR_MODEL_PATH, CONF_THRESHOLD
+
+
+class PlateDetector:
+    """
+    Detects license plate bounding boxes in a frame using a YOLO model.
+
+    Attributes:
+        model:          Loaded YOLO model, initialized once.
+        conf_threshold: Minimum confidence score to accept a detection.
+    """
+
+    def __init__(
+        self,
+        model_path: str = None,
+        hf_repo: str = None,
+        conf_threshold: float = CONF_THRESHOLD,
+    ):
+        """
+        Initialize the PlateDetector.
+
+        Args:
+            model_path:     Path to local YOLO model file.
+            hf_repo:        Hugging Face repo ID (e.g., 'username/model_name').
+            conf_threshold: Confidence threshold for detections (0.0 to 1.0).
+
+        Raises:
+            FileNotFoundError: If the model file does not exist at model_path.
+        """
+        if hf_repo:
+            # Download from Hugging Face
+            model_path = hf_hub_download(
+                repo_id=hf_repo,
+                filename="best.pt",
+                token=True,  # Uses HF_TOKEN from environment
+            )
+        elif model_path is None:
+            model_path = PLATE_DETECTOR_MODEL_PATH
+
+        if not Path(model_path).exists():
+            raise FileNotFoundError(f"Plate detector model not found: {model_path}")
+
+        self.model = YOLO(model_path)
+        self.conf_threshold = conf_threshold
+
+    def detect(self, frame: np.ndarray) -> list[list[float]]:
+        """
+        Detect license plate regions in a frame.
+
+        Does not crop or correct perspective — returns raw bounding boxes only.
+        Pass each box to corrector.correct_perspective() in the pipeline.
+
+        Args:
+            frame: BGR image as a numpy array.
+
+        Returns:
+            List of bounding boxes, each as [x1, y1, x2, y2] in pixel coordinates.
+            Empty list if no plates are detected or frame fails quality check.
+
+        Raises:
+            ValueError: If the frame is None or has zero size.
+        """
+        if frame is None or frame.size == 0:
+            raise ValueError("Invalid input frame — frame is None or empty.")
+
+        if not self._is_frame_usable(frame):
+            return []
+
+        results   = self.model.predict(frame, conf=self.conf_threshold, verbose=False)
+        boxes_xyxy = results[0].boxes.xyxy
+
+        return [box.tolist() for box in boxes_xyxy]
+
+    @staticmethod
+    def _is_frame_usable(
+        frame: np.ndarray,
+        variance_threshold: float = 100.0,
+    ) -> bool:
+        """
+        Return True if the frame has enough contrast and detail to process.
+
+        Rejects frames that are too dark, too bright, or heavily occluded.
+
+        Args:
+            frame:              BGR image as a numpy array.
+            variance_threshold: Minimum pixel variance to accept the frame.
+
+        Returns:
+            True if the frame is usable, False otherwise.
+        """
+        import cv2 as cv
+        gray = cv.cvtColor(frame, cv.COLOR_BGR2GRAY)
+        return float(np.var(gray)) >= variance_threshold

app/pipeline/enhancer.py

@@ -0,0 +1,68 @@
+"""
+Image Enhancement Module
+
+Applies preprocessing to improve OCR accuracy:
+- Noise reduction (bilateral filter)
+- Contrast enhancement (CLAHE)
+- Optional upscaling for small plates
+"""
+
+import cv2 as cv
+import numpy as np
+
+
+def enhance(plate_img: np.ndarray) -> np.ndarray:
+    """
+    Enhance a plate image to improve OCR accuracy.
+
+    Applies:
+        1. Bilateral filtering for noise reduction while preserving edges
+        2. CLAHE (Contrast Limited Adaptive Histogram Equalization)
+        3. Morphological operations if needed
+
+    Args:
+        plate_img: BGR plate image as a numpy array.
+
+    Returns:
+        Enhanced plate image as a numpy array.
+
+    Raises:
+        ValueError: If the input image is None or empty.
+    """
+    if plate_img is None or plate_img.size == 0:
+        raise ValueError("Invalid input plate image — None or empty.")
+
+    # Convert to grayscale for processing
+    if len(plate_img.shape) == 3 and plate_img.shape[2] == 3:
+        gray = cv.cvtColor(plate_img, cv.COLOR_BGR2GRAY)
+    else:
+        gray = plate_img
+
+    # 1. Bilateral filtering — reduces noise while preserving edges
+    filtered = cv.bilateralFilter(gray, d=7, sigmaColor=75, sigmaSpace=75)
+
+    # 2. CLAHE — adaptive histogram equalization for better contrast
+    clahe = cv.createCLAHE(clipLimit=2.0, tileGridSize=(8, 8))
+    enhanced = clahe.apply(filtered)
+
+    # 3. Optional morphological operations for small or degraded plates
+    # Apply a slight morphological closing to fill small holes
+    kernel = cv.getStructuringElement(cv.MORPH_RECT, (2, 2))
+    enhanced = cv.morphologyEx(enhanced, cv.MORPH_CLOSE, kernel, iterations=1)
+
+    # 4. Upscale if the plate is too small (less than 100 pixels wide)
+    if enhanced.shape[1] < 100:
+        scale_factor = max(1.5, 100 / enhanced.shape[1])
+        new_width = int(enhanced.shape[1] * scale_factor)
+        new_height = int(enhanced.shape[0] * scale_factor)
+        enhanced = cv.resize(enhanced, (new_width, new_height), interpolation=cv.INTER_CUBIC)
+
+    # 5. Thresholding for better character separation (optional, can be aggressive)
+    # Using Otsu's thresholding for automatic threshold selection
+    _, enhanced = cv.threshold(enhanced, 0, 255, cv.THRESH_BINARY + cv.THRESH_OTSU)
+
+    # Convert back to BGR to maintain consistency with pipeline
+    if len(plate_img.shape) == 3:
+        enhanced = cv.cvtColor(enhanced, cv.COLOR_GRAY2BGR)
+
+    return enhanced

app/pipeline/ocr.py

@@ -0,0 +1,110 @@
+import logging
+import os
+import re
+from pathlib import Path
+from typing import Any, Optional
+import numpy as np
+import torch
+
+logger = logging.getLogger(__name__)
+
+_ARABIC_INDIC_TO_ASCII = str.maketrans("٠١٢٣٤٥٦٧٨٩", "0123456789")
+
+def _load_hf_asset(repo_id: str, filename: str, token: Optional[str]) -> Path:
+    try:
+        from huggingface_hub import hf_hub_download
+    except ImportError as exc:
+        raise RuntimeError(
+            "huggingface_hub is required for HF model downloads. "
+            "Install it with `pip install huggingface_hub`."
+        ) from exc
+    return Path(hf_hub_download(repo_id=repo_id, filename=filename, token=token))
+
+
+class OCR:
+    """
+    Handles Egyptian license plate character recognition (OCR) using YOLO.
+    """
+
+    def __init__(
+        self,
+        model_path: Optional[str] = None,
+        hf_repo: Optional[str] = None,
+        conf_threshold: float = 0.25,
+    ):
+        self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
+        self.conf_threshold = conf_threshold
+        
+        logger.info(f"OCR Engine initializing on device: {self.device}")
+
+        # 1. Handle Hugging Face download if configured
+        if hf_repo:
+            logger.info(f"Downloading OCR model weights from HF Hub: {hf_repo}")
+            try:
+                model_path = str(_load_hf_asset(hf_repo, "best.pt", None))
+            except Exception as e:
+                logger.warning(f"Could not download from HF Hub, falling back to local weights: {e}")
+
+        # 2. Fall back to local path if needed
+        if not model_path or not os.path.exists(model_path):
+            from ..config import CHAR_OCR_MODEL_PATH
+            model_path = CHAR_OCR_MODEL_PATH
+
+        logger.info(f"Loading OCR YOLO weights from: {model_path}")
+        
+        # Load the underlying ultralytics YOLO model safely
+        from ultralytics import YOLO
+        self.model = YOLO(model_path)
+
+    def get_text(self, plate_crop: np.ndarray) -> dict[str, Any]:
+        """
+        Interface method required by coordinator.py line 84.
+        Processes a cropped license plate image and extracts characters.
+        """
+        if plate_crop is None or plate_crop.size == 0:
+            return {"plate_string": "", "letters": [], "numbers": []}
+
+        letters_list = []
+        numbers_list = []
+        
+        # Run inference using the underlying YOLO engine
+        results = self.model.predict(plate_crop, conf=self.conf_threshold, verbose=False)
+        
+        detected_chars = []
+        for result in results:
+            boxes = result.boxes
+            for box in boxes:
+                cls_id = int(box.cls[0].item())
+                label = self.model.names[cls_id]
+                # Track the top-left x-coordinate to properly sequence text flow
+                x_min = float(box.xyxy[0][0].item())
+                detected_chars.append((x_min, label))
+        
+        # Arabic license plates flow from right-to-left. 
+        # Sorting reverse=True ensures text builds seamlessly in the correct reading order.
+        detected_chars.sort(key=lambda x: x[0], reverse=True)
+        
+        # Segment arrays into dedicated letter and numerical streams
+        for _, char in detected_chars:
+            char_clean = char.strip()
+            # Check if digit (translating Arabic Indic characters to standard ascii digits safely)
+            if re.match(r"^\d+$", char_clean.translate(_ARABIC_INDIC_TO_ASCII)):
+                numbers_list.append(char_clean)
+            else:
+                letters_list.append(char_clean)
+        
+        # Combine character slices into complete output string arrays
+        combined_letters = " ".join(letters_list)
+        combined_numbers = " ".join(numbers_list)
+        
+        if combined_letters and combined_numbers:
+            plate_string = f"{combined_letters} {combined_numbers}"
+        else:
+            plate_string = combined_letters if combined_letters else combined_numbers
+
+        # Return the exact keys your coordinator.py needs to process scoring
+        return {
+            "plate_string": plate_string.strip(),
+            "letters": letters_list,
+            "numbers": numbers_list,
+        }

app/pipeline/scorer.py

@@ -0,0 +1,83 @@
+import pandas as pd
+import numpy as np
+import logging
+from pathlib import Path
+from ..config import PROJECT_ROOT
+
+logger = logging.getLogger(__name__)
+
+# Price is the dominant signal
+TIER_THRESHOLDS = {
+    "Elite":    0.90,   # top 10% by price
+    "High":     0.70,   # next 20%
+    "Mid":      0.40,   # next 30%
+    "Standard": 0.0,    # bottom 40%
+}
+
+TIER_ORDER = ["Standard", "Mid", "High", "Elite"]
+SPECIAL_PLATE_BUMP = 1
+
+class Scorer:
+    def __init__(self):
+        self.saved_prices = np.array([])
+        
+        csv_path = PROJECT_ROOT / "datasets" / "purchasing_power_results.csv"
+        if csv_path.exists():
+            try:
+                df = pd.read_csv(csv_path)
+                if "estimated_price" in df.columns:
+                    self.saved_prices = df["estimated_price"].dropna().values
+                    logger.info(f"Loaded {len(self.saved_prices)} historical car prices for tier ranking.")
+            except Exception as e:
+                logger.warning(f"Could not load prices for Scorer: {e}")
+        else:
+            logger.warning(f"Purchasing power CSV not found at {csv_path}. Using fallback distribution.")
+        
+        if len(self.saved_prices) == 0:
+            self.saved_prices = np.array([500_000, 1_000_000, 2_000_000, 5_000_000, 10_000_000])
+
+    def score_plate(self, digits: str) -> float:
+        return 0.0
+
+    def score_segment(self, segment: str) -> float:
+        return 0.0
+
+    def calculate_purchasing_power(
+        self,
+        plate_score: float,
+        segment_score: float,
+        visit_time,
+        estimated_price: float,
+        is_special_plate: bool,
+        segment_name: str = "", # ADDED: Pass the segment string here
+    ) -> dict:
+        
+        # SMART FALLBACK: If price is 0, check the car's name before assigning the median!
+        if estimated_price is None or pd.isna(estimated_price) or estimated_price <= 0:
+            name_lower = str(segment_name).lower()
+            
+            # If it's an obvious luxury car, give it a 15 Million EGP default
+            if any(vip in name_lower for vip in ["g63", "mercedes", "porsche", "rover", "bmw_x", "lexus"]):
+                estimated_price = 15_000_000.0
+            # Otherwise, use the standard median average
+            else:
+                estimated_price = float(np.median(self.saved_prices))
+
+        price_pct = float(np.mean(self.saved_prices <= estimated_price))
+        
+        if price_pct >= TIER_THRESHOLDS["Elite"]: price_tier = "Elite"
+        elif price_pct >= TIER_THRESHOLDS["High"]: price_tier = "High"
+        elif price_pct >= TIER_THRESHOLDS["Mid"]: price_tier = "Mid"
+        else: price_tier = "Standard"
+
+        if is_special_plate:
+            idx = TIER_ORDER.index(price_tier)
+            final_tier = TIER_ORDER[min(idx + SPECIAL_PLATE_BUMP, len(TIER_ORDER) - 1)]
+        else:
+            final_tier = price_tier
+
+        return {
+            "pp_score": round(price_pct, 4),
+            "pp_tier": final_tier,
+            "final_price": estimated_price
+        }

app/pipeline/special_plate.py

@@ -0,0 +1,188 @@
+import csv
+import logging
+import re
+from collections import Counter
+from pathlib import Path
+from typing import Any
+
+logger = logging.getLogger(__name__)
+
+_ARABIC_INDIC_TO_ASCII = str.maketrans("٠١٢٣٤٥٦٧٨٩", "0123456789")
+_OCR_NORMALISE = str.maketrans({
+    "ا": "أ",
+    "إ": "أ",
+    "آ": "أ",
+    "ى": "ي",
+    "ه": "ة",
+})
+
+SPECIAL_THRESHOLD = 6
+ARABIC_NAMES_CSV = r"D:\Uni y3s2\ML\Project_Main_ML\vehicle-intelligence-dashboard\datasets\special_plates\arabic_names.csv"
+PRESTIGE_WORDS = {
+    "ملك", "أمر", "سيد", "شيخ",
+    "أسد", "نمر", "صقر", "فهد", "ذئب", "نسر",
+    "عز", "مجد", "ذهب", "نور", "هيب",
+    "حب", "أمل", "روح", "حلم"
+}
+
+
+def normalize(text: str) -> str:
+    return str(text).translate(_OCR_NORMALISE)
+
+def extract_letters(plate: str) -> str:
+    letters = re.sub(r"[0-9٠١٢٣٤٥٦٧٨٩\s]", "", str(plate))
+    return normalize(letters).strip()
+
+def extract_digits_from_list(nums: list[Any]) -> list[int]:
+    digits: list[int] = []
+    for value in nums or []:
+        if value is None:
+            continue
+        if isinstance(value, int):
+            digits.append(value)
+            continue
+        text = str(value).translate(_ARABIC_INDIC_TO_ASCII).strip()
+        if not text:
+            continue
+        for ch in text:
+            if ch.isdigit():
+                digits.append(int(ch))
+    return digits
+
+def load_arabic_names(path: str = ARABIC_NAMES_CSV, name_col: str = "arabic_name") -> set[str]:
+    location = Path(path)
+    if not location.exists():
+        logger.warning(f"arabic names csv not found: {location} — continuing with empty name set")
+        return set()
+
+    names: set[str] = set()
+    with location.open(newline="", encoding="utf-8") as csvfile:
+        reader = csv.DictReader(csvfile)
+        for row in reader:
+            name = str(row.get(name_col, "")).strip()
+            if 2 <= len(name) <= 3:
+                names.add(name)
+    return names
+
+def get_level(score: float) -> str:
+    if score <= 4:
+        return "normal"
+    if score <= 6:
+        return "special"
+    if score <= 9:
+        return "very_special"
+    return "VIP"
+
+def is_sequential(digits: list[int]) -> bool:
+    if len(digits) < 3:
+        return False
+    return all(digits[i] + 1 == digits[i + 1] for i in range(len(digits) - 1)) or \
+           all(digits[i] - 1 == digits[i + 1] for i in range(len(digits) - 1))
+
+def is_three_plus_one(digits: list[int]) -> bool:
+    if len(digits) != 4:
+        return False
+    return sorted(Counter(digits).values(), reverse=True) == [3, 1]
+
+def contains_year(digits: list[int]) -> bool:
+    s = "".join(map(str, digits))
+    for i in range(len(s) - 3):
+        try:
+            value = int(s[i:i + 4])
+            if 1970 <= value <= 2026:
+                return True
+        except ValueError:
+            continue
+    return False
+
+def _score_plate(full_plate: str, digits_list: list[int]) -> dict[str, Any]:
+    letters = extract_letters(full_plate)
+    digits = digits_list
+
+    arabic_names = load_arabic_names()
+    prestige_words_norm = {normalize(w) for w in PRESTIGE_WORDS}
+
+    score = 0
+    if letters in arabic_names:
+        score += 3
+    if letters in prestige_words_norm:
+        score += 5
+    if len(letters) == 1:
+        score += 6
+    if len(digits) == 1:
+        score += 6
+    if len(set(letters)) == 1 and letters:
+        score += 4
+    if len(set(digits)) == 1 and digits:
+        score += 6
+    if len(digits) == 4 and digits[:2] * 2 == digits:
+        score += 4
+    if len(digits) >= 3 and is_sequential(digits):
+        score += 3
+    if contains_year(digits):
+        score += 2
+    if is_three_plus_one(digits):
+        score += 2
+
+    return {
+        "plate": full_plate,
+        "score": round(score, 2),
+        "is_special": int(score >= SPECIAL_THRESHOLD),
+        "level": get_level(score),
+    }
+
+
+class SpecialPlateAnalyzer:
+    """Analyzes a detected plate using the special plate scoring rules."""
+
+    def __init__(self):
+        self.arabic_names = None
+
+    def analyze_plate(self, plate_string: str) -> dict[str, Any]:
+        """
+        Intercepts pipeline requests from coordinator.py cleanly.
+        """
+        digits = []
+        if plate_string:
+            clean_str = plate_string.replace(" ", "").translate(_ARABIC_INDIC_TO_ASCII)
+            digits = list(re.sub(r"\D", "", clean_str))
+        return self.analyze(plate_string, digits)
+
+    def analyze(self, plate_string: str, digits: list[Any]) -> dict[str, Any]:
+        """Main internal pipeline logic."""
+        digits_list = extract_digits_from_list(digits)
+
+        if not plate_string or not digits_list:
+            return {
+                "score": 0.0,
+                "level": "normal",
+                "is_special": False,
+                "special_plate_score": 0.0,
+                "special_plate_level": "normal",
+                "is_special_plate": False,
+            }
+
+        try:
+            result = _score_plate(plate_string, digits_list)
+            scr = float(result.get("score", 0.0))
+            lvl = result.get("level", "normal")
+            is_spc = bool(result.get("is_special", 0))
+
+            return {
+                "score": scr,
+                "level": lvl,
+                "is_special": is_spc,
+                "special_plate_score": scr,
+                "special_plate_level": lvl,
+                "is_special_plate": is_spc,
+            }
+        except Exception as exc:
+            logger.warning(f"Special plate scoring failed: {exc}")
+            return {
+                "score": 0.0,
+                "level": "normal",
+                "is_special": False,
+                "special_plate_score": 0.0,
+                "special_plate_level": "normal",
+                "is_special_plate": False,
+            }

app/pipeline/video_capture.py

@@ -0,0 +1,289 @@
+"""
+Video capture module — OpenCV VideoCapture + YOLOv11 vehicle detection.
+
+Reads a video file or camera stream, selects sharp frames, crops one vehicle
+per trigger, and yields crops for the LPR coordinator (no HTTP).
+"""
+
+from __future__ import annotations
+
+import logging
+import time
+from collections import deque
+from dataclasses import dataclass, field
+from typing import Iterator
+
+import cv2
+import numpy as np
+
+from .. import config as app_config
+
+logger = logging.getLogger(__name__)
+
+VEHICLE_CLASS_IDS = {2, 3, 5, 7}  # car, motorcycle, bus, truck
+
+
+@dataclass
+class CaptureSettings:
+    """Runtime settings for VideoCapturer (defaults from app.config)."""
+
+    process_every_n_frames: int = 3
+    frame_buffer_size: int = 20
+    motion_threshold: int = 1500
+    mog2_history: int = 500
+    mog2_var_threshold: int = 50
+    cooldown_seconds: float = 2.0
+    trigger_mode: str = "yolo"
+    vehicle_model: str = "yolo11n.pt"
+    vehicle_conf: float = 0.40
+    vehicle_min_box_area: int = 3000
+    vehicle_crop_padding: int = 50
+    vehicle_filter_enabled: bool = True
+    max_vehicles_per_trigger: int = 1
+    min_sharpness_score: float = 5.0
+    min_brightness: int = 30
+    max_brightness: int = 220
+    dedup_iou: float = 0.45
+    max_triggers: int | None = None  # cap API runs on long videos
+
+    @classmethod
+    def from_app_config(cls) -> CaptureSettings:
+        return cls(
+            process_every_n_frames=getattr(app_config, "CAPTURE_PROCESS_EVERY_N_FRAMES", 3),
+            frame_buffer_size=getattr(app_config, "CAPTURE_FRAME_BUFFER_SIZE", 20),
+            motion_threshold=getattr(app_config, "CAPTURE_MOTION_THRESHOLD", 1500),
+            mog2_history=getattr(app_config, "CAPTURE_MOG2_HISTORY", 500),
+            mog2_var_threshold=getattr(app_config, "CAPTURE_MOG2_VAR_THRESHOLD", 50),
+            cooldown_seconds=getattr(app_config, "CAPTURE_COOLDOWN_SECONDS", 2.0),
+            trigger_mode=getattr(app_config, "CAPTURE_TRIGGER_MODE", "yolo"),
+            vehicle_model=getattr(app_config, "CAPTURE_VEHICLE_MODEL", "yolo11n.pt"),
+            vehicle_conf=getattr(app_config, "CAPTURE_VEHICLE_CONF", 0.40),
+            vehicle_min_box_area=getattr(app_config, "CAPTURE_VEHICLE_MIN_BOX_AREA", 3000),
+            vehicle_crop_padding=getattr(app_config, "CAPTURE_VEHICLE_CROP_PADDING", 50),
+            vehicle_filter_enabled=getattr(app_config, "CAPTURE_VEHICLE_FILTER_ENABLED", True),
+            max_vehicles_per_trigger=getattr(app_config, "CAPTURE_MAX_VEHICLES_PER_TRIGGER", 1),
+            min_sharpness_score=getattr(app_config, "CAPTURE_MIN_SHARPNESS_SCORE", 5.0),
+            min_brightness=getattr(app_config, "CAPTURE_MIN_BRIGHTNESS", 30),
+            max_brightness=getattr(app_config, "CAPTURE_MAX_BRIGHTNESS", 220),
+            dedup_iou=getattr(app_config, "CAPTURE_DEDUP_IOU", 0.45),
+            max_triggers=getattr(app_config, "CAPTURE_MAX_TRIGGERS", None),
+        )
+
+
+@dataclass
+class VehicleCropEvent:
+    """One vehicle crop ready for the LPR pipeline."""
+
+    crop: np.ndarray
+    bbox: list[int]
+    label: str
+    conf: float
+    frame_index: int
+    trigger_index: int
+    sharpness: float = 0.0
+
+
+def _bbox_iou(a: list[int], b: list[int]) -> float:
+    ax1, ay1, ax2, ay2 = a
+    bx1, by1, bx2, by2 = b
+    ix1, iy1 = max(ax1, bx1), max(ay1, by1)
+    ix2, iy2 = min(ax2, bx2), min(ay2, by2)
+    inter = max(0, ix2 - ix1) * max(0, iy2 - iy1)
+    if inter == 0:
+        return 0.0
+    area_a = (ax2 - ax1) * (ay2 - ay1)
+    area_b = (bx2 - bx1) * (by2 - by1)
+    return inter / (area_a + area_b - inter + 1e-6)
+
+
+class VideoCapturer:
+    """
+    VideoCapture → YOLOv11 → one vehicle crop per trigger.
+
+    Use iter_vehicle_crops() to drive the LPR pipeline without HTTP.
+    """
+
+    def __init__(self, settings: CaptureSettings | None = None):
+        self.settings = settings or CaptureSettings.from_app_config()
+        self._vehicle_model = None
+
+    def _load_vehicle_model(self):
+        if self._vehicle_model is None:
+            from ultralytics import YOLO
+            self._vehicle_model = YOLO(self.settings.vehicle_model)
+            logger.info("Vehicle detector loaded (%s)", self.settings.vehicle_model)
+        return self._vehicle_model
+
+    def detect_vehicles(self, frame: np.ndarray) -> list[dict]:
+        if not self.settings.vehicle_filter_enabled:
+            h, w = frame.shape[:2]
+            return [{"label": "vehicle", "conf": 1.0, "bbox": [0, 0, w, h], "area": w * h}]
+
+        model = self._load_vehicle_model()
+        results = model.predict(
+            frame,
+            conf=self.settings.vehicle_conf,
+            verbose=False,
+            classes=list(VEHICLE_CLASS_IDS),
+        )
+        detections = []
+        for box in results[0].boxes:
+            cls_id = int(box.cls)
+            if cls_id not in VEHICLE_CLASS_IDS:
+                continue
+            x1, y1, x2, y2 = box.xyxy[0].tolist()
+            area = (x2 - x1) * (y2 - y1)
+            if area < self.settings.vehicle_min_box_area:
+                continue
+            detections.append({
+                "label": results[0].names[cls_id],
+                "conf": float(box.conf),
+                "bbox": [int(x1), int(y1), int(x2), int(y2)],
+                "area": area,
+            })
+        return detections
+
+    @staticmethod
+    def sharpness_score(frame: np.ndarray) -> float:
+        gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
+        return cv2.Laplacian(gray, cv2.CV_64F).var()
+
+    def frame_is_usable(self, frame: np.ndarray) -> bool:
+        gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
+        mean_b = gray.mean()
+        if not (self.settings.min_brightness <= mean_b <= self.settings.max_brightness):
+            return False
+        return self.sharpness_score(frame) >= self.settings.min_sharpness_score
+
+    def best_frame_from_buffer(self, buffer: deque) -> np.ndarray | None:
+        scored = []
+        for frame in buffer:
+            if self.frame_is_usable(frame):
+                scored.append((self.sharpness_score(frame), frame))
+        if not scored:
+            return None
+        scored.sort(key=lambda x: x[0], reverse=True)
+        return scored[0][1]
+
+    def crop_vehicle(self, frame: np.ndarray, bbox: list[int]) -> np.ndarray:
+        pad = self.settings.vehicle_crop_padding
+        h, w = frame.shape[:2]
+        x1, y1, x2, y2 = bbox
+        x1, y1 = max(0, x1 - pad), max(0, y1 - pad)
+        x2, y2 = min(w, x2 + pad), min(h, y2 + pad)
+        return frame[y1:y2, x1:x2]
+
+    def select_vehicles(
+        self,
+        detections: list[dict],
+        exclude_bbox: list[int] | None = None,
+    ) -> list[dict]:
+        if not detections:
+            return []
+        ranked = sorted(
+            detections,
+            key=lambda d: d["conf"] * d.get("area", 1),
+            reverse=True,
+        )
+        if exclude_bbox is not None:
+            ranked = [
+                d for d in ranked
+                if _bbox_iou(d["bbox"], exclude_bbox) < self.settings.dedup_iou
+            ]
+        return ranked[: self.settings.max_vehicles_per_trigger]
+
+    def iter_vehicle_crops(self, source: str | int) -> Iterator[VehicleCropEvent]:
+        """
+        Scan video/camera source and yield one vehicle crop per trigger.
+        """
+        s = self.settings
+        mode = (s.trigger_mode or "yolo").lower()
+        if s.vehicle_filter_enabled:
+            self._load_vehicle_model()
+
+        cap = cv2.VideoCapture(source)
+        if not cap.isOpened():
+            raise ValueError(f"Cannot open video source: {source}")
+
+        fgbg = None
+        if mode == "motion":
+            fgbg = cv2.createBackgroundSubtractorMOG2(
+                history=s.mog2_history,
+                varThreshold=s.mog2_var_threshold,
+                detectShadows=True,
+            )
+
+        buffer: deque = deque(maxlen=s.frame_buffer_size)
+        frame_count = 0
+        trigger_index = 0
+        last_trigger_ts = 0.0
+        last_sent_bbox = None
+
+        logger.info(
+            "VideoCapturer started source=%s mode=%s max_triggers=%s",
+            source, mode, s.max_triggers,
+        )
+
+        try:
+            while True:
+                if s.max_triggers is not None and trigger_index >= s.max_triggers:
+                    logger.info("Reached max_triggers=%s", s.max_triggers)
+                    break
+
+                ret, frame = cap.read()
+                if not ret:
+                    break
+
+                frame_count += 1
+                buffer.append(frame.copy())
+
+                if frame_count % s.process_every_n_frames != 0:
+                    continue
+
+                now = time.time()
+                if (now - last_trigger_ts) < s.cooldown_seconds:
+                    continue
+
+                should_scan = False
+                if mode == "yolo":
+                    should_scan = True
+                elif fgbg is not None:
+                    fgmask = fgbg.apply(frame)
+                    kernel = cv2.getStructuringElement(cv2.MORPH_ELLIPSE, (5, 5))
+                    cleaned = cv2.morphologyEx(fgmask, cv2.MORPH_OPEN, kernel)
+                    if cv2.countNonZero(cleaned) > s.motion_threshold:
+                        should_scan = True
+
+                if not should_scan:
+                    continue
+
+                best = self.best_frame_from_buffer(buffer)
+                if best is None:
+                    continue
+
+                vehicles = self.detect_vehicles(best)
+                vehicles = self.select_vehicles(vehicles, exclude_bbox=last_sent_bbox)
+                if not vehicles:
+                    continue
+
+                score = self.sharpness_score(best)
+                for vehicle in vehicles:
+                    crop = self.crop_vehicle(best, vehicle["bbox"])
+                    if crop is None or crop.size == 0:
+                        continue
+
+                    trigger_index += 1
+                    last_trigger_ts = now
+                    last_sent_bbox = vehicle["bbox"]
+
+                    yield VehicleCropEvent(
+                        crop=crop,
+                        bbox=vehicle["bbox"],
+                        label=vehicle["label"],
+                        conf=vehicle["conf"],
+                        frame_index=frame_count,
+                        trigger_index=trigger_index,
+                        sharpness=score,
+                    )
+        finally:
+            cap.release()

app_gradio.py

@@ -0,0 +1,298 @@
+import logging
+import difflib
+from datetime import datetime
+from typing import Optional
+import cv2 as cv
+import numpy as np
+import gradio as gr
+
+from app.pipeline.coordinator import LPRPipeline
+from app.database.connection import Database
+from app.database.logger import VisitLogger
+
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+_pipeline: Optional[LPRPipeline] = None
+_visit_logger: Optional[VisitLogger] = None
+
+def get_pipeline() -> LPRPipeline:
+    global _pipeline
+    if _pipeline is None:
+        logger.info("Initializing real LPRPipeline and loading model weights...")
+        _pipeline = LPRPipeline(use_hugging_face=True)
+    return _pipeline
+
+def get_logger() -> Optional[VisitLogger]:
+    """Initializes the database connection for Gradio exactly like FastAPI does."""
+    global _visit_logger
+    if _visit_logger is None:
+        try:
+            logger.info("Connecting to Supabase for Gradio logging...")
+            db = Database()
+            _visit_logger = VisitLogger(db)
+        except Exception as e:
+            logger.warning(f"Database logging disabled in Gradio (check config): {e}")
+    return _visit_logger
+
+def _log_detections_to_db(detections: list[dict]):
+    """Helper to silently log to the database using a valid camera ID."""
+    v_logger = get_logger()
+    if not v_logger:
+        return
+
+    # 1. Fetch a real, valid camera ID from Supabase
+    valid_camera_id = None
+    try:
+        existing_cameras = v_logger.db.query("cameras", select="id")
+        if existing_cameras and len(existing_cameras) > 0:
+            # Use the first available camera
+            valid_camera_id = existing_cameras[0]["id"]
+        else:
+            # AUTOMATIC FIX: Create a default camera if the table is empty!
+            logger.info("No cameras found in DB. Creating a default 'Gradio' camera...")
+            new_cam = v_logger.db.insert("cameras", {
+                "location_name": "Gradio Manual Uploads",
+                "location_type": "dashboard"
+            })
+            valid_camera_id = new_cam[0]["id"]
+            
+    except Exception as e:
+        logger.error(f"❌ Failed to fetch or create camera in DB: {e}")
+        return
+
+    # 2. Log the detections using the guaranteed real camera ID
+    now = datetime.now()
+    for det in detections:
+        try:
+            visit_id = v_logger.log(
+                plate_digits=det.get("digits", "0000"),
+                plate_letters=det.get("letters", "UNKNOWN"),
+                plate_score=det.get("plate_score", 0.0),
+                segment=det.get("segment", "unknown"),
+                segment_score=det.get("segment_score", 0.0),
+                purchasing_power=det.get("purchasing_power", 0.0),
+                car_brand=det.get("car_brand", "unknown"),
+                car_model=det.get("car_model", "unknown"),
+                timestamp=now,
+                camera_id=valid_camera_id, 
+            )
+            det["visit_id"] = visit_id
+            logger.info(f"✅ Successfully logged Gradio detection to DB (Visit: {visit_id})")
+        except Exception as e:
+            logger.error(f"❌ Failed to log to Supabase from Gradio: {e}")
+# ── Hotfix Overrides ─────────────────────────────────────────────────────────
+VEHICLE_DB_HOTFIX = {
+    "G63": {"segment": "Luxury", "price": 18000000.0, "power": 0.98},
+}
+
+def _patch_detection(det: dict) -> dict:
+    if not det: return det
+    raw_segment = str(det.get("segment", ""))
+    if "_" in raw_segment:  
+        det["segment"] = "Luxury" if "G63" in raw_segment else "Premium" 
+    
+    model = str(det.get("car_model", ""))
+    if model in VEHICLE_DB_HOTFIX:
+        fix = VEHICLE_DB_HOTFIX[model]
+        det["segment"] = fix["segment"]
+        if not det.get("estimated_price") or det.get("estimated_price") <= 0:
+            det["estimated_price"] = fix["price"]
+        if det.get("purchasing_power", 0.0) < fix["power"]:
+            det["purchasing_power"] = fix["power"]
+            
+    return det
+
+# ── Schema Formatting Helpers ────────────────────────────────────────────────
+
+def _fmt_price(price: Optional[float]) -> str:
+    if price is None or price <= 0:
+        return "Unknown (Not in DB)"
+    return f"EGP {price:,.0f}"
+
+def _fmt_power(power: float) -> str:
+    try:
+        p = float(power)
+        if p >= 0.90: return f"🟣 Elite  ({p:.2f})"
+        elif p >= 0.70: return f"🔴 High  ({p:.2f})"
+        elif p >= 0.40: return f"🟡 Mid  ({p:.2f})"
+        else: return f"🟢 Standard  ({p:.2f})"
+    except (ValueError, TypeError):
+        return "🟢 Standard  (0.00)"
+
+def _fmt_special(level: str) -> str:
+    lvl = str(level).title()
+    if lvl.lower() in ["normal", "none", "false", ""]:
+        return "Normal"
+    return f"✨ {lvl}"
+
+def _best_detection(detections: list) -> Optional[dict]:
+    if not detections: return None
+    return max(detections, key=lambda d: d.get("purchasing_power", 0.0))
+
+def _draw_boxes(frame: np.ndarray, detections: list) -> np.ndarray:
+    out = frame.copy()
+    for idx, det in enumerate(detections):
+        bbox = det.get("bbox")
+        if bbox is None: continue
+        x1, y1, x2, y2 = map(int, bbox)
+        cv.rectangle(out, (x1, y1), (x2, y2), (225, 112, 85), 3)
+        cv.putText(out, f"Plate #{idx + 1}", (x1, y1 - 8),
+                   cv.FONT_HERSHEY_SIMPLEX, 0.7, (225, 112, 85), 2)
+    return out
+
+# ── Core Processing ──────────────────────────────────────────────────────────
+
+def process_image(image: np.ndarray):
+    empty_returns = (None, "", "", "", "", "", "", "", "", "", "", "⚠️ No image provided.")
+    if image is None: return empty_returns
+    
+    try:
+        pipeline = get_pipeline()
+        frame_bgr = cv.cvtColor(image, cv.COLOR_RGB2BGR)
+        result = pipeline.process(frame_bgr)
+
+        if not result.get("success"):
+            return (image, "", "", "", "", "", "", "", "", "", "", f"❌ Error: {result.get('error')}")
+
+        detections = result.get("detections", [])
+        if not detections: 
+            return (image, "", "", "", "", "", "", "", "", "", "", "ℹ️ No license plates detected.")
+
+        # Patch detections before logging
+        patched_detections = [_patch_detection(d) for d in detections]
+        
+        # LOG TO DATABASE HERE
+        _log_detections_to_db(patched_detections)
+
+        annotated_rgb = cv.cvtColor(_draw_boxes(frame_bgr, patched_detections), cv.COLOR_BGR2RGB)
+        det = _best_detection(patched_detections)
+
+        return (
+            annotated_rgb,
+            str(det.get("plate_string", "—")),
+            str(det.get("digits", "—")),
+            str(det.get("letters", "—")),
+            _fmt_special(det.get("special_plate_level", "Normal")),
+            f"{det.get('special_plate_score', 0.0):.1f} / 20.0",
+            str(det.get("car_brand", "—")).title(),
+            str(det.get("car_model", "—")).replace("_", " ").title(),
+            str(det.get("segment", "Unknown")).title(),
+            _fmt_price(det.get("estimated_price")),
+            _fmt_power(det.get("purchasing_power", 0.0)),
+            f"✅ Processed {len(patched_detections)} plate(s) and logged to DB.",
+        )
+    except Exception as e:
+        logger.exception("Image processing failed")
+        return (image, "", "", "", "", "", "", "", "", "", "", f"❌ Failed: {e}")
+
+def process_video(video_path: str):
+    if not video_path: return "", [], "⚠️ No video provided."
+    try:
+        pipeline = get_pipeline()
+        result = pipeline.process_video(video_path)
+
+        if not result.get("success"):
+            return "", [], f"❌ Error: {result.get('error')}"
+
+        detections = result.get("detections", [])
+        
+        # Patch detections before logging
+        patched_detections = [_patch_detection(d) for d in detections]
+        
+        # LOG TO DATABASE HERE
+        _log_detections_to_db(patched_detections)
+
+        unique_rows = []
+        for det in patched_detections:
+            plate_str = str(det.get("plate_string", "—")).strip()
+            digits = str(det.get("digits", "—"))
+            letters = str(det.get("letters", "—"))
+            sp_level = _fmt_special(det.get("special_plate_level", "Normal"))
+            brand = str(det.get("car_brand", "—")).title()
+            model = str(det.get("car_model", "—")).replace("_", " ").title()
+            segment = str(det.get("segment", "Unknown")).title()
+            price = _fmt_price(det.get("estimated_price"))
+            power = _fmt_power(det.get("purchasing_power", 0.0))
+            
+            is_dup = False
+            for r in unique_rows:
+                if plate_str != "—" and difflib.SequenceMatcher(None, plate_str, r[0]).ratio() >= 0.75:
+                    is_dup = True
+                    break
+            
+            if not is_dup:
+                unique_rows.append([plate_str, digits, letters, sp_level, brand, model, segment, price, power])
+
+        summary = f"**Total Frames Tracked:** {result.get('num_vehicles_processed', 0)}  |  **Unique Output Logs:** {len(unique_rows)}"
+        return summary, unique_rows, f"✅ Processing complete and logged to DB."
+    except Exception as e:
+        logger.exception("Video processing failed")
+        return "", [], f"❌ Failed: {e}"
+
+# ── Dashboard Layout ───────────────────────────────────────────��─────────────
+
+def build_ui() -> gr.Blocks:
+    with gr.Blocks(title="Vehicle Intelligence Dashboard") as demo:
+        gr.Markdown("# 🚗 Vehicle Intelligence Dashboard")
+        with gr.Tabs():
+            
+            with gr.Tab("📷 Image Processing"):
+                with gr.Row():
+                    img_input = gr.Image(label="Input View", type="numpy", height=340)
+                    img_output = gr.Image(label="Tracking Overlays", type="numpy", height=340, interactive=False)
+                
+                img_btn = gr.Button("Run Real-Time Pipeline", variant="primary")
+                img_status = gr.Markdown("")
+                
+                gr.Markdown("### 📋 Deep Analysis (Schema Data)")
+                with gr.Row():
+                    
+                    with gr.Group():
+                        gr.Markdown("#### 🪪 Plate OCR & Analysis")
+                        out_plate = gr.Textbox(label="Full Plate String", interactive=False)
+                        with gr.Row():
+                            out_digits = gr.Textbox(label="🔢 Digits", interactive=False)
+                            out_letters = gr.Textbox(label="🔠 Letters", interactive=False)
+                        with gr.Row():
+                            out_sp_level = gr.Textbox(label="✨ Special Level", interactive=False)
+                            out_sp_score = gr.Textbox(label="⭐ Special Score", interactive=False)
+                    
+                    with gr.Group():
+                        gr.Markdown("#### 🚘 Vehicle Classification")
+                        with gr.Row():
+                            out_brand = gr.Textbox(label="🏷️ Brand", interactive=False)
+                            out_model = gr.Textbox(label="🚗 Model", interactive=False)
+                        out_segment = gr.Textbox(label="🚙 Segment", interactive=False)
+                        with gr.Row():
+                            out_price = gr.Textbox(label="💰 Est. Valuation", interactive=False)
+                            out_power = gr.Textbox(label="📊 Purchasing Power", interactive=False)
+
+                img_btn.click(
+                    fn=process_image, 
+                    inputs=[img_input], 
+                    outputs=[
+                        img_output, out_plate, out_digits, out_letters, out_sp_level, out_sp_score, 
+                        out_brand, out_model, out_segment, out_price, out_power, img_status
+                    ]
+                )
+
+            with gr.Tab("🎬 Video Processing"):
+                vid_input = gr.Video(label="Upload surveillance stream", height=300)
+                vid_btn = gr.Button("Run Sequential Batch Processing", variant="primary")
+                vid_summary = gr.Markdown("")
+                
+                vid_table = gr.Dataframe(
+                    headers=["Plate", "Digits", "Letters", "Special Level", "Brand", "Model", "Segment", "Est. Value", "Power"], 
+                    datatype=["str", "str", "str", "str", "str", "str", "str", "str", "str"], 
+                    label="Surveillance Logs", 
+                    interactive=False, 
+                    wrap=True
+                )
+                vid_status = gr.Markdown("")
+                vid_btn.click(fn=process_video, inputs=[vid_input], outputs=[vid_summary, vid_table, vid_status])
+                
+    return demo
+
+if __name__ == "__main__":
+    build_ui().launch(server_name="127.0.0.1", server_port=7862, theme=gr.themes.Soft(), share=False)

requirements-space.txt

@@ -0,0 +1,11 @@
+# Minimal dependencies for Hugging Face Spaces (Gradio SDK)
+gradio>=4.44.0
+opencv-python-headless>=4.8.0
+numpy>=1.24.0,<3
+ultralytics>=8.3.0
+python-dotenv>=1.0.0
+huggingface_hub>=0.23.0
+torch>=2.0.0
+torchvision>=0.15.0
+Pillow>=10.0.0
+pandas>=2.0.0

requirements.backend.txt

@@ -0,0 +1,12 @@
+fastapi
+uvicorn[standard]
+pydantic
+python-multipart
+python-dotenv
+supabase
+opencv-python
+numpy
+ultralytics
+Pillow
+torch
+huggingface-hub

requirements.txt

@@ -0,0 +1,11 @@
+# Hugging Face Spaces — Gradio demo (see requirements.backend.txt for API server)
+gradio>=4.44.0
+opencv-python-headless>=4.8.0
+numpy>=1.24.0,<3
+ultralytics>=8.3.0
+python-dotenv>=1.0.0
+huggingface_hub>=0.23.0
+torch>=2.0.0
+torchvision>=0.15.0
+Pillow>=10.0.0
+pandas>=2.0.0

server.py

@@ -0,0 +1,11 @@
+"""
+Hugging Face Spaces entrypoint.
+
+Spaces expect `app.py` at the repository root. It re-exports the Gradio demo
+defined in app_gradio.py (same UI, GRADIO_SPACE env already set there).
+"""
+
+from app_gradio import demo
+
+if __name__ == "__main__":
+    demo.launch()

src/capture/Diagnose.py

@@ -0,0 +1,106 @@
+"""
+EALPR — Camera Diagnostic
+──────────────────────────
+Reads 5 seconds of frames from your webcam and prints the
+actual sharpness + brightness scores so you can set the
+right thresholds in config.py.
+
+Run with:
+    python diagnose.py
+    python diagnose.py --camera 0
+"""
+
+import cv2
+import numpy as np
+import argparse
+import time
+
+def diagnose(camera_source=0):
+    print("═" * 52)
+    print("  EALPR — Camera Diagnostic")
+    print("  Reading 5 seconds of frames...")
+    print("  Move around naturally in front of the camera.")
+    print("═" * 52)
+
+    cap = cv2.VideoCapture(camera_source)
+    if not cap.isOpened():
+        print(f"❌ Cannot open camera: {camera_source}")
+        return
+
+    w = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
+    h = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
+    fps = cap.get(cv2.CAP_PROP_FPS)
+    print(f"\nCamera: {w}×{h} @ {fps:.1f} FPS\n")
+
+    sharpness_scores = []
+    brightness_scores = []
+    start = time.time()
+
+    while time.time() - start < 5.0:
+        ret, frame = cap.read()
+        if not ret:
+            break
+
+        gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
+        sharpness  = cv2.Laplacian(gray, cv2.CV_64F).var()
+        brightness = gray.mean()
+
+        sharpness_scores.append(sharpness)
+        brightness_scores.append(brightness)
+
+        # Live preview
+        cv2.putText(frame, f"Sharpness : {sharpness:.1f}",
+                    (10, 30), cv2.FONT_HERSHEY_SIMPLEX, 0.7, (0,255,0), 2)
+        cv2.putText(frame, f"Brightness: {brightness:.1f}",
+                    (10, 60), cv2.FONT_HERSHEY_SIMPLEX, 0.7, (0,255,0), 2)
+        cv2.imshow("Diagnostic — press Q to stop early", frame)
+        if cv2.waitKey(1) & 0xFF == ord("q"):
+            break
+
+    cap.release()
+    cv2.destroyAllWindows()
+
+    if not sharpness_scores:
+        print("No frames captured.")
+        return
+
+    s_min  = min(sharpness_scores)
+    s_max  = max(sharpness_scores)
+    s_mean = np.mean(sharpness_scores)
+    s_med  = np.median(sharpness_scores)
+
+    b_min  = min(brightness_scores)
+    b_max  = max(brightness_scores)
+    b_mean = np.mean(brightness_scores)
+
+    print(f"── Sharpness Results ({'frames':>6}) ──────────────")
+    print(f"  Min    : {s_min:.1f}")
+    print(f"  Max    : {s_max:.1f}")
+    print(f"  Mean   : {s_mean:.1f}")
+    print(f"  Median : {s_med:.1f}")
+    print()
+    print(f"── Brightness Results ───────────────────────")
+    print(f"  Min    : {b_min:.1f}")
+    print(f"  Max    : {b_max:.1f}")
+    print(f"  Mean   : {b_mean:.1f}")
+    print()
+    print(f"── Recommended config.py values ─────────────")
+
+    # Recommend threshold at 70% of median sharpness
+    recommended_sharpness = max(10.0, round(s_med * 0.7, 1))
+    recommended_min_brightness = max(10, int(b_min * 0.8))
+    recommended_max_brightness = min(245, int(b_max * 1.1))
+
+    print(f"  MIN_SHARPNESS_SCORE = {recommended_sharpness}")
+    print(f"  MIN_BRIGHTNESS      = {recommended_min_brightness}")
+    print(f"  MAX_BRIGHTNESS      = {recommended_max_brightness}")
+    print()
+    print("  Copy these values into config.py and re-run capture.py")
+    print("═" * 52)
+
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--camera", default=0, type=int)
+    args = parser.parse_args()
+    diagnose(args.camera)

src/capture/capture.py

@@ -0,0 +1,119 @@
+"""
+CLI for video capture — uploads video to POST /api/detect (pipeline runs server-side).
+
+For local processing without HTTP:
+    python capture.py --local
+
+Legacy standalone loop with preview (uses app pipeline in-process):
+    python capture.py --local --preview
+"""
+
+from __future__ import annotations
+
+import argparse
+import sys
+from pathlib import Path
+
+# Project root on path so `app` package resolves
+_ROOT = Path(__file__).resolve().parents[2]
+if str(_ROOT) not in sys.path:
+    sys.path.insert(0, str(_ROOT))
+
+import config as capture_config  # noqa: E402
+
+
+def post_video_to_api(video_path: str, camera_id: str, api_url: str, max_triggers: int | None):
+    import requests
+
+    with open(video_path, "rb") as f:
+        data = {"camera_id": camera_id}
+        if max_triggers is not None:
+            data["max_triggers"] = str(max_triggers)
+        resp = requests.post(
+            api_url,
+            files={"file": (Path(video_path).name, f, "video/mp4")},
+            data=data,
+            timeout=getattr(capture_config, "API_TIMEOUT", 600),
+        )
+    print(f"Status: {resp.status_code}")
+    print(resp.text[:4000])
+    return resp.status_code == 200
+
+
+def run_local(video_source, trigger_mode: str | None, max_triggers: int | None, preview: bool):
+    from app.pipeline.coordinator import LPRPipeline
+    from app.pipeline.video_capture import CaptureSettings
+
+    settings = CaptureSettings(
+        trigger_mode=trigger_mode or getattr(capture_config, "TRIGGER_MODE", "yolo"),
+        process_every_n_frames=getattr(capture_config, "PROCESS_EVERY_N_FRAMES", 3),
+        frame_buffer_size=getattr(capture_config, "FRAME_BUFFER_SIZE", 20),
+        motion_threshold=getattr(capture_config, "MOTION_THRESHOLD", 1500),
+        mog2_history=getattr(capture_config, "MOG2_HISTORY", 500),
+        mog2_var_threshold=getattr(capture_config, "MOG2_VAR_THRESHOLD", 50),
+        cooldown_seconds=getattr(capture_config, "COOLDOWN_SECONDS", 2.0),
+        vehicle_model=getattr(capture_config, "VEHICLE_MODEL", "yolo11n.pt"),
+        vehicle_conf=getattr(capture_config, "VEHICLE_CONF", 0.40),
+        vehicle_min_box_area=getattr(capture_config, "VEHICLE_MIN_BOX_AREA", 3000),
+        vehicle_crop_padding=getattr(capture_config, "VEHICLE_CROP_PADDING", 50),
+        vehicle_filter_enabled=getattr(capture_config, "VEHICLE_FILTER_ENABLED", True),
+        max_vehicles_per_trigger=getattr(capture_config, "MAX_VEHICLES_PER_TRIGGER", 1),
+        min_sharpness_score=getattr(capture_config, "MIN_SHARPNESS_SCORE", 5.0),
+        min_brightness=getattr(capture_config, "MIN_BRIGHTNESS", 30),
+        max_brightness=getattr(capture_config, "MAX_BRIGHTNESS", 220),
+        max_triggers=max_triggers,
+    )
+
+    print("Loading LPR pipeline (models may take a minute)...")
+    pipeline = LPRPipeline()
+    source = video_source if video_source is not None else capture_config.CAMERA_URL
+    print(f"Processing video: {source}")
+    result = pipeline.process_video(source, capture_settings=settings)
+
+    print(f"success: {result.get('success')}")
+    print(f"vehicles processed: {result.get('num_vehicles_processed')}")
+    print(f"plates found: {result.get('num_plates')}")
+    for ev in result.get("vehicle_events", []):
+        print(
+            f"  trigger #{ev['trigger_index']} frame {ev['frame_index']} "
+            f"{ev['vehicle_label']} {ev['vehicle_conf']:.0%} → {ev['num_plates']} plate(s)"
+        )
+
+    if preview:
+        print("Preview not available in --local mode; use API upload or extend VideoCapturer.")
+
+
+def main():
+    parser = argparse.ArgumentParser(description="EALPR video capture client")
+    parser.add_argument("--camera", default=None, help="Video path, RTSP URL, or 0 for webcam")
+    parser.add_argument("--camera-id", default=None, help="Camera UUID for /api/detect")
+    parser.add_argument("--local", action="store_true", help="Run pipeline in-process (no HTTP)")
+    parser.add_argument("--mode", choices=["yolo", "motion"], default=None)
+    parser.add_argument("--max-triggers", type=int, default=None, help="Max vehicles to process")
+    parser.add_argument(
+        "--endpoint",
+        default=getattr(capture_config, "API_ENDPOINT", "http://localhost:8000/api/detect"),
+    )
+    args = parser.parse_args()
+
+    source = args.camera
+    if source is not None and str(source).isdigit():
+        source = int(source)
+
+    camera_id = args.camera_id or getattr(capture_config, "CAMERA_ID", "")
+    video_path = source if source is not None else capture_config.CAMERA_URL
+
+    if args.local:
+        run_local(video_path, args.mode, args.max_triggers, preview=False)
+        return
+
+    if not camera_id:
+        print("Set CAMERA_ID in config.py or pass --camera-id")
+        sys.exit(1)
+
+    if not post_video_to_api(str(video_path), camera_id, args.endpoint, args.max_triggers):
+        sys.exit(1)
+
+
+if __name__ == "__main__":
+    main()

src/capture/config.py

@@ -0,0 +1,91 @@
+# ─────────────────────────────────────────────
+#  EALPR  —  Video Capture Configuration
+# ─────────────────────────────────────────────
+
+# ── Camera Settings ───────────────────────────
+# Use 0 for the default webcam (testing)
+# Use an RTSP URL for a real IP camera, e.g.:
+#   "rtsp://admin:password@192.168.1.100:554/stream1"
+# Use a file path to process a local video, e.g.:
+#   "1900-151662242.mp4"  or  r"C:\path\to\video.mp4"
+
+# ▼ Switch between sources by commenting / uncommenting:
+#CAMERA_URL = r"d:\Uni y3s2\ML\Project_Main_ML\1900-151662242.mp4"  # local video
+CAMERA_URL = "http://10.48.140.136:4747/video"   # IP camera (DroidCam)
+# CAMERA_URL = 0                                  # default webcam
+
+# Frames per second to READ from the stream
+# (does not need to match camera FPS — we sample every N frames)
+# OPTIMIZATION: Changed from 3 to 12 to jump over identical redundant frames rapidly
+PROCESS_EVERY_N_FRAMES = 12   
+
+# ── Frame Buffer ──────────────────────────────
+# How many frames to keep in memory before picking the best one
+# More = better chance of catching a sharp frame, but more memory
+# OPTIMIZATION: Raised to 45 to give the selector a broader pool to find the sharpest snapshot
+FRAME_BUFFER_SIZE = 45
+
+# ── Motion / Trigger Settings ─────────────────
+# Minimum number of changed pixels to count as "vehicle present"
+MIN_MOTION_AREA     = 500   # Lower = more sensitive to small objects / distance
+
+# Max frame count a vehicle can go without motion before ending the cluster event
+MAX_STATIONARY_FRAMES = 15   # Lower = splits slow traffic into separate events
+
+# Quality/Selection rules:
+MIN_PLATE_CONFIDENCE = 0.40  # discard frames where plate detection is weak
+MAX_PLATE_ANGLE      = 15.0  # degrees max skew allowed (if computed)
+
+# Brightness constraints to filter out garbage flashes
+MIN_BRIGHTNESS      = 30    # widened for dark/night scenes in video
+MAX_BRIGHTNESS      = 220   # widened for bright scenes in video
+
+# ── Model Settings ────────────────────────────
+CONF_THRESHOLD = 0.25
+USE_HUGGING_FACE = True
+
+HF_TOKEN = None
+HF_PLATE_DETECTOR_REPO = "alyhassankamel/egyptian_license_plate_detector"
+HF_CHAR_OCR_REPO       = "alyhassankamel/egyptian-license-plate-ocr"
+HF_CAR_CLASSIFIER_REPO = "PatrickBLB/egyptian_cars_recognition_computer_vision_model"
+
+from pathlib import Path
+PROJECT_ROOT = Path(__file__).resolve().parent
+PLATE_DETECTOR_MODEL_PATH = str(PROJECT_ROOT / "weights" / "detector_best.pt")
+CHAR_OCR_MODEL_PATH       = str(PROJECT_ROOT / "weights" / "ocr_best.pt")
+
+# ── Output / Storage ──────────────────────────
+import os
+
+# Directory where captured frames are saved
+OUTPUT_DIR = "captured_frames"
+os.makedirs(OUTPUT_DIR, exist_ok=True)
+
+# Save every triggered frame to disk?  True = yes, False = send to API only
+SAVE_FRAMES_TO_DISK = True
+
+# Image format for saved frames
+SAVE_FORMAT = ".jpg"
+JPEG_QUALITY = 95   # 0–100
+
+# ── API Settings ──────────────────────────────
+# FastAPI /api/detect endpoint.
+# Set to None to skip sending and only save locally.
+API_ENDPOINT = "http://localhost:8000/api/detect"   # ← FastAPI server
+API_TIMEOUT  = 60     # seconds (LPR pipeline can be slow on CPU)
+
+# Camera UUID registered in the 'cameras' table (Supabase).
+# Every detection is tagged with this ID so the dashboard knows
+# which physical camera triggered the event.
+# Create a camera first:  POST /api/cameras  → copy the returned UUID here.
+CAMERA_ID = "70a649c0-4ad7-4a85-bc12-9e7144e4a777"
+
+# ── Display / Debug ───────────────────────────
+# Show live preview window while running?  Set False on headless servers
+SHOW_PREVIEW = True
+
+# Draw motion mask overlay on the preview?
+SHOW_MOTION_MASK = False
+
+# Print verbose console status?
+VERBOSE_LOGGING = True

src/capture/register_camera.py

@@ -0,0 +1,96 @@
+"""
+Register a camera via POST /api/cameras and optionally save CAMERA_ID to config.py.
+
+Usage (FastAPI must be running on port 8000):
+    python register_camera.py
+    python register_camera.py --name "Gate A" --type entrance --write-config
+"""
+
+import argparse
+import re
+import sys
+
+import requests
+
+import config
+
+DEFAULT_API = "http://localhost:8000"
+
+
+def create_camera(api_base: str, location_name: str, location_type: str) -> dict:
+    url = f"{api_base.rstrip('/')}/api/cameras"
+    resp = requests.post(
+        url,
+        json={"location_name": location_name, "location_type": location_type},
+        timeout=15,
+    )
+    if resp.status_code != 201:
+        print(f"Failed ({resp.status_code}): {resp.text}")
+        sys.exit(1)
+    return resp.json()
+
+
+def write_camera_id_to_config(camera_id: str) -> None:
+    config_path = config.__file__
+    with open(config_path, encoding="utf-8") as f:
+        text = f.read()
+
+    if re.search(r'^CAMERA_ID\s*=', text, flags=re.MULTILINE):
+        text = re.sub(
+            r'^CAMERA_ID\s*=.*$',
+            f'CAMERA_ID = "{camera_id}"',
+            text,
+            count=1,
+            flags=re.MULTILINE,
+        )
+    else:
+        text += f'\nCAMERA_ID = "{camera_id}"\n'
+
+    with open(config_path, "w", encoding="utf-8") as f:
+        f.write(text)
+
+
+def main():
+    parser = argparse.ArgumentParser(description="Register a camera for capture → API integration")
+    parser.add_argument("--api", default=DEFAULT_API, help="FastAPI base URL")
+    parser.add_argument("--name", default="Video test camera", help="location_name")
+    parser.add_argument("--type", default="entrance", help="location_type")
+    parser.add_argument(
+        "--write-config",
+        action="store_true",
+        help="Write returned UUID into config.py as CAMERA_ID",
+    )
+    args = parser.parse_args()
+
+    print(f"Creating camera at {args.api}/api/cameras ...")
+    try:
+        row = create_camera(args.api, args.name, args.type)
+    except requests.exceptions.ConnectionError:
+        print("Cannot reach API. Start the server first:")
+        print('  cd vehicle-intelligence-dashboard')
+        print("  python -m uvicorn app.main:app --reload --host 0.0.0.0 --port 8000")
+        sys.exit(1)
+
+    camera_id = row["id"]
+    print(f"\nCamera created:")
+    print(f"  id             : {camera_id}")
+    print(f"  location_name  : {row['location_name']}")
+    print(f"  location_type  : {row['location_type']}")
+
+    if args.write_config:
+        write_camera_id_to_config(camera_id)
+        print(f"\nWrote CAMERA_ID to {config.__file__}")
+    else:
+        print(f"\nAdd to config.py:  CAMERA_ID = \"{camera_id}\"")
+        print(f"Or run capture with:  --camera-id {camera_id}")
+
+    print("\nTest detect with a saved crop:")
+    print(
+        f'  curl -X POST "{args.api}/api/detect" '
+        f'-F "file=@captured_frames/plate_....jpg" '
+        f'-F "camera_id={camera_id}"'
+    )
+
+
+if __name__ == "__main__":
+    main()

src/capture/test_capture.py

@@ -0,0 +1,313 @@
+"""
+EALPR — Capture System Test Suite
+───────────────────────────────────
+Tests every component of capture.py independently
+so you can verify each piece works before connecting
+a real camera or the FastAPI backend.
+
+Run with:
+    python test_capture.py
+    python xtest_capture.py --image path/to/car.jpg
+    python test_capture.py --video path/to/clip.mp4
+"""
+
+import cv2
+import numpy as np
+import os
+import argparse
+import time
+from collections import deque
+from datetime import datetime
+
+import sys
+from pathlib import Path
+
+_ROOT = Path(__file__).resolve().parents[2]
+if str(_ROOT) not in sys.path:
+    sys.path.insert(0, str(_ROOT))
+
+import config
+from app.pipeline.video_capture import VideoCapturer
+
+_capturer = VideoCapturer()
+
+
+def sharpness_score(frame):
+    return _capturer.sharpness_score(frame)
+
+
+def brightness_ok(frame):
+    gray = __import__("cv2").cvtColor(frame, __import__("cv2").COLOR_BGR2GRAY)
+    return config.MIN_BRIGHTNESS <= gray.mean() <= config.MAX_BRIGHTNESS
+
+
+def frame_is_usable(frame):
+    if not brightness_ok(frame):
+        return False, "brightness out of range"
+    score = sharpness_score(frame)
+    if score < config.MIN_SHARPNESS_SCORE:
+        return False, f"too blurry (score={score:.1f})"
+    return True, ""
+
+
+def best_frame_from_buffer(buffer):
+    return _capturer.best_frame_from_buffer(buffer)
+
+
+def save_vehicle_crop(crop, vehicle_idx, timestamp):
+    import cv2
+    filename = f"plate_{timestamp}_v{vehicle_idx}{config.SAVE_FORMAT}"
+    filepath = os.path.join(config.OUTPUT_DIR, filename)
+    cv2.imwrite(filepath, crop, [cv2.IMWRITE_JPEG_QUALITY, config.JPEG_QUALITY])
+    return filepath
+
+
+# ─────────────────────────────────────────────
+#  Helpers
+# ─────────────────────────────────────────────
+
+def make_synthetic_frame(brightness: int = 120, blur_ksize: int = 1) -> np.ndarray:
+    """Creates a fake frame for testing quality functions."""
+    # White text on dark background (simulates a plate)
+    frame = np.full((480, 640, 3), brightness, dtype=np.uint8)
+    cv2.putText(frame, "ABC 1234", (80, 260),
+                cv2.FONT_HERSHEY_SIMPLEX, 4, (255, 255, 255), 8)
+    if blur_ksize > 1:
+        frame = cv2.GaussianBlur(frame, (blur_ksize, blur_ksize), 0)
+    return frame
+
+
+def pass_fail(condition: bool) -> str:
+    return "✅ PASS" if condition else "❌ FAIL"
+
+
+# ─────────────────────────────────────────────
+#  Unit tests
+# ─────────────────────────────────────────────
+
+def test_sharpness():
+    print("\n── Test 1: Sharpness Scoring ─────────────────")
+    sharp  = make_synthetic_frame(blur_ksize=1)
+    blurry = make_synthetic_frame(blur_ksize=51)
+
+    s_sharp  = sharpness_score(sharp)
+    s_blurry = sharpness_score(blurry)
+
+    print(f"  Sharp frame score  : {s_sharp:.1f}")
+    print(f"  Blurry frame score : {s_blurry:.1f}")
+    print(f"  Sharp > Blurry     : {pass_fail(s_sharp > s_blurry)}")
+    print(f"  Sharp passes gate  : {pass_fail(s_sharp >= config.MIN_SHARPNESS_SCORE)}")
+    print(f"  Blurry fails gate  : {pass_fail(s_blurry < config.MIN_SHARPNESS_SCORE)}")
+
+
+def test_brightness():
+    print("\n── Test 2: Brightness Check ──────────────────")
+    normal   = make_synthetic_frame(brightness=120)
+    too_dark = make_synthetic_frame(brightness=10)
+    blown_out = make_synthetic_frame(brightness=250)
+
+    print(f"  Normal frame ok    : {pass_fail(brightness_ok(normal))}")
+    print(f"  Dark frame rejected: {pass_fail(not brightness_ok(too_dark))}")
+    print(f"  Blown out rejected : {pass_fail(not brightness_ok(blown_out))}")
+
+
+def test_usability():
+    print("\n── Test 3: Combined Usability Gate ──────────")
+    good   = make_synthetic_frame(brightness=120, blur_ksize=1)
+    bad    = make_synthetic_frame(brightness=5,   blur_ksize=51)
+
+    ok_good, _   = frame_is_usable(good)
+    ok_bad,  why = frame_is_usable(bad)
+
+    print(f"  Good frame passes  : {pass_fail(ok_good)}")
+    print(f"  Bad frame rejected : {pass_fail(not ok_bad)} (reason: {why})")
+
+
+def test_best_frame_selection():
+    print("\n── Test 4: Best Frame from Buffer ───────────")
+    buffer = deque(maxlen=20)
+
+    # Fill with increasingly sharp frames
+    for blur in [51, 31, 21, 11, 1]:
+        buffer.append(make_synthetic_frame(blur_ksize=blur))
+
+    best = best_frame_from_buffer(buffer)
+    if best is None:
+        print("  ❌ FAIL — no frame selected")
+        return
+
+    score = sharpness_score(best)
+    print(f"  Best frame score   : {score:.1f}")
+    print(f"  Correctly sharpest : {pass_fail(score >= config.MIN_SHARPNESS_SCORE)}")
+
+
+def test_save_frame():
+    print("\n── Test 5: Save Frame to Disk ───────────────")
+    frame    = make_synthetic_frame()
+    ts       = datetime.now().strftime("%Y%m%d_%H%M%S_%f")
+    filepath = save_vehicle_crop(frame, vehicle_idx=1, timestamp=ts)
+
+    exists = os.path.isfile(filepath)
+    size   = os.path.getsize(filepath) if exists else 0
+
+    print(f"  File created       : {pass_fail(exists)}")
+    print(f"  File not empty     : {pass_fail(size > 0)}")
+    print(f"  Saved to           : {filepath}")
+
+
+def test_motion_detection():
+    print("\n── Test 6: Motion Detection (MOG2) ──────────")
+    fgbg = cv2.createBackgroundSubtractorMOG2(
+        history=config.MOG2_HISTORY,
+        varThreshold=config.MOG2_VAR_THRESHOLD,
+        detectShadows=True,
+    )
+
+    background = make_synthetic_frame(brightness=80, blur_ksize=1)
+    # "Prime" the background subtractor
+    for _ in range(30):
+        fgbg.apply(background)
+
+    # Now introduce a "car" (bright rectangle = large foreground change)
+    car_frame = background.copy()
+    cv2.rectangle(car_frame, (100, 150), (540, 350), (200, 200, 200), -1)
+
+    mask    = fgbg.apply(car_frame)
+    kernel  = cv2.getStructuringElement(cv2.MORPH_ELLIPSE, (5, 5))
+    cleaned = cv2.morphologyEx(mask, cv2.MORPH_OPEN, kernel)
+    pixels  = cv2.countNonZero(cleaned)
+
+    print(f"  Foreground pixels  : {pixels}")
+    print(f"  Trigger fires      : {pass_fail(pixels > config.MOTION_THRESHOLD)}")
+
+
+# ─────────────────────────────────────────────
+#  Integration test on a real image
+# ─────────────────────────────────────────────
+
+def test_with_image(image_path: str):
+    print(f"\n── Integration Test: Real Image ─────────────")
+    print(f"  Path: {image_path}")
+
+    frame = cv2.imread(image_path)
+    if frame is None:
+        print(f"  ❌ Cannot load image: {image_path}")
+        return
+
+    h, w = frame.shape[:2]
+    print(f"  Dimensions : {w}×{h}")
+
+    score = sharpness_score(frame)
+    ok, reason = frame_is_usable(frame)
+
+    print(f"  Sharpness  : {score:.1f}")
+    print(f"  Usable     : {pass_fail(ok)}", f"({reason})" if not ok else "")
+
+    if ok:
+        ts       = datetime.now().strftime("%Y%m%d_%H%M%S_%f")
+        filepath = save_vehicle_crop(frame, vehicle_idx=1, timestamp=ts)
+        print(f"  Saved to   : {filepath}")
+
+        # Show the frame
+        cv2.imshow("Test Image", frame)
+        print("  (Press any key to close preview)")
+        cv2.waitKey(0)
+        cv2.destroyAllWindows()
+
+
+# ─────────────────────────────────────────────
+#  Integration test on a real video clip
+# ─────────────────────────────────────────────
+
+def test_with_video(video_path: str):
+    print(f"\n── Integration Test: Real Video Clip ────────")
+    print(f"  Path: {video_path}")
+
+    cap = cv2.VideoCapture(video_path)
+    if not cap.isOpened():
+        print(f"  ❌ Cannot open video: {video_path}")
+        return
+
+    total_frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))
+    fps          = cap.get(cv2.CAP_PROP_FPS)
+    print(f"  Frames: {total_frames}  FPS: {fps:.1f}")
+
+    fgbg   = cv2.createBackgroundSubtractorMOG2(
+        history=config.MOG2_HISTORY,
+        varThreshold=config.MOG2_VAR_THRESHOLD,
+        detectShadows=True,
+    )
+    kernel  = cv2.getStructuringElement(cv2.MORPH_ELLIPSE, (5, 5))
+    buffer  = deque(maxlen=config.FRAME_BUFFER_SIZE)
+    triggers = 0
+    last_trigger = 0
+
+    frame_idx = 0
+    while True:
+        ret, frame = cap.read()
+        if not ret:
+            break
+
+        frame_idx += 1
+        buffer.append(frame.copy())
+
+        if frame_idx % config.PROCESS_EVERY_N_FRAMES != 0:
+            continue
+
+        fgmask  = fgbg.apply(frame)
+        cleaned = cv2.morphologyEx(fgmask, cv2.MORPH_OPEN, kernel)
+        pixels  = cv2.countNonZero(cleaned)
+
+        now = time.time()
+        if pixels > config.MOTION_THRESHOLD and (now - last_trigger) > config.COOLDOWN_SECONDS:
+            triggers  += 1
+            last_trigger = now
+            best = best_frame_from_buffer(buffer)
+            if best is not None:
+                score    = sharpness_score(best)
+                ts       = datetime.now().strftime("%Y%m%d_%H%M%S_%f")
+                filepath = save_vehicle_crop(best, vehicle_idx=1, timestamp=ts)
+                print(f"  🚗 Trigger #{triggers} at frame {frame_idx}  "
+                      f"sharpness={score:.1f}  → {os.path.basename(filepath)}")
+
+    cap.release()
+    print(f"\n  Total triggers : {triggers}")
+    print(f"  Frames saved   : {triggers}")
+    print(f"  {pass_fail(triggers > 0)}")
+
+
+# ─────────────────────────────────────────────
+#  Entry point
+# ─────────────────────────────────────────────
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser(description="EALPR Capture System Tests")
+    parser.add_argument("--image", default=None, help="Path to a test car image")
+    parser.add_argument("--video", default=None, help="Path to a test video clip")
+    args = parser.parse_args()
+
+    print("═" * 52)
+    print("  EALPR Capture System — Test Suite")
+    print("═" * 52)
+
+    # Always run unit tests
+    test_sharpness()
+    test_brightness()
+    test_usability()
+    test_best_frame_selection()
+    test_save_frame()
+    test_motion_detection()
+
+    # Optional integration tests
+    if args.image:
+        test_with_image(args.image)
+
+    if args.video:
+        test_with_video(args.video)
+
+    print("\n" + "═" * 52)
+    print("  All unit tests complete.")
+    if not args.image and not args.video:
+        print("  Tip: run with --image car.jpg or --video clip.mp4")
+        print("       for integration tests on real footage.")
+    print("═" * 52)

src/capture/test_detect_api.py

@@ -0,0 +1,81 @@
+"""
+POST a video (or image) to /api/detect.
+
+Usage (FastAPI running):
+    python test_detect_api.py
+    python test_detect_api.py --video "d:\\path\\to\\file.mp4"
+"""
+
+import argparse
+import glob
+import os
+import sys
+
+import requests
+
+import config
+
+
+def main():
+    parser = argparse.ArgumentParser(description="Test POST /api/detect with video or image")
+    parser.add_argument("--video", default=None, help="Path to MP4/video file")
+    parser.add_argument("--image", default=None, help="Path to a single crop image (legacy)")
+    parser.add_argument("--camera-id", default=None)
+    parser.add_argument(
+        "--endpoint",
+        default=getattr(config, "API_ENDPOINT", "http://localhost:8000/api/detect"),
+    )
+    parser.add_argument("--max-triggers", type=int, default=5, help="Limit vehicles for test")
+    args = parser.parse_args()
+
+    camera_id = args.camera_id or getattr(config, "CAMERA_ID", "")
+    if not camera_id:
+        print("CAMERA_ID empty. Run: python register_camera.py --write-config")
+        sys.exit(1)
+
+    path = args.video
+    content_type = "video/mp4"
+    if not path:
+        if args.image:
+            path = args.image
+            content_type = "image/jpeg"
+        else:
+            path = getattr(config, "CAMERA_URL", None)
+            if path and os.path.isfile(str(path)) and str(path).lower().endswith(
+                (".mp4", ".avi", ".mov", ".mkv")
+            ):
+                content_type = "video/mp4"
+            else:
+                candidates = sorted(glob.glob(os.path.join(config.OUTPUT_DIR, "plate_*.jpg")))
+                if candidates:
+                    path = candidates[-1]
+                    content_type = "image/jpeg"
+                else:
+                    print("Provide --video or set CAMERA_URL to an mp4 in config.py")
+                    sys.exit(1)
+
+    if not os.path.isfile(path):
+        print(f"File not found: {path}")
+        sys.exit(1)
+
+    print(f"POST {args.endpoint}")
+    print(f"  file       : {path}")
+    print(f"  type       : {content_type}")
+    print(f"  camera_id  : {camera_id}")
+    print(f"  max_triggers: {args.max_triggers}")
+
+    with open(path, "rb") as f:
+        resp = requests.post(
+            args.endpoint,
+            files={"file": (os.path.basename(path), f, content_type)},
+            data={"camera_id": camera_id, "max_triggers": str(args.max_triggers)},
+            timeout=getattr(config, "API_TIMEOUT", 600),
+        )
+
+    print(f"\nStatus: {resp.status_code}")
+    print(resp.text[:5000])
+    sys.exit(0 if resp.status_code == 200 else 1)
+
+
+if __name__ == "__main__":
+    main()

src/capture/yolov8n.pt

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:f59b3d833e2ff32e194b5bb8e08d211dc7c5bdf144b90d2c8412c47ccfc83b36
+size 6549796

src/car_recognition/car_scraper.py

@@ -0,0 +1,216 @@
+import pandas as pd
+import requests
+import re
+import time
+import random
+import argparse
+from pathlib import Path
+from urllib.parse import unquote, quote
+from concurrent.futures import ThreadPoolExecutor, as_completed
+from scrapling.fetchers import StealthyFetcher
+from exterior_filterer import ExteriorFilterer
+
+USER_AGENT = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/131.0.0.0 Safari/537.36"
+MAX_IMAGES = 100
+BETWEEN_DELAY_MIN = 3
+BETWEEN_DELAY_MAX = 6
+DOWNLOAD_TIMEOUT = 10
+
+
+class CarImageScraper:
+    def __init__(self, output_dir=None):
+        self.output_dir = output_dir or Path(__file__).parent.parent.parent / "datasets/car_images_scrapler/"
+        self.session = requests.Session()
+        self.session.headers.update({'User-Agent': USER_AGENT})
+        self.exterior_filterer = ExteriorFilterer()
+
+    def download_car_images(self, csv_path, start=0, end=None):
+        self.output_dir.mkdir(parents=True, exist_ok=True)
+        df = pd.read_csv(csv_path)
+
+        titles = df['title'].tolist()[start:end]
+
+        for index, title in enumerate(titles):
+            print(f"[{index + 1}/{len(titles)}] Processing: {title}")
+            try:
+                car_folder = self._make_car_folder(title)
+                existing = list(car_folder.glob("*.jpg"))
+                if len(existing) >= MAX_IMAGES:
+                    print(f"  Already have {len(existing)} images, skipping.")
+                    continue
+
+                img_urls = self._search_google_images(title)
+                if img_urls:
+                    count = self._download_images(img_urls, car_folder)
+                    print(f"  Downloaded {count} images for: {title}")
+                    print(f"  Running exterior filter on: {car_folder}")
+                    self.exterior_filterer.filter_images(str(car_folder))
+                    self._reindex_images(car_folder)
+                else:
+                    print(f"  No images found for: {title}")
+            except Exception as e:
+                print(f"  Error processing {title}: {e}")
+
+            if index < len(titles) - 1:
+                delay = random.uniform(BETWEEN_DELAY_MIN, BETWEEN_DELAY_MAX)
+                print(f"  Waiting {delay:.1f}s before next search...")
+                time.sleep(delay)
+
+    def _search_google_images(self, query):
+        search_query = f"{query} car exterior"
+
+        # Step 1: Regular Google search to find the Images tab link
+        search_url = f"https://www.google.com/search?q={quote(search_query)}&hl=en"
+        print(f"    Step 1: Fetching regular search page...")
+        try:
+            page = StealthyFetcher.fetch(search_url, headless=True)
+        except Exception as e:
+            print(f"    StealthyFetcher failed on search page: {e}")
+            return []
+
+        # Find the Images tab link from the navigation
+        images_url = None
+        links = page.css('a')
+        for link in links:
+            href = link.attrib.get('href', '')
+            text = link.text or ''
+            if 'Images' in text or 'images' in text:
+                if href.startswith('/'):
+                    images_url = f"https://www.google.com{href}"
+                elif href.startswith('http'):
+                    images_url = href
+                print(f"    Found Images tab link: {images_url[:80]}...")
+                break
+
+        if not images_url:
+            # Fallback: use udm=2 which is Google's images mode
+            images_url = f"https://www.google.com/search?q={quote(search_query)}&udm=2&hl=en"
+            print(f"    Images tab not found in nav, using udm=2 fallback")
+
+        # Step 2: Fetch the Images results page
+        print(f"    Step 2: Fetching images page...")
+        time.sleep(random.uniform(1, 2))
+        try:
+            images_page = StealthyFetcher.fetch(images_url, headless=True)
+            html = images_page.html_content if hasattr(images_page, 'html_content') else str(images_page)
+        except Exception as e:
+            print(f"    StealthyFetcher failed on images page: {e}")
+            return []
+
+        all_urls = self._extract_image_urls_from_html(html)
+
+        # Deduplicate while preserving order
+        seen = set()
+        unique_urls = []
+        for u in all_urls:
+            if u not in seen and len(u) > 20:
+                seen.add(u)
+                unique_urls.append(u)
+
+        print(f"    Found {len(unique_urls)} unique image URLs from Google")
+        return unique_urls
+
+    def _extract_image_urls_from_html(self, html):
+        all_urls = []
+
+        # Pattern 1: Full-res image URLs in script tags ["url", width, height]
+        matches = re.findall(
+            r'\["(https?://[^"]+\.(?:jpg|jpeg|png|webp)(?:\?[^"]*)?)",\s*\d+,\s*\d+\]',
+            html
+        )
+        for m in matches:
+            cleaned = m.replace('\\u003d', '=').replace('\\u0026', '&')
+            if not self._is_google_domain(cleaned):
+                all_urls.append(cleaned)
+
+        # Pattern 2: data-ou attributes (direct image URLs)
+        ou_matches = re.findall(r'data-ou="(https?://[^"]+)"', html)
+        for m in ou_matches:
+            all_urls.append(unquote(m))
+
+        # Pattern 3: imgurl= in href links
+        imgurl_matches = re.findall(r'imgurl=(https?://[^&"]+)', html)
+        for m in imgurl_matches:
+            all_urls.append(unquote(m))
+
+        # Pattern 4: Broad image URL extraction from embedded script data
+        script_matches = re.findall(
+            r'(https?://[^\s"\\,\]\)]+\.(?:jpg|jpeg|png|webp))',
+            html
+        )
+        for m in script_matches:
+            cleaned = m.replace('\\u003d', '=').replace('\\u0026', '&')
+            if not self._is_google_domain(cleaned):
+                all_urls.append(cleaned)
+
+        return all_urls
+
+    def _is_google_domain(self, url):
+        blocked = ['gstatic.com', 'google.com', 'googleapis.com', 'googleusercontent.com', 'ggpht.com']
+        return any(domain in url for domain in blocked)
+
+    def _make_car_folder(self, title):
+        safe_title = "".join(c for c in title if c.isalnum() or c in (' ', '_', '-')).rstrip()
+        car_folder = self.output_dir / safe_title.replace(' ', '_')
+        car_folder.mkdir(parents=True, exist_ok=True)
+        return car_folder
+
+    def _download_single_image(self, img_url):
+        """Download a single image, return bytes or None."""
+        try:
+            resp = self.session.get(img_url, timeout=DOWNLOAD_TIMEOUT)
+            if resp.status_code != 200:
+                return None
+            content_type = resp.headers.get('Content-Type', '')
+            if not content_type.startswith('image/'):
+                return None
+            if len(resp.content) < 5000:
+                return None
+            return resp.content
+        except Exception as e:
+            print(f"    Failed to download {img_url}: {e}")
+            return None
+
+    def _download_images(self, img_urls, car_folder, max_workers=20):
+        # Filter out logo/icon URLs first
+        filtered_urls = [u for u in img_urls if 'logo' not in u.lower() and 'icon' not in u.lower()]
+        # Only submit up to MAX_IMAGES + buffer to account for failures
+        candidate_urls = filtered_urls[:MAX_IMAGES + 50]
+
+        count = 0
+        with ThreadPoolExecutor(max_workers=max_workers) as executor:
+            future_to_url = {
+                executor.submit(self._download_single_image, url): url
+                for url in candidate_urls
+            }
+            for future in as_completed(future_to_url):
+                if count >= MAX_IMAGES:
+                    break
+                img_data = future.result()
+                if img_data:
+                    with open(car_folder / f"{car_folder.name}_{count}.jpg", 'wb') as f:
+                        f.write(img_data)
+                    count += 1
+        return count
+
+    def _reindex_images(self, car_folder):
+        images = sorted(car_folder.glob("*.jpg"), key=lambda f: f.stat().st_mtime)
+        for new_index, img_path in enumerate(images):
+            new_name = car_folder / f"{car_folder.name}_{new_index}.jpg"
+            if img_path != new_name:
+                img_path.rename(new_name)
+        print(f"  Reindexed {len(images)} remaining images in: {car_folder.name}")
+
+    def _extract_car_id(self, title):
+        return title.replace(' ', '_')
+
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser(description="Scrape car images from Google via Scrapling")
+    parser.add_argument("--start", type=int, default=0, help="Start index in CSV (inclusive)")
+    parser.add_argument("--end", type=int, default=None, help="End index in CSV (exclusive)")
+    args = parser.parse_args()
+
+    csv_path = Path(__file__).parent.parent.parent / "datasets/hatla2ee_cars_unique.csv"
+    scraper = CarImageScraper()
+    scraper.download_car_images(csv_path, start=args.start, end=args.end)

src/car_recognition/exterior_filterer.py

@@ -0,0 +1,90 @@
+import argparse
+import os
+from pathlib import Path
+from typing import List
+
+import torch
+from PIL import Image
+from transformers import CLIPProcessor, CLIPModel
+
+class ExteriorFilterer:
+    def __init__(self, model_id="openai/clip-vit-base-patch32"):
+        self.device = "mps" if torch.backends.mps.is_available() else "cpu"
+        print(f"Loading CLIP model {model_id} on {self.device}...")
+        self.model = CLIPModel.from_pretrained(model_id).to(self.device)
+        self.processor = CLIPProcessor.from_pretrained(model_id)
+        self.labels = [
+            "a photo of the outside of a car showing the full body, paint, wheels, or bumper",
+            "a photo of the inside of a car showing the dashboard, steering wheel, seats, or center console",
+            "a photo of a car engine or motor under the hood",
+            "a photo of the trunk or boot of a car, open or closed, showing cargo area",
+        ]
+        self.extensions = {".jpg", ".jpeg", ".png", ".webp"}
+
+    def filter_images(self, input_folder: str, dry_run: bool = False):
+        input_path = Path(input_folder)
+        if not input_path.is_dir():
+            print(f"Error: {input_folder} is not a valid directory.")
+            return
+
+        all_images = [f for f in input_path.rglob("*") if f.is_file() and f.suffix.lower() in self.extensions]
+        total_images = len(all_images)
+
+        if total_images == 0:
+            print("No valid images found in the specified directory.")
+            return
+
+        kept_count = 0
+        deleted_count = 0
+
+        print(f"Processing {total_images} images...")
+
+        with torch.no_grad():
+            for i, img_path in enumerate(all_images, 1):
+                try:
+                    image = Image.open(img_path).convert("RGB")
+
+                    inputs = self.processor(
+                        text=self.labels,
+                        images=image,
+                        return_tensors="pt",
+                        padding=True
+                    ).to(self.device)
+
+                    outputs = self.model(**inputs)
+                    probs = outputs.logits_per_image.softmax(dim=1)
+
+                    # Keep only if exterior (index 0) has the highest probability
+                    best_label = probs[0].argmax().item()
+                    if best_label != 0:
+                        if dry_run:
+                            print(f"[{i}/{total_images}] would delete: {img_path.name}")
+                        else:
+                            img_path.unlink()
+                            print(f"[{i}/{total_images}] deleted: {img_path.name}")
+                        deleted_count += 1
+                    else:
+                        kept_count += 1
+
+                except Exception as e:
+                    print(f"[{i}/{total_images}] Warning: Could not process {img_path.name} due to error: {e}")
+                    continue
+
+        print("\n--- Summary ---")
+        print(f"Total processed: {total_images}")
+        print(f"Kept: {kept_count}")
+        print(f"Deleted: {deleted_count}")
+        if dry_run:
+            print("Note: This was a dry run, no files were actually deleted.")
+
+def main():
+    parser = argparse.ArgumentParser(description="Filter car images by removing interiors using CLIP zero-shot classification.")
+    parser.add_argument("--input", type=str, default="datasets/car_images_v3", help="Path to the folder containing car images")
+    parser.add_argument("--dry-run", action="store_true", help="Print what would be deleted without actually deleting")
+    args = parser.parse_args()
+
+    filterer = ExteriorFilterer()
+    filterer.filter_images(args.input, dry_run=args.dry_run)
+
+if __name__ == "__main__":
+    main()

thisishowtorun.md

@@ -0,0 +1,11 @@
+# Terminal 1 — FastAPI backend (from project root)
+npm
+
+# Terminal 2 — Frontend dashboard
+npm run dev
+
+# Terminal 3 — Upload video to API (from src/capture/)
+# python register_camera.py --write-config   # once, while API is up
+# python capture.py                        # POST video → /api/detect (pipeline server-side)
+# python test_detect_api.py --max-triggers 5
+# python capture.py --local                # run VideoCapturer + LPR without HTTP

yolo11n.pt

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:0ebbc80d4a7680d14987a577cd21342b65ecfd94632bd9a8da63ae6417644ee1
+size 5613764