Blame for app.py

import os
import spaces
import gc
import cv2
import numpy as np
import pandas as pd
import gradio as gr
from pathlib import Path
from ultralytics import YOLO

# ============================================================
# COCO Labels (only what matters for driving)
# ============================================================
RELEVANT_CLASSES = {
    "person", "bicycle", "car", "motorcycle", "bus", "truck",
    "traffic light", "stop sign",
}

# ============================================================
# Model Loading (Cached)
# ============================================================
model_cache = {}

def get_model(model_name: str):
    """Load YOLOv8 model (cached to avoid reloading)."""
    if model_name not in model_cache:
        model_cache[model_name] = YOLO(model_name)
    return model_cache[model_name]

# ============================================================
# Detection (YOLOv8)
# ============================================================
def detect_road_objects(frame, model, score_threshold, imgsz=512):
    """Run YOLOv8 on a frame and return filtered detections."""
    results = model.predict(
        source=frame,
        conf=score_threshold,
        imgsz=imgsz,
        verbose=False,
        device="cpu",
    )
    detections = []
    if not results:
        return detections

    r = results[0]
    if r.boxes is None:
        return detections

    boxes = r.boxes.xyxy.cpu().numpy()
    confs = r.boxes.conf.cpu().numpy()
    cls_ids = r.boxes.cls.cpu().numpy().astype(int)
    names = r.names

    for box, conf, cid in zip(boxes, confs, cls_ids):
        label = names.get(cid, str(cid))
        if label not in RELEVANT_CLASSES:
            continue
        x1, y1, x2, y2 = box.astype(int).tolist()
        detections.append({
            "box": [x1, y1, x2, y2],
            "label": label,
            "confidence": float(conf),
        })
    return detections

# ============================================================
# IoU + Tracker
# ============================================================
def iou(a, b):
    ax1, ay1, ax2, ay2 = a
    bx1, by1, bx2, by2 = b
    inter = max(0, min(ax2, bx2) - max(ax1, bx1)) * max(0, min(ay2, by2) - max(ay1, by1))
    union = (ax2 - ax1) * (ay2 - ay1) + (bx2 - bx1) * (by2 - by1) - inter
    return inter / union if union > 0 else 0.0


class IoUTracker:
    def __init__(self, threshold=0.30, max_missing=12):
        self.threshold = threshold
        self.max_missing = max_missing
        self.tracks = {}
        self.next_id = 1

    def update(self, detections):
        used = set()
        for track_id, track in list(self.tracks.items()):
            choices = [
                (iou(track["box"], d["box"]), i)
                for i, d in enumerate(detections)
                if i not in used and d["label"] == track["label"]
            ]
            best, idx = max(choices, default=(0, -1))
            if best >= self.threshold:
                self.tracks[track_id].update(box=detections[idx]["box"], missing=0)
                detections[idx]["track_id"] = track_id
                used.add(idx)
            else:
                self.tracks[track_id]["missing"] += 1

        self.tracks = {k: v for k, v in self.tracks.items() if v["missing"] <= self.max_missing}

        for i, d in enumerate(detections):
            if i not in used:
                d["track_id"] = self.next_id
                self.tracks[self.next_id] = {"box": d["box"], "label": d["label"], "missing": 0}
                self.next_id += 1
        return detections

# ============================================================
# Risk Assessment
# ============================================================
def risk_for_object(box, label, frame_w, frame_h):
    x1, y1, x2, y2 = box
    area_ratio = ((x2 - x1) * (y2 - y1)) / (frame_w * frame_h)
    center_x = (x1 + x2) / 2
    in_lane = 0.30 * frame_w < center_x < 0.70 * frame_w
    important = label in {"person", "bicycle", "car", "motorcycle", "bus", "truck"}

    if important and in_lane and area_ratio > 0.075:
        return "DANGER", area_ratio
    if important and (in_lane or area_ratio > 0.025):
        return "CAUTION", area_ratio
    return "SAFE", area_ratio

