Upload app.py with huggingface_hub

app.py

@@ -1,12 +1,12 @@
 # ============================================================================
 # app.py - الواجهة التفاعلية لمشروع Interfuser
 # ============================================================================
-# هذا الملف يقوم بإنشاء واجهة ويب باستخدام Gradio لعرض أداء نموذج
-# Interfuser للقيادة الذاتية. يسمح للمستخدمين باختيار نماذج مختلفة
-# وسيناريوهات قيادة ومقارنة النتائج.
+# هذا الملف مسؤول فقط عن بناء وتشغيل واجهة المستخدم باستخدام Gradio.
+# يعتمد على:
+# - model_utils.py: لإدارة وتحميل النماذج.
+# - simulation_modules.py: لمعالجة المخرجات والتحكم والعرض.
 # ============================================================================
 
-# --- الجزء الأول: الاستيرادات ---
 import os
 import torch
 import numpy as np
@@ -15,17 +15,16 @@ 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 model_utils import get_available_models, load_model_by_name, get_current_model
     from simulation_modules import (
         DisplayInterface,
         InterfuserController,
@@ -35,18 +34,16 @@ try:
         render
     )
 except ImportError as e:
-    print(f"خطأ في الاستيراد: تأكد من وجود ملفات model_definition.py و simulation_modules.py. الخطأ: {e}")
+    print(f"خطأ في الاستيراد: تأكد من وجود ملفات model_utils.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"
+DEVICE = torch.device("cuda" if torch.cuda.is_available() else "cpu")
 
-# --- الجزء الثالث: تعريف التحويلات (Transforms) ---
-# هذه التحويلات يجب أن تطابق تمامًا ما تم استخدامه أثناء تدريب النموذج
+# تعريف التحويلات (يجب أن تطابق ما تم استخدامه أثناء التدريب)
 RGB_TRANSFORM = transforms.Compose([
     transforms.Resize((224, 224)),
     transforms.ToTensor(),
@@ -58,89 +55,57 @@ LIDAR_TRANSFORM = transforms.Compose([
     transforms.ToTensor()
 ])
 
-# --- الجزء الرابع: الكائنات العامة ومخبأ النماذج ---
-logging.info("تهيئة المكونات العامة...")
-# قاموس لتخزين النماذج التي تم تحميلها لتجنب إعادة التحميل البطيئة
-GLOBAL_MODELS_CACHE: Dict[str, torch.nn.Module] = {}
-
-# تهيئة واجهة العرض والمتحكم (يتم تهيئتها مرة واحدة)
+# --- الجزء الثالث: الكائنات العامة للواجهة ---
+logging.info("تهيئة مكونات الواجهة...")
 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}'. يرجى مراجعة السجلات.")
+logging.info(f"تم تهيئة مكونات الواجهة. الجهاز المستخدم: {DEVICE}")
 
+# --- الجزء الرابع: دوال المعالجة الرئيسية ---
 
-def process_and_run_model(
+def process_and_run_inference(
     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
+    paths: dict
 ):
     """
-    محرك المعالجة: يعالج إطارًا واحدًا من البيانات بناءً على المسارات
-    ويقوم بتشغيل النموذج عليه.
+    محرك المعالجة: يعالج بيانات إطار واحد بناءً على المسارات ويشغل النموذج.
     """
     try:
-        # 1. قراءة ومعالجة المدخلات
-        if not rgb_image_path or not os.path.exists(rgb_image_path): 
+        rgb_path = paths['rgb']
+        if not rgb_path or not os.path.exists(rgb_path): 
             raise FileNotFoundError("ملف الصورة الأمامية (RGB) غير موجود.")
 
-        rgb_image = Image.open(rgb_image_path).convert("RGB")
+        rgb_image = Image.open(rgb_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)
+        rgb_left_image = open_optional_image(paths.get('left'), rgb_image)
+        rgb_right_image = open_optional_image(paths.get('right'), rgb_image)
+        rgb_center_image = open_optional_image(paths.get('center'), rgb_image)
 
-        if lidar_image_path and os.path.exists(lidar_image_path):
-            lidar_array = np.load(lidar_image_path)
+        lidar_path = paths.get('lidar')
+        if lidar_path and os.path.exists(lidar_path):
+            lidar_array = np.load(lidar_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. تطبيق التحويلات
+        target_point_list = json.load(open(paths['target_point']))['value']
+        
         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),
+            'measurements': torch.tensor([json.load(open(paths['measurements']))['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)
         
@@ -151,16 +116,16 @@ def process_and_run_model(
         raise gr.Error(f"حدث خطأ أثناء معالجة البيانات: {e}")
 
 
-def run_simulation(model_name: str, scenario_name: str):
+def run_simulation(scenario_name: str):
     """
     المنسق الرئيسي: يجهز المدخلات، يستدعي محرك المعالجة، ثم يعرض النتائج.
     """
-    logging.info(f"بدء المحاكاة للنموذج: '{model_name}', السيناريو: '{scenario_name}'")
+    logging.info(f"بدء المحاكاة للسيناريو: '{scenario_name}'")
     
-    # 1. احصل على النموذج (من المخبأ أو قم بتحميله)
-    model = get_model(model_name)
+    model = get_current_model()
+    if model is None:
+        raise gr.Error("لا يوجد نموذج محمل. يرجى اختيار نموذج من القائمة أولاً.")
 
-    # 2. جهز المسارات للسيناريو المختار
     scenario_path = os.path.join(SAMPLE_DATA_DIR, scenario_name)
     paths = {
         'rgb': os.path.join(scenario_path, 'rgb.png'),
@@ -172,22 +137,12 @@ def run_simulation(model_name: str, scenario_name: str):
         '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
-    )
+    traffic, waypoints, is_junction, traffic_light, stop_sign, _ = process_and_run_inference(model, paths)
     
-    # 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
+    current_speed = json.load(open(paths['measurements'])).get('values', [0]*7)[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(),
@@ -196,19 +151,15 @@ def run_simulation(model_name: str, scenario_name: str):
         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'}"
         }
@@ -220,36 +171,26 @@ def run_simulation(model_name: str, scenario_name: str):
     return dashboard, metadata
 
 
-# --- الجزء السادس: بناء واجهة Gradio ---
+# --- الجزء الخامس: بناء واجهة Gradio ---
 with gr.Blocks(title="Interfuser Demo", theme=gr.themes.Soft()) as demo:
     gr.Markdown("# 🚀 Interfuser: واجهة القيادة التفاعلية")
     gr.Markdown("اختر النموذج والسيناريو، ثم اضغط على 'تشغيل' للمقارنة بين أداء النماذج المختلفة.")
 
-    # قراءة النماذج والسيناريوهات المتاحة من المجلدات
     try:
-        # 💡 أضف هذين السطرين للطباعة التشخيصية
-        print("================== DEBUGGING ==================")
-        print(f"🔍 Current working directory is: {os.getcwd()}")
-        print(f"📂 Searching for models in folder: {os.path.abspath(MODEL_DIR)}")
-        
-        available_models = [f for f in os.listdir(MODEL_DIR) if f.endswith(('.pth', '.pt'))]
-        
-        # 💡 وهذا السطر أيضًا
-        print(f"✅ Found models: {available_models}")
-        print("=============================================")
-    
+        available_models = get_available_models()
         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 = []
+        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)
+                value=available_models[0] if available_models else None,
+                interactive=True
+            )
+            model_load_status = gr.Textbox(label="حالة تحميل النموذج", interactive=False)
             
             scenario_selector = gr.Dropdown(
                 choices=available_scenarios, label="2. اختر سيناريو القيادة",
@@ -257,23 +198,37 @@ with gr.Blocks(title="Interfuser Demo", theme=gr.themes.Soft()) as demo:
             
             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)
 
-    # ربط زر التشغيل بالدالة الرئيسية
+    # --- ربط الأحداث ---
+    
+    # 1. عند تغيير النمو��ج في القائمة، قم بتحميله
+    model_selector.change(
+        fn=load_model_by_name,
+        inputs=[model_selector],
+        outputs=[model_load_status] # عرض حالة التحميل للمستخدم
+    )
+    
+    # 2. عند الضغط على زر التشغيل، قم بتشغيل المحاكاة
     if available_models and available_scenarios:
         run_button.click(
             fn=run_simulation,
-            inputs=[model_selector, scenario_selector],
+            inputs=[scenario_selector],
             outputs=[dashboard_output, controller_output]
         )
     else:
         gr.Warning("لا يمكن العثور على نماذج أو سيناريوهات. يرجى التأكد من إعداد المجلدات بشكل صحيح.")
+        
+    # تحميل النموذج الافتراضي عند بدء تشغيل الواجهة
+    demo.load(
+        fn=load_model_by_name,
+        inputs=[model_selector],
+        outputs=[model_load_status]
+    )
 
-# --- الجزء السابع: إطلاق الواجهة ---
+# --- الجزء السادس: إطلاق الواجهة ---
 if __name__ == "__main__":
-    demo.launch(debug=True) # استخدم debug=True للمساعدة في اكتشاف الأخطاء أثناء التطوير
+    demo.launch(debug=True)```