""" Overflow Environment Implementation. A 2D road grid with N cars. One car (Car 0) is the LLM agent, others follow scripted rules. An observer checks for collisions each step. The environment returns text observations describing the traffic scene and rewards based on safety. Observations carry both text (for the LLM) and structured data (for the frontend). """ import math import random import re from dataclasses import dataclass, field from typing import Any, List, Optional from uuid import uuid4 try: from openenv.core.env_server.interfaces import Environment from openenv.core.env_server.types import State except ImportError: class Environment: # stub for training-only mode pass class State: pass try: from ..models import ( CarStateData, LaneOccupancyData, OverflowAction, OverflowObservation, OverflowState, Position, ProximityData, ) from ..policies.flat_mlp_policy import FlatMLPPolicy from ..policies.ticket_attention_policy import TicketAttentionPolicy from ..policies.policy_spec import OBS_DIM from .policy_adapter import overflow_obs_to_policy_obs, policy_action_to_decision except ImportError: from models import ( CarStateData, LaneOccupancyData, OverflowAction, OverflowObservation, OverflowState, Position, ProximityData, ) from policies.flat_mlp_policy import FlatMLPPolicy from policies.ticket_attention_policy import TicketAttentionPolicy from policies.policy_spec import OBS_DIM from server.policy_adapter import overflow_obs_to_policy_obs, policy_action_to_decision # --- Constants --- NUM_LANES = 3 ROAD_LENGTH = 200 NUM_CARS = 5 MAX_STEPS = 100 CRASH_DISTANCE = 5.0 NEAR_MISS_DISTANCE = 15.0 LANE_WIDTH = 3.7 # metres — matches frontend's makeCar convention # Reward values REWARD_CRASH = -5.0 REWARD_NEAR_MISS = -1.0 REWARD_SAFE_STEP = 0.5 REWARD_REACHED_GOAL = 3.0 REWARD_REASONING_MAX = 0.3 # Speed bounds MIN_SPEED = 20 MAX_SPEED = 90 SPEED_DELTA = 5 @dataclass class Car: """Represents a car on the road grid.""" car_id: int lane: int # 1-indexed: 1, 2, or 3 position: float speed: float goal_position: float is_agent: bool = False reached_goal: bool = False prev_speed: float = 0.0 # speed last step, for acceleration calc def distance_to(self, other: "Car") -> float: """Euclidean-ish distance considering lane and position.""" lane_diff = abs(self.lane - other.lane) * 10.0 # lanes are ~10 units apart pos_diff = abs(self.position - other.position) return math.sqrt(lane_diff**2 + pos_diff**2) @property def acceleration(self) -> float: """Speed delta since last step.""" return self.speed - self.prev_speed def to_state_data(self) -> CarStateData: """Convert to frontend-compatible CarStateData.""" return CarStateData( carId=self.car_id, lane=self.lane, position=Position(x=self.position, y=self.lane * LANE_WIDTH), speed=self.speed, acceleration=self.acceleration, ) def _parse_decision(action: OverflowAction) -> str: """Extract a valid decision from the action, being forgiving about format.""" valid = {"accelerate", "brake", "lane_change_left", "lane_change_right", "maintain"} # Try the decision field directly decision = action.decision.strip().lower().replace(" ", "_") if decision in valid: return decision # Try to extract from free text (the LLM might wrap it in tags) text = f"{action.decision} {action.reasoning}".lower() # Check for ... tags match = re.search(r"\s*(\w+)\s*", text) if match: candidate = match.group(1).strip().replace(" ", "_") if candidate in valid: return candidate # Check for keywords anywhere (ordered: most specific first to avoid ambiguity) for v in ["lane_change_left", "lane_change_right", "accelerate", "brake", "maintain"]: if v in text: return v return "maintain" def _compute_reasoning_bonus(reasoning: str) -> float: """ Compute a small reasoning quality bonus (0.0 to 0.3). Gives a minor reward for providing structured reasoning, kept low so driving performance remains the dominant training signal. """ if not reasoning: return 0.0 score = 0.0 lower = reasoning.lower() # Small bonus for providing any reasoning at all if len(reasoning) > 20: score += 0.1 # Bonus for structured reasoning (not just keyword stuffing) if "" in lower or "because" in lower: score += 0.1 if any(word in lower for word in ["therefore", "so i should", "best option", "i will"]): score += 0.1 return min(score, REWARD_REASONING_MAX) def _scripted_car_action(car: Car, all_cars: List[Car], rng: random.Random) -> str: """ Simple scripted AI for non-agent cars. Rules: - If car ahead in same lane is close (< 20 units): brake - If speed is low and random chance: accelerate - Otherwise: maintain """ # Find nearest car ahead in same lane nearest_ahead_dist = float("inf") for other in all_cars: if other.car_id == car.car_id: continue if other.lane == car.lane and other.position > car.position: dist = other.position - car.position if dist < nearest_ahead_dist: nearest_ahead_dist = dist if nearest_ahead_dist < 20: return "brake" if car.speed < 60 and rng.random() < 0.1: return "accelerate" # Occasionally change lanes to make traffic more dynamic if rng.random() < 0.05: if car.lane > 1 and rng.random() < 0.5: return "lane_change_left" elif car.lane < NUM_LANES: return "lane_change_right" return "maintain" def _apply_action(car: Car, decision: str) -> None: """Apply a driving decision to a car, mutating it in place.""" if decision == "accelerate": car.speed = min(car.speed + SPEED_DELTA, MAX_SPEED) elif decision == "brake": car.speed = max(car.speed - SPEED_DELTA, MIN_SPEED) elif decision == "lane_change_left": if car.lane > 1: car.lane -= 1 elif decision == "lane_change_right": if car.lane < NUM_LANES: car.lane += 1 # "maintain" — no change def _generate_scene_description(agent_car: Car, cars: List[Car]) -> str: """Generate a text description of the current traffic scene.""" lines = [ f"You are Car 0 in lane {agent_car.lane}, position {agent_car.position:.0f}, speed {agent_car.speed:.0f}.", f"Goal: reach position {agent_car.goal_position:.0f}.", "Nearby cars:", ] for car in cars: if car.car_id == agent_car.car_id: continue detail = f"- Car {car.car_id}: lane {car.lane}, position {car.position:.0f}, speed {car.speed:.0f}" # Add context about relative position if car.lane == agent_car.lane: pos_diff = car.position - agent_car.position if pos_diff > 0: detail += f" [AHEAD IN YOUR LANE - {pos_diff:.0f} units away]" else: detail += f" [BEHIND IN YOUR LANE - {abs(pos_diff):.0f} units away]" if car.reached_goal: detail += " [REACHED GOAL]" lines.append(detail) return "\n".join(lines) def _build_structured_data( cars: List[Car], proximity_pairs: List[ProximityData], ) -> tuple[List[CarStateData], List[LaneOccupancyData]]: """Build structured arrays for the observation.""" cars_data = [c.to_state_data() for c in cars] # Lane occupancies lane_map: dict[int, list[int]] = {} for car in cars: if not car.reached_goal: lane_map.setdefault(car.lane, []).append(car.car_id) lane_occupancies = [ LaneOccupancyData(lane=lane, carIds=ids) for lane, ids in sorted(lane_map.items()) ] return cars_data, lane_occupancies class OverflowEnvironment(Environment): """ Autonomous vehicle fleet oversight environment. A 2D road grid with N cars. Car 0 is the LLM agent, others follow scripted rules. The observer detects crashes and near-misses and computes rewards based on safety. """ def __init__(self): super().__init__() self._state = OverflowState(episode_id=str(uuid4())) self._cars: List[Car] = [] self._rng = random.Random() self._done = False self._last_obs: Optional[OverflowObservation] = None self._policies = { "flat_mlp": FlatMLPPolicy(obs_dim=OBS_DIM), "ticket_attention": TicketAttentionPolicy(obs_dim=OBS_DIM), } def _build_observation( self, incident_report: str, reward: float, proximities: Optional[List[ProximityData]] = None, ) -> OverflowObservation: """Build a full observation with text + structured data.""" agent = self._cars[0] scene = _generate_scene_description(agent, self._cars) prox = proximities or [] cars_data, lane_occ = _build_structured_data(self._cars, prox) return OverflowObservation( scene_description=scene, incident_report=incident_report, done=self._done, reward=reward, cars=cars_data, proximities=prox, lane_occupancies=lane_occ, ) def reset( self, seed: Optional[int] = None, episode_id: Optional[str] = None, **kwargs: Any, ) -> OverflowObservation: """Reset the environment: create road and spawn cars.""" if seed is not None: self._rng = random.Random(seed) else: self._rng = random.Random() self._state = OverflowState( episode_id=episode_id or str(uuid4()), step_count=0, crash_count=0, near_miss_count=0, cars_reached_goal=0, total_cars=NUM_CARS, ) self._done = False # Spawn cars with random positions, speeds, lanes, and goals self._cars = [] for i in range(NUM_CARS): # Ensure no two cars spawn within crash distance for _attempt in range(100): lane = self._rng.randint(1, NUM_LANES) position = float(self._rng.randint(10, 80)) too_close = False for existing in self._cars: lane_diff = abs(lane - existing.lane) * 10.0 pos_diff = abs(position - existing.position) dist = math.sqrt(lane_diff**2 + pos_diff**2) if dist < CRASH_DISTANCE * 2: too_close = True break if not too_close: break speed = float(self._rng.randint(40, 70)) goal = float(self._rng.randint(160, 195)) self._cars.append( Car( car_id=i, lane=lane, position=position, speed=speed, goal_position=goal, is_agent=(i == 0), prev_speed=speed, # no delta on first step ) ) self._last_obs = self._build_observation(incident_report="", reward=0.0) return self._last_obs def step( self, action: OverflowAction, timeout_s: Optional[float] = None, **kwargs: Any, ) -> OverflowObservation: """Execute one simulation step.""" if self._done: return self._build_observation( incident_report="Episode is over. Call reset() to start a new one.", reward=0.0, ) # Policy intercept: decision="policy:flat_mlp" or "policy:ticket_attention" if action.decision.startswith("policy:") and self._last_obs is not None: policy_name = action.decision.split(":", 1)[1].lower() if policy_name in self._policies: obs_vec = overflow_obs_to_policy_obs(self._last_obs) act_vec = self._policies[policy_name].predict(obs_vec) decision, reasoning = policy_action_to_decision(act_vec) action = OverflowAction( decision=decision, reasoning=f"[{policy_name}] {reasoning}", ) self._state.step_count += 1 reward = 0.0 incidents = [] # Snapshot previous speeds for acceleration tracking for car in self._cars: car.prev_speed = car.speed # 1. Parse and apply the agent's action to Car 0 decision = _parse_decision(action) _apply_action(self._cars[0], decision) # 2. Compute and apply scripted actions for Cars 1-N for car in self._cars[1:]: if car.reached_goal: continue scripted_decision = _scripted_car_action(car, self._cars, self._rng) _apply_action(car, scripted_decision) # 3. Move all cars forward based on speed (speed is in units/step, scaled down) for car in self._cars: if car.reached_goal: continue car.position += car.speed * 0.1 # scale factor for reasonable movement # 4. Collision detection (pairwise) agent_crash = False proximity_list: List[ProximityData] = [] active_cars = [c for c in self._cars if not c.reached_goal] agent_id = self._cars[0].car_id for i in range(len(active_cars)): for j in range(i + 1, len(active_cars)): dist = active_cars[i].distance_to(active_cars[j]) involves_agent = active_cars[i].car_id == agent_id or active_cars[j].car_id == agent_id if dist < CRASH_DISTANCE: self._state.crash_count += 1 proximity_list.append( ProximityData( carA=active_cars[i].car_id, carB=active_cars[j].car_id, distance=round(dist, 2), ) ) incidents.append( f"CRASH between Car {active_cars[i].car_id} and Car {active_cars[j].car_id}! " f"(distance: {dist:.1f})" ) if involves_agent: agent_crash = True elif dist < NEAR_MISS_DISTANCE: self._state.near_miss_count += 1 # Only penalize near misses involving the agent if involves_agent: reward += REWARD_NEAR_MISS proximity_list.append( ProximityData( carA=active_cars[i].car_id, carB=active_cars[j].car_id, distance=round(dist, 2), ) ) incidents.append( f"NEAR MISS between Car {active_cars[i].car_id} and Car {active_cars[j].car_id} " f"(distance: {dist:.1f})" ) if agent_crash: reward += REWARD_CRASH self._done = True else: # 5. Goal check for agent car agent = self._cars[0] if agent.position >= agent.goal_position: agent.reached_goal = True self._state.cars_reached_goal += 1 reward += REWARD_REACHED_GOAL incidents.append( f"Car 0 reached its goal at position {agent.goal_position:.0f}!" ) self._done = True # Check goal for scripted cars too (for state tracking) for car in self._cars[1:]: if not car.reached_goal and car.position >= car.goal_position: car.reached_goal = True self._state.cars_reached_goal += 1 # 6. Safe step bonus (no crash, agent still active) if not self._done: reward += REWARD_SAFE_STEP # 7. Reasoning quality bonus reasoning_bonus = _compute_reasoning_bonus(action.reasoning) reward += reasoning_bonus # 8. Max steps check if self._state.step_count >= MAX_STEPS and not self._done: self._done = True incidents.append(f"Maximum steps ({MAX_STEPS}) reached.") incident_report = ( "\n".join(incidents) if incidents else "Observer: No incidents this step." ) self._last_obs = self._build_observation( incident_report=incident_report, reward=reward, proximities=proximity_list, ) return self._last_obs @property def state(self) -> OverflowState: """Get the current environment state.""" return self._state