Upload 3 files

app.py

@@ -3,6 +3,9 @@ import numpy as np
 import pygame
 import threading
 import queue
+import time
+import io
+import base64
 from PIL import Image, ImageDraw, ImageFont
 from car_game_classes_gradio import Car, RayCasting, MapGenerator
 from ppo_agent_gradio import PPOAgent
@@ -10,7 +13,7 @@ from collections import deque
 
 # Initialize Pygame
 pygame.init()
-SCREEN_WIDTH, SCREEN_HEIGHT = 1600, 900
+SCREEN_WIDTH, SCREEN_HEIGHT = 800, 450  # Reduced resolution for better streaming
 screen = pygame.Surface((SCREEN_WIDTH, SCREEN_HEIGHT))
 clock = pygame.time.Clock()
 
@@ -18,13 +21,18 @@ clock = pygame.time.Clock()
 ACTION_SPACE = 8
 state_size = 22
 LOADED_MODEL = "model/18.6_car_model_epoch_270_lvl3_final.pth"
-SIMULATION_FPS = 120  # Fixed simulation FPS, matching training
-DT = 1.0 / SIMULATION_FPS  # Fixed time step
-DISPLAY_FPS = 24
+SIMULATION_FPS = 60  # Reduced from 120
+DT = 1.0 / SIMULATION_FPS
+STREAM_FPS = 20  # Separate streaming FPS
 DIFFICULTY = 2
 
-# Frame queue for Gradio
-frame_queue = queue.Queue(maxsize=1)
+# Optimized frame buffer using threading-safe deque
+frame_buffer = deque(maxlen=2)  # Small buffer to prevent memory buildup
+buffer_lock = threading.Lock()
+
+# Global variables for performance monitoring
+game_running = False
+performance_stats = {}
 
 def point_to_line_distance(point, line_p0, line_p1):
     p = pygame.math.Vector2(point)
@@ -43,10 +51,11 @@ def point_to_line_distance(point, line_p0, line_p1):
 class TrainingEnvironment:
     def __init__(self, action_space=8):
         self.action_space = action_space
-        self.map_generator = MapGenerator(SCREEN_WIDTH, SCREEN_HEIGHT, DIFFICULTY, 18)  # Difficulty 2 = lvl 3 in training
+        self.map_generator = MapGenerator(SCREEN_WIDTH, SCREEN_HEIGHT, DIFFICULTY, 18)
         car_surf = pygame.image.load("assets/car_transparent.png").convert_alpha()
         w, h = car_surf.get_size()
