Upload app.py with huggingface_hub

app.py

@@ -0,0 +1,269 @@
+# ============================================================================
+# app.py - الواجهة التفاعلية لمشروع Interfuser
+# ============================================================================
+# هذا الملف يقوم بإنشاء واجهة ويب باستخدام Gradio لعرض أداء نموذج
+# Interfuser للقيادة الذاتية. يسمح للمستخدمين باختيار نماذج مختلفة
+# وسيناريوهات قيادة ومقارنة النتائج.
+# ============================================================================
+
+# --- الجزء الأول: الاستيرادات ---
+import os
+import torch
+import numpy as np
+import cv2
+import json
+import logging
+import traceback
+from PIL import Image
+from typing import Dict
+
+# مكتبة الواجهة الرسومية
+import gradio as gr
+
+# مكتبات معالجة الصور من Torch
+from torchvision import transforms
+
+# استيراد الوحدات الخاصة بنا من الملفات الأخرى
+try:
+    from model_definition import load_and_prepare_model, create_model_config
+    from simulation_modules import (
+        DisplayInterface,
+        InterfuserController,
+        ControllerConfig,
+        render_waypoints,
+        render_self_car,
+        render
+    )
+except ImportError as e:
+    print(f"خطأ في الاستيراد: تأكد من وجود ملفات model_definition.py و simulation_modules.py. الخطأ: {e}")
+    exit()
+
+# --- الجزء الثاني: الإعدادات والثوابت ---
+logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
+
+DEVICE = torch.device("cuda" if torch.cuda.is_available() else "cpu")
+MODEL_DIR = "model"
+SAMPLE_DATA_DIR = "sample_data"
+
+# --- الجزء الثالث: تعريف التحويلات (Transforms) ---
+# هذه التحويلات يجب أن تطابق تمامًا ما تم استخدامه أثناء تدريب النموذج
+RGB_TRANSFORM = transforms.Compose([
+    transforms.Resize((224, 224)),
+    transforms.ToTensor(),
+    transforms.Normalize(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225])
+])
+
+LIDAR_TRANSFORM = transforms.Compose([
+    transforms.Resize((224, 224)),
+    transforms.ToTensor()
+])
+
+# --- الجزء الرابع: الكائنات العامة ومخبأ النماذج ---
+logging.info("تهيئة المكونات العامة...")
+# قاموس لتخزين النماذج التي تم تحميلها لتجنب إعادة التحميل البطيئة
+GLOBAL_MODELS_CACHE: Dict[str, torch.nn.Module] = {}
+
+# تهيئة واجهة العرض والمتحكم (يتم تهيئتها مرة واحدة)
+GLOBAL_DISPLAY_INTERFACE = DisplayInterface()
+controller_config = ControllerConfig()
+GLOBAL_CONTROLLER = InterfuserController(controller_config)
+logging.info(f"تم تهيئة المكونات. يعمل التطبيق على الجهاز: {DEVICE}")
+
+# --- الجزء الخامس: دوال المساعدة والمعالجة ---
+
+def get_model(model_name: str) -> torch.nn.Module:
+    """
+    تقوم بتحميل النموذج إذا لم يكن في المخبأ، ثم تعيده.
+    """
+    if model_name in GLOBAL_MODELS_CACHE:
+        logging.info(f"إعادة استخدام النموذج '{model_name}' من المخبأ.")
+        return GLOBAL_MODELS_CACHE[model_name]
+
+    logging.info(f"النموذج '{model_name}' غير موجود في المخبأ. بدء التحميل...")
+    gr.Info(f"⏳ جاري تحميل نموذج جديد: {model_name}... قد يستغرق هذا بضع ثوانٍ.")
+
+    model_path = os.path.join(MODEL_DIR, model_name)
+    if not os.path.exists(model_path):
+        raise FileNotFoundError(f"ملف النموذج '{model_path}' غير موجود!")
+    
+    model_config = create_model_config(model_path=model_path)
+    model = load_and_prepare_model(model_config, DEVICE)
+    
+    if model:
+        GLOBAL_MODELS_CACHE[model_name] = model
+        logging.info(f"✅ تم تحميل وتخزين النموذج '{model_name}' في المخبأ.")
+        return model
+    else:
+        raise gr.Error(f"فشل تحميل النموذج '{model_name}'. يرجى مراجعة السجلات.")
+
+
+def process_and_run_model(
+    model: torch.nn.Module,
+    rgb_image_path: str, rgb_left_image_path: str, rgb_right_image_path: str,
+    rgb_center_image_path: str, lidar_image_path: str,
+    measurements_path: str, target_point_list: list
+):
+    """
+    محرك المعالجة: يعالج إطارًا واحدًا من البيانات بناءً على المسارات
+    ويقوم بتشغيل النموذج عليه.
+    """
+    try:
+        # 1. قراءة ومعالجة المدخلات
+        if not rgb_image_path or not os.path.exists(rgb_image_path): 
+            raise FileNotFoundError("ملف الصورة الأمامية (RGB) غير موجود.")
+
+        rgb_image = Image.open(rgb_image_path).convert("RGB")
+        
+        def open_optional_image(path, default_img):
+            return Image.open(path).convert("RGB") if path and os.path.exists(path) else default_img
+
+        rgb_left_image = open_optional_image(rgb_left_image_path, rgb_image)
+        rgb_right_image = open_optional_image(rgb_right_image_path, rgb_image)
+        rgb_center_image = open_optional_image(rgb_center_image_path, rgb_image)
+
+        if lidar_image_path and os.path.exists(lidar_image_path):
+            lidar_array = np.load(lidar_image_path)
+            max_val = lidar_array.max()
+            if max_val > 0: lidar_array = (lidar_array / max_val) * 255.0
+            lidar_image = Image.fromarray(lidar_array.astype(np.uint8)).convert('RGB')
+        else:
+            lidar_image = Image.fromarray(np.zeros((224, 224, 3), dtype=np.uint8))
+
+        # 2. تطبيق التحويلات
+        inputs = {
+            'rgb': RGB_TRANSFORM(rgb_image).unsqueeze(0).to(DEVICE), 
+            'rgb_left': RGB_TRANSFORM(rgb_left_image).unsqueeze(0).to(DEVICE), 
+            'rgb_right': RGB_TRANSFORM(rgb_right_image).unsqueeze(0).to(DEVICE),
+            'rgb_center': RGB_TRANSFORM(rgb_center_image).unsqueeze(0).to(DEVICE), 
+            'lidar': LIDAR_TRANSFORM(lidar_image).unsqueeze(0).to(DEVICE),
+            'measurements': torch.tensor([json.load(open(measurements_path))['values']], dtype=torch.float32).to(DEVICE),
+            'target_point': torch.tensor([target_point_list], dtype=torch.float32).to(DEVICE)
+        }
+
+        # 3. تشغيل النموذج
+        with torch.no_grad():
+            outputs = model(inputs)
+        
+        return outputs
+
+    except Exception as e:
+        logging.error(traceback.format_exc())
+        raise gr.Error(f"حدث خطأ أثناء معالجة البيانات: {e}")
+
+
+def run_simulation(model_name: str, scenario_name: str):
+    """
+    المنسق الرئيسي: يجهز المدخلات، يستدعي محرك المعالجة، ثم يعرض النتائج.
+    """
+    logging.info(f"بدء المحاكاة للنموذج: '{model_name}', السيناريو: '{scenario_name}'")
+    
+    # 1. احصل على النموذج (من المخبأ أو قم بتحميله)
+    model = get_model(model_name)
+
+    # 2. جهز المسارات للسيناريو المختار
+    scenario_path = os.path.join(SAMPLE_DATA_DIR, scenario_name)
+    paths = {
+        'rgb': os.path.join(scenario_path, 'rgb.png'),
+        'left': os.path.join(scenario_path, 'rgb_left.png'),
+        'right': os.path.join(scenario_path, 'rgb_right.png'),
+        'center': os.path.join(scenario_path, 'rgb_center.png'),
+        'lidar': os.path.join(scenario_path, 'lidar.npy'),
+        'measurements': os.path.join(scenario_path, 'measurements.json'),
+        'target_point': os.path.join(scenario_path, 'target_point.json')
+    }
+
+    # 3. استدعاء محرك المعالجة والنموذج
+    target_point_list = json.load(open(paths['target_point']))['value']
+    
+    traffic, waypoints, is_junction, traffic_light, stop_sign, _ = process_and_run_model(
+        model=model, rgb_image_path=paths['rgb'], rgb_left_image_path=paths['left'],
+        rgb_right_image_path=paths['right'], rgb_center_image_path=paths['center'],
+        lidar_image_path=paths['lidar'], measurements_path=paths['measurements'],
+        target_point_list=target_point_list
+    )
+    
+    # 4. معالجة المخرجات
+    pred_wp = waypoints[0].cpu().numpy()
+    pred_traffic_map = traffic[0].sigmoid().cpu().numpy().reshape(20, 20, 7)
+    current_speed = json.load(open(paths['measurements'])).get('values', [0]*7)[3] # Speed is at index 3
+
+    # 5. استخدام متحكم القيادة للحصول على الأوامر
+    steer, throttle, brake, metadata = GLOBAL_CONTROLLER.run_step(
+        current_speed=current_speed, waypoints=pred_wp,
+        junction=torch.sigmoid(is_junction)[0,1].item(),
+        traffic_light_state=torch.sigmoid(traffic_light)[0,0].item(),
+        stop_sign=torch.sigmoid(stop_sign)[0,1].item(),
+        meta_data={}
+    )
+    
+    # 6. إنشاء الخرائط المرئية
+    traffic_render, _ = render(pred_traffic_map)
+    waypoints_render = render_waypoints(pred_wp)
+    combined_map = cv2.addWeighted(traffic_render, 1.0, waypoints_render, 1.0, 0.0)
+    final_map = render_self_car(combined_map)
+
+    # 7. تجهيز البيانات النهائية للعرض
+    display_data = {
+        'camera_view': np.array(Image.open(paths['rgb'])),
+        'map_t0': final_map, 'map_t1': np.zeros_like(final_map), 'map_t2': np.zeros_like(final_map),
+        'text_info': {
+            "Model": f"Model: {model_name}",
+            "Scenario": f"Scenario: {scenario_name}",
+            "Controller": metadata,
+            "Brake": f"Brake Activated: {'YES' if brake else 'NO'}"
+        }
+    }
+    
+    dashboard = GLOBAL_DISPLAY_INTERFACE.run_interface(display_data)
+    logging.info("انتهت المحاكاة بنجاح.")
+    
+    return dashboard, metadata
+
+
+# --- الج��ء السادس: بناء واجهة Gradio ---
+with gr.Blocks(title="Interfuser Demo", theme=gr.themes.Soft()) as demo:
+    gr.Markdown("# 🚀 Interfuser: واجهة القيادة التفاعلية")
+    gr.Markdown("اختر النموذج والسيناريو، ثم اضغط على 'تشغيل' للمقارنة بين أداء النماذج المختلفة.")
+
+    # قراءة النماذج والسيناريوهات المتاحة من المجلدات
+    try:
+        available_models = [f for f in os.listdir(MODEL_DIR) if f.endswith(('.pth', '.pt'))]
+        available_scenarios = [d for d in os.listdir(SAMPLE_DATA_DIR) if os.path.isdir(os.path.join(SAMPLE_DATA_DIR, d))]
+    except FileNotFoundError as e:
+        gr.Error(f"خطأ في إعداد الواجهة: {e}. هل قمت بإنشاء مجلدي 'model' و 'sample_data'؟")
+        available_models = []
+        available_scenarios = []
+
+    with gr.Row():
+        with gr.Column(scale=1, min_width=300):
+            # مدخلات المستخدم
+            model_selector = gr.Dropdown(
+                choices=available_models, label="1. اختر النموذج",
+                value=available_models[0] if available_models else None)
+            
+            scenario_selector = gr.Dropdown(
+                choices=available_scenarios, label="2. اختر سيناريو القيادة",
+                value=available_scenarios[0] if available_scenarios else None)
+            
+            run_button = gr.Button("▶️ تشغيل المحاكاة", variant="primary")
+            
+            # مخرجات نصية
+            controller_output = gr.Textbox(label="بيانات متحكم القيادة", interactive=False)
+
+        with gr.Column(scale=3):
+            # مخرجات مرئية
+            dashboard_output = gr.Image(label="لوحة المعلومات الحية (Dashboard)", interactive=False)
+
+    # ربط زر التشغيل بالدالة الرئيسية
+    if available_models and available_scenarios:
+        run_button.click(
+            fn=run_simulation,
+            inputs=[model_selector, scenario_selector],
+            outputs=[dashboard_output, controller_output]
+        )
+    else:
+        gr.Warning("لا يمكن العثور على نماذج أو سيناريوهات. يرجى التأكد من إعداد المجلدات بشكل صحيح.")
+
+# --- الجزء السابع: إطلاق الواجهة ---
+if __name__ == "__main__":
+    demo.launch(debug=True) # استخدم debug=True للمساعدة في اكتشاف الأخطاء أثناء التطوير