Spaces:
Sleeping
Sleeping
| title: Autonomous Traffic Control Environment | |
| emoji: 🚦 | |
| colorFrom: red | |
| colorTo: green | |
| sdk: docker | |
| pinned: false | |
| app_port: 7860 | |
| base_path: /web | |
| tags: | |
| - openenv | |
| - reinforcement-learning | |
| - traffic-control | |
| - emergency-vehicles | |
| - autonomous-systems | |
| # 🚦 Autonomous Traffic Control — OpenEnv Environment | |
| An LLM-driven reinforcement learning environment for autonomous traffic signal control at a 4-way intersection. Built for the **Meta × PyTorch × SST × OpenEnv Hackathon**. | |
| - **HF Space:** [amanmurari/sst-hack](https://huggingface.co/spaces/amanmurari/sst-hack) | |
| - **GitHub:** [amanmurari/openenv-hc2](https://github.com/amanmurari/openenv-hc2) | |
| --- | |
| ## Overview | |
| An LLM agent controls traffic signals to maximise vehicle throughput while prioritising emergency vehicles. The environment features: | |
| - Sinusoidal traffic wave patterns (realistic rush-hour simulation) | |
| - Emergency vehicles with escalating urgency (urgency^1.5 penalty per waiting step) | |
| - Yellow-light transition state machine | |
| - Traffic surge events in hard tasks | |
| - Multi-objective grading aligned with real traffic KPIs | |
| --- | |
| ## Tasks | |
| | Task | Difficulty | Steps | Key Challenge | | |
| |---|---|---|---| | |
| | `basic_flow` | Easy | 200 | Maximise throughput (target 1.8 veh/step) | | |
| | `emergency_priority` | Medium | 300 | Clear emergencies fast (avg delay < 3 steps) | | |
| | `dynamic_scenarios` | Hard | 400 | Surge traffic + simultaneous emergencies, no collisions | | |
| --- | |
| ## Action & Observation Space | |
| ### Action | |
| ```python | |
| TrafficAction(light_phase: int) | |
| # 0 = NS_GREEN (North-South green, East-West red) | |
| # 1 = EW_GREEN (East-West green, North-South red) | |
| # 2 = ALL_RED (All red — emergency clearance) | |
| ``` | |
| ### Observation | |
| ```python | |
| TrafficObservation( | |
| current_phase: int, # Active phase (0-4, incl. yellow transitions) | |
| time_in_phase: int, # Steps held in current phase | |
| queue_lengths: List[int], # Regular vehicle queue [N, S, E, W] | |
| emergency_queue: List[int], # Emergency vehicle count [N, S, E, W] | |
| emergency_urgency: List[int],# Max urgency 0-10 per approach | |
| vehicles_passed: int, # Regular vehicles cleared this step | |
| emergency_passed: int, # Emergency vehicles cleared this step | |
| avg_wait_time: float, # Avg waiting time across all queued vehicles | |
| queue_trend: List[int], # Queue growth since last step [N, S, E, W] | |
| collision: bool, # Gridlock-induced collision flag | |
| done: bool, | |
| reward: float, | |
| ) | |
| ``` | |
| --- | |
| ## Reward Function | |
| ``` | |
| +0.30 × regular vehicles cleared per step | |
| +12.0 × emergency vehicles cleared per step | |
| -0.08 × total vehicles waiting (queue pressure) | |
| -(urgency^1.5) × 0.5 per waiting emergency vehicle (every step!) | |
| -0.50 to -2.0 for unnecessary phase switch (proportional to empty-queue ratio) | |
| +0.05 stability bonus when traffic flows without switching | |
| -200 gridlock collision (episode-ending catastrophe) | |
| ``` | |
| --- | |
| ## Grading Weights | |
| ### basic_flow | |
| ``` | |
| score = throughput_score × 0.60 + efficiency_score × 0.40 + stability_bonus | |
| throughput_score = min(vehicles_per_step / 1.8, 1.0) | |
| efficiency_score = 1 / (1 + avg_waiting × 0.1) | |
| stability_bonus = max(0, 0.05 × (1 − min(switch_rate × 4, 1.0))) | |
| ``` | |
| ### emergency_priority | |
| ``` | |
| score = throughput × 0.30 + em_rate × 0.35 + delay × 0.20 + efficiency × 0.15 | |
| em_rate = min(emergency_cleared_per_step / (1/20), 1.0) | |
| delay = max(0, 1 − avg_em_delay / 12) | |
| ``` | |
| ### dynamic_scenarios | |
| ``` | |
| score = throughput × 0.25 + em_rate × 0.30 + delay × 0.20 | |
| + efficiency × 0.15 + adaptability × 0.10 | |
| adaptability = 1 / (1 + phase_changes_per_step × 0.5) | |
| ``` | |
| --- | |
| ## Agent Architecture | |
| The inference agent uses a **hybrid heuristic + LLM** architecture: | |
| 1. **Heuristic recommender** — computes directional pressure using the actual reward formula (`urgency^1.5 × 0.5`), applies 5 priority rules (critical emergency, moderate emergency, hysteresis, pressure-based switch, default hold). | |
| 2. **Live score projection** — computes current projected grading scores (throughput, emergency rate, delay, efficiency, adaptability) from `env.state()` and includes them in every LLM prompt. | |
| 3. **Chain-of-thought LLM** — the model reasons through scoring implications then outputs `{"light_phase": N}` on the final line. | |
| 4. **Heuristic fallback** — if LLM output is unparseable, silently falls back to the heuristic. No crashes, no missed steps. | |
| --- | |
| ## Quick Start | |
| ### Connect to the live HF Space | |
| ```python | |
| from traffic_control.client import TrafficControlEnv | |
| from traffic_control.models import TrafficAction | |
| with TrafficControlEnv(base_url="https://amanmurari-sst-hack.hf.space").sync() as env: | |
| result = env.reset(task_id="basic_flow", seed=42) | |
| obs = result.observation | |
| while not result.done: | |
| action = TrafficAction(light_phase=0) # replace with your agent | |
| result = env.step(action) | |
| obs = result.observation | |
| print(f"Cleared: {obs.vehicles_passed} regular, {obs.emergency_passed} emergency | reward={result.reward:.2f}") | |
| ``` | |
| ### Run inference locally | |
| ```bash | |
| # Set required env vars | |
| export API_BASE_URL="https://router.huggingface.co/v1" | |
| export HF_TOKEN="hf_..." | |
| export MODEL_NAME="Qwen/Qwen2.5-72B-Instruct" | |
| export SERVER_URL="http://localhost:7860" | |
| python inference.py | |
| ``` | |
| ### Build and run with Docker | |
| ```bash | |
| docker build -t traffic-control-env . | |
| docker run -p 7860:7860 traffic-control-env | |
| ``` | |
| --- | |
| ## API Endpoints | |
| Once the server is running at `http://localhost:7860`: | |
| | Endpoint | Description | | |
| |---|---| | |
| | `GET /health` | Health check — returns `{"status": "ok"}` | | |
| | `POST /reset` | Reset episode — body: `{"task_id": "basic_flow", "seed": 42}` | | |
| | `POST /step` | Execute action — body: `{"light_phase": 0}` | | |
| | `GET /state` | Cumulative episode state | | |
| | `WS /ws` | WebSocket endpoint for low-latency multi-step sessions | | |
| | `GET /web` | Interactive web dashboard | | |
| | `GET /docs` | OpenAPI / Swagger docs | | |
| --- | |
| ## Project Structure | |
| ``` | |
| traffic_control/ | |
| ├── inference.py # LLM agent (heuristic + chain-of-thought LLM) | |
| ├── client.py # TrafficControlEnv WebSocket client | |
| ├── models.py # TrafficAction / TrafficObservation / TrafficState | |
| ├── environment.py # Core simulation engine | |
| ├── tasks.py # Task graders (basic_flow, emergency_priority, dynamic_scenarios) | |
| ├── dashboard.py # Web UI dashboard | |
| ├── analytics.py # Episode analytics | |
| ├── arena.py # Multi-agent arena | |
| ├── openenv.yaml # OpenEnv manifest | |
| ├── pyproject.toml # Package metadata | |
| ├── Dockerfile # Container (port 7860) | |
| └── server/ | |
| ├── app.py # FastAPI app (HTTP + WebSocket) | |
| └── traffic_control_environment.py | |
| ``` | |
| --- | |
| ## Environment Variables | |
| | Variable | Required | Description | | |
| |---|---|---| | |
| | `API_BASE_URL` | Yes | LLM proxy endpoint (injected by validator) | | |
| | `API_KEY` | Yes | Proxy API key (injected by validator) | | |
| | `HF_TOKEN` | Alt | Hugging Face token (used if `API_KEY` not set) | | |
| | `MODEL_NAME` | No | LLM model (default: `Qwen/Qwen2.5-72B-Instruct`) | | |
| | `SERVER_URL` | No | Env server URL (default: `http://localhost:7860`) | | |
| --- | |
| ## Stdout Format | |
| ``` | |
| [START] task=basic_flow env=traffic_control model=Qwen/Qwen2.5-72B-Instruct | |
| [STEP] step=1 action=light_phase=0 reward=0.65 done=false error=null | |
| [STEP] step=2 action=light_phase=0 reward=0.80 done=false error=null | |
| ... | |
| [END] success=true steps=200 score=0.847 rewards=0.65,0.80,... | |
| ``` | |