-        self.car_image = pygame.transform.scale(car_surf, (w//18, h//18))
+        # Further scale down car for performance
+        self.car_image = pygame.transform.scale(car_surf, (w//24, h//24))
         self.start_pos = None
         self.start_dir = None
         self.previous_speed = 0.0
@@ -61,6 +70,7 @@ class TrainingEnvironment:
         self.map_generator.draw(self.background)
         self.reset()
 
+    # ... (keeping all your existing methods unchanged for brevity)
     def generate_new_map(self):
         start_pos, start_dir = self.map_generator.update()
         self.start_pos = pygame.math.Vector2(start_pos)
@@ -92,6 +102,38 @@ class TrainingEnvironment:
         self.map_generator.draw(self.background)
         self.reset_car()
 
+    def get_state(self):
+        self.rays.update(self.car.pos, self.car.direction)
+        ray_distances = self.rays.get_ray_distances()
+        normalized_ray_distances = [min(distance / self.rays.max_distance, 1.0) for distance in ray_distances]
+        steering_info = self.detect_steering_need(ray_distances)
+        state = [
+            self.car.velocity / self.car.max_velocity,
+            self.car.direction.x,
+            self.car.direction.y,
+            int(self.car.is_braking),
+        ]
+        state.extend(normalized_ray_distances)
+        state.extend([
+            float(steering_info['needed']),
+            steering_info['direction'],
+            steering_info['urgency'],
+            steering_info['space_diff'],
+            steering_info['left_space'],
+            steering_info['right_space'],
+            steering_info['center_distance']
+        ])
+        total_goals = len(self.map_generator.goals)
+        progress = self.current_goal_index / max(total_goals, 1)
+        state.append(progress)
+        state.append(self.last_action / 4.0)
+        speed_change = self.car.velocity - self.previous_speed
+        self.speed_history.append(speed_change)
+        avg_speed_change = sum(self.speed_history) / max(len(self.speed_history), 1)
+        state.extend([speed_change / 100.0, avg_speed_change / 100.0])
+        self.previous_speed = self.car.velocity
+        return np.array(state, dtype=np.float32)
+
     def detect_steering_need(self, ray_distances):
         norm_distances = [min(d / self.rays.max_distance, 1.0) for d in ray_distances]
         left_rays = norm_distances[0:3]
@@ -126,38 +168,6 @@ class TrainingEnvironment:
             'center_distance': center_ray
         }
 
-    def get_state(self):
-        self.rays.update(self.car.pos, self.car.direction)
-        ray_distances = self.rays.get_ray_distances()
-        normalized_ray_distances = [min(distance / self.rays.max_distance, 1.0) for distance in ray_distances]
-        steering_info = self.detect_steering_need(ray_distances)
-        state = [
-            self.car.velocity / self.car.max_velocity,
-            self.car.direction.x,
-            self.car.direction.y,
-            int(self.car.is_braking),
-        ]
-        state.extend(normalized_ray_distances)
-        state.extend([
-            float(steering_info['needed']),
-            steering_info['direction'],
-            steering_info['urgency'],
-            steering_info['space_diff'],
-            steering_info['left_space'],
-            steering_info['right_space'],
-            steering_info['center_distance']
-        ])
-        total_goals = len(self.map_generator.goals)
-        progress = self.current_goal_index / max(total_goals, 1)
-        state.append(progress)
-        state.append(self.last_action / 4.0)
-        speed_change = self.car.velocity - self.previous_speed
-        self.speed_history.append(speed_change)
-        avg_speed_change = sum(self.speed_history) / max(len(self.speed_history), 1)
-        state.extend([speed_change / 100.0, avg_speed_change / 100.0])
-        self.previous_speed = self.car.velocity
-        return np.array(state, dtype=np.float32)
-
     def execute_action_dispatch(self, action, dt):
         action_int = int(action)
         self.last_action = action_int
@@ -261,56 +271,6 @@ class TrainingEnvironment:
                 reward += -0.5
             if hasattr(self, 'speed_consistency_tracker'):
                 self.speed_consistency_tracker = {'appropriate': False, 'count': 0}
-        if steering_info['needed']:
-            steering_urgency = steering_info['urgency']
-            base_steering_reward = 5.0 + (steering_urgency * 4.0)
-            correct_action = False
-            if steering_info['direction'] == -1:
-                if self.last_action in [1, 3, 6]:
-                    correct_action = True
-                    if self.steering_persistence['direction'] == -1:
-                        self.steering_persistence['count'] += 2
-                        persistence_bonus = min(3.0, self.steering_persistence['count'] * 0.5)
-                        base_steering_reward += persistence_bonus
-                    else:
-                        self.steering_persistence = {'direction': -1, 'count': 1}
-                    if curve_severity > 0.4 and self.last_action == 6:
-                        base_steering_reward += 5.0
-            elif steering_info['direction'] == 1:
-                if self.last_action in [2, 4, 7]:
-                    correct_action = True
-                    if self.steering_persistence['direction'] == 1:
-                        self.steering_persistence['count'] += 2
-                        persistence_bonus = min(3.0, self.steering_persistence['count'] * 0.5)
-                        base_steering_reward += persistence_bonus
-                    else:
-                        self.steering_persistence = {'direction': 1, 'count': 1}
-                    if curve_severity > 0.4 and self.last_action == 7:
-                        base_steering_reward += 5.0
-            if correct_action:
-                reward += base_steering_reward
-            else:
-                self.steering_persistence = {'direction': 0, 'count': 0}
-                if steering_urgency > 0.6:
-                    penalty = -4.0 * steering_urgency
-                    reward += penalty
-        else:
-            self.steering_persistence = {'direction': 0, 'count': 0}
-        speed_factor = self.car.velocity / self.car.max_velocity
-        min_side_distance = min([ray_distances[i] for i in [0, 1, 2, 4, 5, 6]]) / self.rays.max_distance
-        center_distance = ray_distances[3] / self.rays.max_distance
-        if min_side_distance < 0.01:
-            danger_penalty = -5.0 * (1.0 - min_side_distance * 12.5) * (0.5 + speed_factor * 0.5)
-            reward += danger_penalty
-        elif min_side_distance < 0.0166:
-            if speed_factor > 0.7:
-                minor_penalty = -2 * (1.0 - min_side_distance * 6.67)
-                reward += minor_penalty
-        if 0.01666 <= min_side_distance and center_distance > 0.2:
-            safety_efficiency_bonus = 0.3
-            if 0.3 <= speed_factor <= 0.8:
-                safety_efficiency_bonus += 0.2
-            reward += safety_efficiency_bonus
         total_goals = len(self.map_generator.goals)
         if self.current_goal_index < total_goals:
             goal = self.map_generator.goals[self.current_goal_index]
@@ -327,7 +287,45 @@ class TrainingEnvironment:
         self.last_steering_reward = reward
         return reward, collision, all_goals_completed
 
+def create_optimized_frame(screen, env, fps):
+    """Create and optimize frame for streaming"""
+    # Use a smaller render surface for better performance
+    render_surface = pygame.Surface((SCREEN_WIDTH, SCREEN_HEIGHT))
+    render_surface.blit(env.background, (0, 0))
+    env.car.draw(render_surface)
+    
+    # Draw only every other ray for better performance
+    for i in range(0, len(env.rays.last_collisions), 2):  # Skip every other ray
+        if i < len([-60, -30, -15, 0, 15, 30, 60]):
+            angle = [-60, -30, -15, 0, 15, 30, 60][i]
+            offset = [14, 22, 27, 26, 27, 22, 14][i]
+            ray_dir = env.car.direction.rotate(angle).normalize()
+            origin = env.car.pos + ray_dir * offset
+            end_point = env.rays.last_collisions[i]
+            pygame.draw.line(render_surface, (255, 255, 0), 
+                           (int(origin.x), int(origin.y)),
+                           (int(end_point.x), int(end_point.y)),
+                           width=1)  # Thinner lines
+    
+    # Convert to PIL with optimizations
+    raw_str = pygame.image.tostring(render_surface, 'RGB')
+    pil_image = Image.frombytes('RGB', (SCREEN_WIDTH, SCREEN_HEIGHT), raw_str)
+    
+    # Add FPS overlay more efficiently
+    draw = ImageDraw.Draw(pil_image)
+    try:
+        font = ImageFont.load_default()
+        fps_text = f"FPS: {fps:.0f}"
+        draw.text((10, 10), fps_text, fill=(255, 255, 255), font=font)
+    except:
+        pass
+    
+    return pil_image
+
 def game_loop():
+    """Optimized game loop with better frame management"""
+    global game_running, performance_stats
+    
     try:
         pygame.init()
         pygame.display.set_mode((1, 1), pygame.NOFRAME)
@@ -336,22 +334,28 @@ def game_loop():
         env.reset()
         agent = PPOAgent(state_size, ACTION_SPACE)
         agent.load(LOADED_MODEL)
+        
         steps = 0
         action_counts = [0] * ACTION_SPACE
         clock = pygame.time.Clock()
         
-        # FPS measurement variables
-        start_time = pygame.time.get_ticks()
-        frame_count = 0
-        fps = 0
+        # Performance tracking
+        frame_times = deque(maxlen=60)
+        last_stream_time = time.time()
+        stream_interval = 1.0 / STREAM_FPS
+        
+        game_running = True
         
-        while True:
+        while game_running:
+            frame_start = time.time()
+            
+            # Handle pygame events
             for event in pygame.event.get():
                 if event.type == pygame.QUIT:
-                    pygame.quit()
+                    game_running = False
                     return
             
-            # Game logic
+            # Game logic (runs at full simulation FPS)
             state = env.get_state()
             action, _ = agent.select_action(state)
             action_int = int(action)
@@ -359,91 +363,138 @@ def game_loop():
             env.execute_action_dispatch(action_int, DT)
             reward, collision, all_goals_completed = env.calculate_reward()
             env.rays.update(env.car.pos, env.car.direction)
+            
             if collision or all_goals_completed:
                 env.reset()
             
             steps += 1
-            frame_count += 1
-            
-            # Calculate FPS every second
-            current_time = pygame.time.get_ticks()
-            elapsed = (current_time - start_time) / 1000  # in seconds
-            if elapsed >= 1:
-                fps = frame_count / elapsed
-                frame_count = 0
-                start_time = current_time
-            
-            # Render
-            screen.blit(env.background, (0, 0))  # Cached map
-            env.car.draw(screen)
-            for i, (angle, offset) in enumerate(zip([-60, -30, -15, 0, 15, 30, 60], [14, 22, 27, 26, 27, 22, 14])):
-                if i < len(env.rays.last_collisions):
-                    ray_dir = env.car.direction.rotate(angle).normalize()
-                    origin = env.car.pos + ray_dir * offset
-                    end_point = env.rays.last_collisions[i]
-                    pygame.draw.line(screen, (255, 255, 0), 
-                                   (int(origin.x), int(origin.y)),
-                                   (int(end_point.x), int(end_point.y)),
-                                   width=2)
             
-            # Convert to PIL image and resize for display
-            raw_str = pygame.image.tostring(screen, 'RGB')
-            pil_image = Image.frombytes('RGB', (SCREEN_WIDTH, SCREEN_HEIGHT), raw_str)
-            pil_image = pil_image.resize((int(SCREEN_WIDTH/1), int(SCREEN_HEIGHT/1)))
-            
-            # Add FPS text overlay to the image
-            try:
-                draw = ImageDraw.Draw(pil_image)
-                # Try to use a default font, fallback to basic if not available
-                try:
-                    font = ImageFont.truetype("arial.ttf", 20)
-                except:
-                    font = ImageFont.load_default()
+            # Stream frame at reduced rate
+            current_time = time.time()
+            if current_time - last_stream_time >= stream_interval:
+                fps = 1.0 / (current_time - frame_start + 0.001)  # Avoid division by zero
+                frame_times.append(current_time - frame_start)
+                avg_fps = len(frame_times) / sum(frame_times) if frame_times else 0
                 
-                fps_text = f"FPS: {fps:.1f}"
-                draw.text((10, 10), fps_text, fill=(255, 255, 255), font=font)
-            except Exception as e:
-                print(f"Warning: Could not add FPS overlay: {e}")
-            
-            # Send only the PIL image to Gradio
-            try:
-                frame_queue.put(pil_image, block=False)
-            except queue.Full:
-                pass
+                # Create and buffer frame
+                frame = create_optimized_frame(screen, env, avg_fps)
+                
+                with buffer_lock:
+                    frame_buffer.append(frame)
+                
+                performance_stats['fps'] = avg_fps
+                performance_stats['frame_time'] = sum(frame_times) / len(frame_times) if frame_times else 0
+                last_stream_time = current_time
             
+            # Control simulation FPS
             clock.tick(SIMULATION_FPS)
+            
     except Exception as e:
         print(f"Error in game loop: {e}")
+        game_running = False
 
 def get_frame():
+    """Get the latest frame from buffer"""
     try:
-        return frame_queue.get_nowait()
-    except queue.Empty:
-        return Image.new('RGB', (SCREEN_WIDTH//2, SCREEN_HEIGHT//2), (0, 0, 0))
+        with buffer_lock:
+            if frame_buffer:
+                return frame_buffer[-1]  # Get most recent frame
+    except Exception as e:
+        print(f"Error getting frame: {e}")
+    
+    # Return black frame as fallback
+    return Image.new('RGB', (SCREEN_WIDTH, SCREEN_HEIGHT), (0, 0, 0))
 
-css = """#game-image { 
-    height: calc(100vw * 9 / 16) !important;  
+def start_game():
+    """Start the game and return initial status"""
+    global game_running
+    if not game_running:
+        game_thread = threading.Thread(target=game_loop, daemon=True)
+        game_thread.start()
+        return "Game Started!"
+    return "Game Already Running!"
+
+def stop_game():
+    """Stop the game"""
+    global game_running
+    game_running = False
+    return "Game Stopped!"
+
+def update_difficulty(diff_level):
+    """Update difficulty level"""
+    global DIFFICULTY
+    DIFFICULTY = int(diff_level)
+    return f"Difficulty set to Level {DIFFICULTY}"
+
+# Optimized CSS for better performance
+css = """
+#game-image { 
+    height: 450px !important;
+    width: 800px !important;
     object-fit: contain;  
     margin: 0 auto;  
-    display: block;  
-}"""
+    display: block;
+    image-rendering: pixelated;
+    image-rendering: -moz-crisp-edges;
+    image-rendering: crisp-edges;
+}
+.gradio-container {
+    max-width: 900px !important;
+}
+"""
 
-def update_difficulty(input_diff_lvl):
-    global DIFFICULTY
-    DIFFICULTY = input_diff_lvl
+def create_interface():
+    """Create the Gradio interface with optimizations"""
+    with gr.Blocks(css=css, title="AI Car Game") as demo:
+        gr.Markdown("# 🚗 AI Car Racing Game")
+        gr.Markdown("Real-time AI car racing with PPO agent")
+        
+        img = gr.Image(
+            type="pil", 
+            label="Game Stream", 
+            elem_id="game-image",
+            interactive=False,
+            show_download_button=False,
+            show_share_button=False
+        )
+    
 
-def execute_interference():
-    game_thread = threading.Thread(target=game_loop)
-    game_thread.daemon = True
-    game_thread.start()
-    with gr.Blocks(css=css) as demo:
-        with gr.Column():
-            img = gr.Image(type="pil", label="Game Stream", elem_id="game-image", scale=1, min_width=0)
-            difficulty_slider = gr.Slider(minimum=1, maximum=2, step=1, value=2, label="Difficulty Level")
-            difficulty_slider.change(fn=update_difficulty, inputs=difficulty_slider)
-        timer = gr.Timer(value=1/DISPLAY_FPS) 
-        timer.tick(fn=get_frame, inputs=None, outputs=img)
-        demo.launch()
+        start_btn = gr.Button("🚀 Start Game", variant="primary")
+        stop_btn = gr.Button("⏹️ Stop Game", variant="secondary")
+        
+        difficulty_slider = gr.Slider(
+            minimum=1, 
+            maximum=2, 
+            step=1, 
+            value=2, 
+            label="Difficulty Level"
+        )
+        
+        status_text = gr.Textbox(
+            label="Status", 
+            value="Click Start Game to begin",
+            interactive=False
+        )
+        
+        # Event handlers
+        start_btn.click(fn=start_game, outputs=status_text)
+        stop_btn.click(fn=stop_game, outputs=status_text)
+        difficulty_slider.change(fn=update_difficulty, inputs=difficulty_slider, outputs=status_text)
+        
+        # Optimized timer with longer interval
+        timer = gr.Timer(value=1/STREAM_FPS)  # 20 FPS for streaming
+        timer.tick(fn=get_frame, outputs=img)
+        
+        # Add queue for better performance on HF Spaces  
+        demo.queue(
+            max_size=3,  # Small queue
+            api_open=False
+        )
+    
+    return demo
 
 if __name__ == "__main__":
-    execute_interference()
+    # Enable queue and configure for HF Spaces
+    demo = create_interface()
+    demo.launch(
+    )

car_game_classes_gradio.py

@@ -26,7 +26,6 @@ class MapGenerator:
             road_width = 100
         
         if self.difficulty == 2:
-
             border = 50
             num_ctrl = 5
             x_coords = [i * (self.width/(num_ctrl-1)) for i in range(num_ctrl)]
@@ -35,12 +34,18 @@ class MapGenerator:
             # Generate y coordinates with alternating high/low pattern
             y_coords = []
             
-            # Define high and low ranges
-            high_range = (self.height//2 + 2*border, self.height - int(border*5.5))
-            low_range = (int(border*5.5), self.height//2 - 2*border)
-            
-            # Generate first point randomly
-            first_y = np.random.randint(int(border*5.5), self.height-int(border*5.5))
+            # Generate first point randomly with safe bounds
+            min_y = int(border*5.5)
+            max_y = self.height - int(border*5.5)
+            # Ensure we have valid bounds
+            if min_y >= max_y:
+                min_y = border * 2
+                max_y = self.height - border * 2
+            if min_y >= max_y:  # Still invalid, use fallback
+                min_y = 50
+                max_y = self.height - 50
+                
+            first_y = np.random.randint(min_y, max_y)
             y_coords.append(first_y)
             
             # Determine if first point is high or low
@@ -52,19 +57,35 @@ class MapGenerator:
                 y_previous = y_coords[index_prev]
                 diff_min = 75
                 diff_max = 225
+                
                 if is_first_high:
-
                     if (i - 1) % 2 == 0:  # Previous was high (first point)
-                        y = np.random.randint(y_previous-diff_max, y_previous-diff_min)
+                        low_bound = max(border, y_previous - diff_max)
+                        high_bound = min(self.height - border, y_previous - diff_min)
                     else:  # Previous was low
-                        y = np.random.randint(y_previous+diff_min, y_previous+diff_max)
+                        low_bound = max(border, y_previous + diff_min)
+                        high_bound = min(self.height - border, y_previous + diff_max)
                 else:
-                    # If previous point was low, make this one high
                     if (i - 1) % 2 == 0:  # Previous was low (first point)
-                        y = np.random.randint(y_previous+diff_min, y_previous+diff_max)
+                        low_bound = max(border, y_previous + diff_min)
+                        high_bound = min(self.height - border, y_previous + diff_max)
                     else:  # Previous was high
-                        y = np.random.randint(y_previous-diff_max, y_previous-diff_min)
+                        low_bound = max(border, y_previous - diff_max)
+                        high_bound = min(self.height - border, y_previous - diff_min)
                 
+                # Ensure valid bounds
+                if low_bound >= high_bound:
+                    # Fallback: use a safe range around the previous point
+                    center = y_previous
+                    safe_range = 50
+                    low_bound = max(border, center - safe_range)
+                    high_bound = min(self.height - border, center + safe_range)
+                    if low_bound >= high_bound:
+                        # Ultimate fallback
+                        low_bound = border
+                        high_bound = self.height - border
+                        
+                y = np.random.randint(low_bound, high_bound)
                 y_coords.append(y)
                 
             ctrl_pts = list(zip(x_coords, y_coords))