README.md

metadata

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
• GitHub: 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

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

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

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

# 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

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,...