bundle overflow_env locally, drop openenv-core git dep (websockets conflict fix)

app.py

@@ -12,7 +12,7 @@ import torch
 import torch.optim as optim
 import gradio as gr
 
-from overflow_env.server.overflow_environment import OverflowEnvironment
+from overflow_env.environment import OverflowEnvironment
 from overflow_env.models import OverflowAction
 from policies.flat_mlp_policy import FlatMLPPolicy
 from policies.ticket_attention_policy import TicketAttentionPolicy

overflow_env/__init__.py

@@ -0,0 +1,2 @@
+from .environment import OverflowEnvironment
+from .models import OverflowAction, OverflowObservation

overflow_env/environment.py

@@ -0,0 +1,295 @@
+"""
+Overflow Environment — standalone bundled version (no openenv.core dependency).
+2D road grid, 5 cars, 3 lanes. Car 0 is the RL agent.
+"""
+
+import math
+import random
+import re
+from dataclasses import dataclass
+from typing import Any, List, Optional
+from uuid import uuid4
+
+from .models import (
+    CarStateData, LaneOccupancyData, OverflowAction,
+    OverflowObservation, OverflowState, Position, ProximityData,
+)
+
+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
+
+REWARD_CRASH = -5.0
+REWARD_NEAR_MISS = -1.0
+REWARD_SAFE_STEP = 0.5
+REWARD_REACHED_GOAL = 3.0
+REWARD_REASONING_MAX = 0.3
+
+MIN_SPEED = 20
+MAX_SPEED = 90
+SPEED_DELTA = 5
+
+
+@dataclass
+class Car:
+    car_id: int
+    lane: int
+    position: float
+    speed: float
+    goal_position: float
+    is_agent: bool = False
+    reached_goal: bool = False
+    prev_speed: float = 0.0
+
+    def distance_to(self, other: "Car") -> float:
+        lane_diff = abs(self.lane - other.lane) * 10.0
+        pos_diff = abs(self.position - other.position)
+        return math.sqrt(lane_diff ** 2 + pos_diff ** 2)
+
+    @property
+    def acceleration(self) -> float:
+        return self.speed - self.prev_speed
+
+    def to_state_data(self) -> 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:
+    valid = {"accelerate", "brake", "lane_change_left", "lane_change_right", "maintain"}
+    decision = action.decision.strip().lower().replace(" ", "_")
+    if decision in valid:
+        return decision
+    text = f"{action.decision} {action.reasoning}".lower()
+    match = re.search(r"<action>\s*(\w+)\s*</action>", text)
+    if match:
+        candidate = match.group(1).strip().replace(" ", "_")
+        if candidate in valid:
+            return candidate
+    for v in ["lane_change_left", "lane_change_right", "accelerate", "brake", "maintain"]:
+        if v in text:
+            return v
+    return "maintain"
+
+
+def _scripted_car_action(car: Car, all_cars: List[Car], rng: random.Random) -> str:
+    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"
+    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:
+    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
+
+
+def _generate_scene_description(agent_car: Car, cars: List[Car]) -> str:
+    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}"
+        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]):
+    cars_data = [c.to_state_data() for c in cars]
+    lane_map: dict = {}
+    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:
+    def __init__(self):
+        self._state = OverflowState(episode_id=str(uuid4()))
+        self._cars: List[Car] = []
+        self._rng = random.Random()
+        self._done = False
+        self._last_obs: Optional[OverflowObservation] = None
+
+    def _build_observation(self, incident_report: str, reward: float,
+                           proximities: Optional[List[ProximityData]] = None) -> OverflowObservation:
+        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, **kwargs: Any) -> OverflowObservation:
+        if seed is not None:
+            self._rng = random.Random(seed)
+        else:
+            self._rng = random.Random()
+        self._state = OverflowState(
+            episode_id=str(uuid4()), step_count=0,
+            crash_count=0, near_miss_count=0, cars_reached_goal=0, total_cars=NUM_CARS,
+        )
+        self._done = False
+        self._cars = []
+        for i in range(NUM_CARS):
+            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)
+                    if math.sqrt(lane_diff ** 2 + pos_diff ** 2) < 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,
+            ))
+        self._last_obs = self._build_observation(incident_report="", reward=0.0)
+        return self._last_obs
+
+    def step(self, action: OverflowAction, **kwargs: Any) -> OverflowObservation:
+        if self._done:
+            return self._build_observation(
+                incident_report="Episode is over. Call reset() to start a new one.", reward=0.0
+            )
+        self._state.step_count += 1
+        reward = 0.0
+        incidents = []
+
+        for car in self._cars:
+            car.prev_speed = car.speed
+
+        decision = _parse_decision(action)
+        _apply_action(self._cars[0], decision)
+
+        for car in self._cars[1:]:
+            if car.reached_goal:
+                continue
+            _apply_action(car, _scripted_car_action(car, self._cars, self._rng))
+
+        for car in self._cars:
+            if not car.reached_goal:
+                car.position += car.speed * 0.1
+
+        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
+                    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:
+            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
+            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
+            if not self._done:
+                reward += REWARD_SAFE_STEP
+
+        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

overflow_env/models.py

@@ -0,0 +1,50 @@
+from typing import List, Optional
+from pydantic import BaseModel, Field
+
+
+class Position(BaseModel):
+    x: float = 0.0
+    y: float = 0.0
+
+
+class CarStateData(BaseModel):
+    carId: int
+    lane: int
+    position: Position
+    speed: float
+    acceleration: float = 0.0
+
+
+class ProximityData(BaseModel):
+    carA: int
+    carB: int
+    distance: float
+
+
+class LaneOccupancyData(BaseModel):
+    lane: int
+    carIds: List[int]
+
+
+class OverflowAction(BaseModel):
+    decision: str = Field(default="maintain")
+    reasoning: str = Field(default="")
+
+
+class OverflowObservation(BaseModel):
+    done: bool = False
+    reward: float = 0.0
+    scene_description: str = ""
+    incident_report: str = ""
+    cars: List[CarStateData] = Field(default_factory=list)
+    proximities: List[ProximityData] = Field(default_factory=list)
+    lane_occupancies: List[LaneOccupancyData] = Field(default_factory=list)
+
+
+class OverflowState(BaseModel):
+    episode_id: str = ""
+    step_count: int = 0
+    crash_count: int = 0
+    near_miss_count: int = 0
+    cars_reached_goal: int = 0
+    total_cars: int = 5

requirements.txt

@@ -2,7 +2,5 @@
 torch==2.5.1+cpu
 numpy>=1.24.0
 pillow==10.4.0
-gradio>=4.44.0
 pydantic>=2.0.0
 requests>=2.31.0
-openenv-overflow-env @ git+https://huggingface.co/spaces/SteveDusty/overflow_env