Delete app copy 2.py

app copy 2.py

@@ -1,400 +0,0 @@
-import gradio as gr
-import numpy as np
-import pygame
-import threading
-import queue
-from PIL import Image
-from car_game_classes_gradio import Car, RayCasting, MapGenerator
-from ppo_agent_gradio import PPOAgent
-from collections import deque
-
-# Initialize Pygame
-pygame.init()
-SCREEN_WIDTH, SCREEN_HEIGHT = 1600, 900
-screen = pygame.Surface((SCREEN_WIDTH, SCREEN_HEIGHT))
-clock = pygame.time.Clock()
-
-# Define constants
-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
-
-# Frame queue for Gradio
-frame_queue = queue.Queue(maxsize=1)
-
-def point_to_line_distance(point, line_p0, line_p1):
-    p = pygame.math.Vector2(point)
-    a = pygame.math.Vector2(line_p0)
-    b = pygame.math.Vector2(line_p1)
-    ap = p - a
-    ab = b - a
-    magnitude_ab = ab.length_squared()
-    if magnitude_ab == 0:
-        return ap.length()
-    projection = ap.dot(ab) / magnitude_ab
-    projection = max(0, min(1, projection))
-    closest = a + projection * ab
-    return p.distance_to(closest)
-
-class TrainingEnvironment:
-    def __init__(self, action_space=8):
-        self.action_space = action_space
-        self.map_generator = MapGenerator(SCREEN_WIDTH, SCREEN_HEIGHT, 2, 18)  # Difficulty 2 = lvl 3 in training
-        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))
-        self.start_pos = None
-        self.start_dir = None
-        self.previous_speed = 0.0
-        self.speed_history = deque(maxlen=5)
-        self.steering_history = deque(maxlen=10)
-        self.steering_persistence = {'direction': 0, 'count': 0}
-        self.last_steering_reward = 0.0
-        self.prev_pos = None
-        self.passed_goals = []
-        self.last_action = 0
-        self.reset()
-
-    def generate_new_map(self):
-        start_pos, start_dir = self.map_generator.update()
-        self.start_pos = pygame.math.Vector2(start_pos)
-        self.start_dir = pygame.math.Vector2(start_dir).normalize()
-
-    def reset_car(self):
-        if self.start_pos is None or self.start_dir is None:
-            raise ValueError("Map not generated yet")
-        self.car = Car(self.car_image, self.start_pos)
-        self.car.direction = self.start_dir
-        self.rays = RayCasting(
-            self.car.pos, self.car.direction, self.map_generator.surface,
-            [-60, -30, -15, 0, 15, 30, 60],
-            [14, 22, 27, 26, 27, 22, 14],
-            draw_rays=False)
-        self.current_goal_index = 0
-        self.passed_goals = []
-        self.prev_pos = self.car.pos.copy()
-        self.previous_speed = 0.0
-        self.speed_history.clear()
-        self.steering_history.clear()
-        self.steering_persistence = {'direction': 0, 'count': 0}
-        self.last_steering_reward = 0.0
-        self.last_action = 0
-
-    def reset(self):
-        self.map_generator.difficulty = 2  # Ensure difficulty matches training
-        self.generate_new_map()
-        self.reset_car()
-
-    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]
-        center_ray = norm_distances[3]
-        right_rays = norm_distances[4:7]
-        left_avg = sum(left_rays) / len(left_rays)
-        right_avg = sum(right_rays) / len(right_rays)
-        space_difference = right_avg - left_avg
-        steering_needed = False
-        preferred_direction = 0
-        urgency = 0.0
-        min_side_distance = min(min(left_rays), min(right_rays))
-        if min_side_distance < 0.3:
-            steering_needed = True
-            urgency = 1.0 - min_side_distance
-            preferred_direction = 1 if left_avg < right_avg else -1
-        elif abs(space_difference) > 0.15:
-            steering_needed = True
-            urgency = min(abs(space_difference), 1.0)
-            preferred_direction = 1 if space_difference > 0 else -1
-        elif center_ray < 0.4:
-            steering_needed = True
-            urgency = 1.0 - center_ray
-            preferred_direction = 1 if left_avg < right_avg else -1
-        return {
-            'needed': steering_needed,
-            'direction': preferred_direction,
-            'urgency': urgency,
-            'space_diff': space_difference,
-            'left_space': left_avg,
-            'right_space': right_avg,
-            '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
-        keys = {'w': False, 'a': False, 'd': False, 's': False}
-        steer_factor = 1.0
-        if action_int == 0:
-            keys['w'] = True
-        elif action_int == 1:
-            keys['a'] = True
-            steer_factor = 1.0
-        elif action_int == 2:
-            keys['d'] = True
-            steer_factor = 1.0
-        elif action_int == 3:
-            keys['w'] = True
-            keys['a'] = True
-            steer_factor = 0.8
-        elif action_int == 4:
-            keys['w'] = True
-            keys['d'] = True
-            steer_factor = 0.8
-        elif action_int == 5:
-            keys['s'] = True
-            original_brake = self.car.brake_deceleration
-            self.car.brake_deceleration *= 0.6
-            self.car.update(dt, keys, steer_factor)
-            self.car.brake_deceleration = original_brake
-            return
-        elif action_int == 6:
-            keys['s'] = True
-            keys['a'] = True
-            steer_factor = 1.2
-        elif action_int == 7:
-            keys['s'] = True
-            keys['d'] = True
-            steer_factor = 1.2
-        self.car.update(dt, keys, steer_factor)
-
-    def detect_curve_severity(self, ray_distances, steering_info):
-        normalized_ray_distances = [min(distance / self.rays.max_distance, 1.0) for distance in ray_distances]
-        longest_distance = max(normalized_ray_distances)
-        longest_index = normalized_ray_distances.index(longest_distance)
-        main_ray_index = 3
-        if longest_index == main_ray_index:
-            return 0.0
-        left_adjacent_index = max(0, longest_index - 1)
-        right_adjacent_index = min(len(normalized_ray_distances) - 1, longest_index + 1)
-        left_adjacent_distance = normalized_ray_distances[left_adjacent_index]
-        right_adjacent_distance = normalized_ray_distances[right_adjacent_index]
-        if left_adjacent_distance > right_adjacent_distance:
-            longer_adjacent_index = left_adjacent_index
-            longer_adjacent_distance = left_adjacent_distance
-        else:
-            longer_adjacent_index = right_adjacent_index
-            longer_adjacent_distance = right_adjacent_distance
-        weight = longer_adjacent_distance / longest_distance
-        ray_angles = [-60, -30, -15, 0, 15, 30, 60]
-        longest_ray_angle = ray_angles[longest_index]
-        longer_adjacent_angle = ray_angles[longer_adjacent_index]
-        longest_pos_angle = longest_ray_angle + weight * (longer_adjacent_angle - longest_ray_angle)
-        front_ray_angle = ray_angles[main_ray_index]
-        angle_difference = abs(longest_pos_angle - front_ray_angle)
-        if angle_difference >= 50:
-            curve_severity = 1.0
-        else:
-            curve_severity = angle_difference / 40.0
-        return curve_severity
-
-    def calculate_reward(self):
-        self.rays.update(self.car.pos, self.car.direction)
-        ray_distances = self.rays.get_ray_distances()
-        collision = self.rays.has_zero_length_ray()
-        if collision:
-            return -100.0, collision, False
-        steering_info = self.detect_steering_need(ray_distances)
-        reward = 0.0
-        curve_severity = self.detect_curve_severity(ray_distances, steering_info)
-        center_distance = ray_distances[3] / self.rays.max_distance
-        if center_distance > 0.25:
-            reward += 0.25 * self.car.velocity - 6
-        if curve_severity > 0.1:
-            if curve_severity > 0.5:
-                if 0.125 < self.car.velocity < 0.25:
-                    reward += self.car.velocity / 10
-                else:
-                    reward += -2
-            elif curve_severity > 0.25:
-                if 0.15 < self.car.velocity < 0.35:
-                    reward += self.car.velocity / 12
-                else:
-                    reward += -2
-            else:
-                if 0.35 < self.car.velocity < 0.7:
-                    reward += self.car.velocity / 10
-                else:
-                    reward += -2
-        else:
-            if self.car.velocity > 40:
-                reward += self.car.velocity / 20
-            elif self.car.velocity < 25:
-                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.0)
-                        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]
-            dist = point_to_line_distance(self.car.pos, goal[0], goal[1])
-            if dist < 8:
-                reward += 7.5
-                self.passed_goals.append(self.current_goal_index)
-                self.current_goal_index += 1
-        all_goals_completed = self.current_goal_index >= total_goals
-        if all_goals_completed:
-            reward += 20.0
-        reward = max(-15.0, min(reward, 12.0))
-        self.prev_pos = pygame.math.Vector2(self.car.pos)
-        self.last_steering_reward = reward
-        return reward, collision, all_goals_completed
-
-def game_loop():
-    try:
-        pygame.init()
-        pygame.display.set_mode((1, 1), pygame.NOFRAME)
-        screen = pygame.Surface((SCREEN_WIDTH, SCREEN_HEIGHT))
-        env = TrainingEnvironment(ACTION_SPACE)
-        env.reset()
-        agent = PPOAgent(state_size, ACTION_SPACE)
-        agent.load(LOADED_MODEL)
-        steps = 0
-        action_counts = [0] * ACTION_SPACE
-        while True:
-            for event in pygame.event.get():
-                if event.type == pygame.QUIT:
-                    pygame.quit()
-                    return
-            state = env.get_state()
-            action, _ = agent.select_action(state)
-            action_int = int(action)
-            action_counts[action_int] += 1
-            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
-            screen.fill((255, 255,255))
-            env.map_generator.draw(screen)
-            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)
-            frame_array = pygame.surfarray.array3d(screen)
-            frame_array = np.transpose(frame_array, (1, 0, 2))
-            pil_image = Image.fromarray(frame_array)
-            pil_image = pil_image.resize((SCREEN_WIDTH, SCREEN_HEIGHT))
-            try:
-                frame_queue.put(pil_image, block=False)
-            except queue.Full:
-                pass
-            clock.tick(SIMULATION_FPS)
-    except Exception as e:
-        print(f"Error in game loop: {e}")
-
-def get_frame():
-    try:
-        return frame_queue.get_nowait()
-    except queue.Empty:
-        return Image.new('RGB', (SCREEN_WIDTH, SCREEN_HEIGHT), (0, 0, 0))
-
-css = """#game-image { 
-    height: calc(100vw * 9 / 16) !important;  
-    object-fit: contain;  
-    margin: 0 auto;  
-    display: block;  
-}"""
-
-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)
-        timer = gr.Timer(value=0.00833)
-        timer.tick(fn=get_frame, inputs=None, outputs=img)
-        demo.launch()
-
-if __name__ == "__main__":
-    execute_interference()