# ============================================================
# Video Processing Pipeline
# ============================================================
@spaces.GPU
def process_driving_video(
    input_video_path,
    model_choice,
    score_threshold,
    iou_threshold,
    max_missing,
    skip_frames,
    imgsz
):
    if not input_video_path:
        return None, None, "⚠️ الرجاء رفع فيديو أولاً."

    model = get_model(model_choice)
    cap = cv2.VideoCapture(input_video_path)
    if not cap.isOpened():
        return None, None, "❌ تعذر فتح الفيديو المرفق."

    fps = cap.get(cv2.CAP_PROP_FPS) or 25
    w = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
    h = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
    total_frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))

    output_path = "driving_assistant_output.mp4"
    csv_path = "driving_events.csv"

    import imageio.v2 as imageio
    writer = imageio.get_writer(
        output_path,
        fps=fps,
        codec="libx264",
        quality=7,
        pixelformat="yuv420p",
    )

    tracker = IoUTracker(threshold=iou_threshold, max_missing=max_missing)
    events = []
    frame_no = 0
    analyzed = 0

    rank = {"SAFE": 0, "CAUTION": 1, "DANGER": 2}
    colors = {"SAFE": (0, 200, 0), "CAUTION": (0, 190, 255), "DANGER": (0, 0, 255)}
    last_annotated = None

    while True:
        ok, frame = cap.read()
        if not ok:
            break

        frame_no += 1

        if frame_no % skip_frames != 0 and last_annotated is not None:
            writer.append_data(cv2.cvtColor(last_annotated, cv2.COLOR_BGR2RGB))
            continue

        analyzed += 1
        frame_risk = "SAFE"
        detections = detect_road_objects(frame, model, score_threshold, imgsz)
        tracked = tracker.update(detections)

        for d in tracked:
            risk, area = risk_for_object(d["box"], d["label"], w, h)
            if rank[risk] > rank[frame_risk]:
                frame_risk = risk

            x1, y1, x2, y2 = d["box"]
            color = colors[risk]
            cv2.rectangle(frame, (x1, y1), (x2, y2), color, 2)
            cv2.putText(
                frame,
                f"#{d['track_id']} {d['label']} {d['confidence']:.2f} {risk}",
                (x1, max(25, y1 - 8)),
                cv2.FONT_HERSHEY_SIMPLEX, 0.5, color, 2,
            )
            if risk != "SAFE":
                events.append({
                    "frame": frame_no,
                    "time_seconds": round(frame_no / fps, 2),
                    "track_id": d["track_id"],
                    "object": d["label"],
                    "risk": risk,
                    "box_area_ratio": round(area, 4),
                })

        cv2.rectangle(frame, (0, 0), (w, 60), colors[frame_risk], -1)
        cv2.putText(
            frame, f"RISK: {frame_risk}", (20, 40),
            cv2.FONT_HERSHEY_SIMPLEX, 1, (255, 255, 255), 2,
        )

        last_annotated = frame.copy()
        writer.append_data(cv2.cvtColor(frame, cv2.COLOR_BGR2RGB))

    cap.release()
    writer.close()

    events_df = pd.DataFrame(events).drop_duplicates() if events else pd.DataFrame(columns=["frame", "time_seconds", "track_id", "object", "risk", "box_area_ratio"])
    events_df.to_csv(csv_path, index=False)

    summary_msg = f"✅ تم الانتهاء بنجاح! تم تحليل {analyzed} إطار من أصل {total_frames}."
    return output_path, csv_path, summary_msg

# ============================================================
# Gradio UI Design
# ============================================================
with gr.Blocks() as demo:
    gr.Markdown("# 🚗 Smart Front-View Driving Assistant")
    gr.Markdown(" نظام مساعد القيادة الذكية باستخدام **YOLOv8** وتتبع الأجسام وتقييم المخاطر المرورية.")

    with gr.Row():
        with gr.Column(scale=1):
            video_input = gr.Video(label="📹 ارفع فيديو كاميرا القيادة (Dashcam)")
            
            model_choice = gr.Dropdown(
                choices=["yolov8n.pt", "yolov8s.pt", "yolov8m.pt"],
                value="yolov8s.pt",
                label="اختر نموذج YOLOv8"
            )
            
            with gr.Accordion("⚙️ إعدادات متقدمة", open=False):
                skip_frames = gr.Slider(1, 5, value=1, step=1, label="تخطي الإطارات (لزيادة السرعة)")
                imgsz = gr.Slider(320, 640, value=512, step=96, label="حجم صورة المعالجة (Inference Size)")
                score_threshold = gr.Slider(0.1, 0.9, value=0.35, step=0.05, label="عتبة الثقة (Confidence Threshold)")
                iou_threshold = gr.Slider(0.1, 0.8, value=0.30, step=0.05, label="عتبة التتبع (IoU Threshold)")
                max_missing = gr.Slider(1, 30, value=12, step=1, label="أقصى إطارات مفقودة للتتبع")

            submit_btn = gr.Button("🚀 بدء التحليل", variant="primary")

        with gr.Column(scale=1):
            video_output = gr.Video(label="🎬 فيديو النتائج النهائي")
            csv_output = gr.File(label="📥 تحميل تقرير الأحداث (CSV)")
            status_output = gr.Textbox(label="حالة العملية التنفيذية")

    submit_btn.click(
        fn=process_driving_video,
        inputs=[
            video_input, model_choice, score_threshold,
            iou_threshold, max_missing, skip_frames, imgsz
        ],
        outputs=[video_output, csv_output, status_output]
    )

if __name__ == "__main__":
    demo.launch(theme=gr.themes.Soft())