trafficenv

India Unregulated Intersection Traffic Controller

OpenEnv-compatible reinforcement learning environment for traffic signal control in Indian-style unregulated intersections, including non-compliance, obstructions, emergency windows, and congestion propagation.

Key Features

• Dense reward in [0, 1] for stable RL training loops
• Three tasks (easy, medium, hard) with increasing realism and difficulty
• Structured lane/intersection observations with reward breakdown diagnostics
• FastAPI server mode for HTTP/WebSocket interaction
• Local class mode and OpenEnv container modes
• Inference runner (inference.py) with strict [START]/[STEP]/[END] stdout contract

Table of Contents

• Tech Stack
• Project Structure
• Prerequisites
• Getting Started
• Running the Environment
• Inference Runner
• RL API
• Observation Space
• Action Space
• Environment Dynamics
• Reward Function
• Tasks
• Grading Utilities
• HTTP API Endpoints
• Environment Variables
• Docker
• OpenEnv Commands
• Troubleshooting

Tech Stack

• Language: Python 3.10+
• Core runtime: openenv-core[core]>=0.2.2
• Numerics: NumPy
• Validation/models: Pydantic
• Server: FastAPI + Uvicorn
• Packaging: Setuptools + pyproject.toml
• Container: Docker (multi-stage build using openenv-base)

Project Structure

traffic_env/
├── __init__.py                      # Package exports (TrafficEnv, grading helpers, models)
├── client.py                        # EnvClient wrapper for remote/container interaction
├── models.py                        # Action/observation/reward/grading schemas
├── inference.py                     # Inference runner with logging contract
├── server/
│   ├── app.py                       # FastAPI app wiring
│   ├── traffic_env.py               # Environment dynamics implementation
│   ├── __init__.py
│   └── requirements.txt
├── openenv.yaml                     # OpenEnv metadata (app entrypoint, port)
├── pyproject.toml                   # Package metadata and script entrypoint
├── Dockerfile
└── README.md

Prerequisites

• Python >=3.10
• uv (recommended) or pip
• Docker (optional, only for container mode)
• Optional LLM credentials for inference.py:
  • HF_TOKEN or API_KEY
  • API_BASE_URL
  • MODEL_NAME

Getting Started

1. Clone and enter project

git clone <your-repo-url>
cd traffic_env

2. Install dependencies

Using uv:

uv sync

Using pip:

python -m venv .venv
# Linux/macOS
source .venv/bin/activate
# Windows PowerShell
# .venv\Scripts\Activate.ps1

pip install -U pip
pip install -e .

3. Quick import check

python -c "from traffic_env import TrafficEnv; print(TrafficEnv.__name__)"

Expected output:

TrafficEnv

Running the Environment

Option A: Local class mode (no server)

from traffic_env import TrafficEnv
from traffic_env.server.traffic_env import heuristic_policy

env = TrafficEnv(task_id="medium", seed=42)
obs = env.reset(seed=42, task_id="medium")

for _ in range(100):
    action = heuristic_policy(obs)
    obs = env.step(action)
    if obs.done:
        break

print("Final score:", obs.metadata.get("final_score"))

Option B: Local FastAPI server (no Docker)

cd traffic_env
uv sync
uv run uvicorn server.app:app --host 0.0.0.0 --port 8000

Open docs:

• http://127.0.0.1:8000/docs

Option C: OpenEnv script entrypoint

After editable install:

server

This calls the entrypoint from pyproject.toml:

• traffic_env.server.app:main

Inference Runner

inference.py supports four environment modes:

• local (default): direct TrafficEnv(...)
• server: connects to TRAFFIC_SERVER_URL
• docker: from_docker_image(IMAGE_NAME)
• openenv: from_env(OPENENV_REPO_ID, use_docker=...)

Minimal heuristic run (no LLM required)

# Linux/macOS
export TRAFFIC_ENV_MODE=local
export USE_LLM_POLICY=false
export TRAFFIC_TASK=medium
export MAX_STEPS=10
python inference.py

# Windows PowerShell
# $env:TRAFFIC_ENV_MODE='local'
# $env:USE_LLM_POLICY='false'
# $env:TRAFFIC_TASK='medium'
# $env:MAX_STEPS='10'
# python inference.py

LLM-backed run

# Linux/macOS
export TRAFFIC_ENV_MODE=local
export USE_LLM_POLICY=true
export HF_TOKEN=<token>
export API_BASE_URL=https://router.huggingface.co/v1
export MODEL_NAME=Qwen/Qwen2.5-72B-Instruct
python inference.py

STDOUT contract (produced by inference.py)

[START] task=<task_name> env=<benchmark> model=<model_name>
[STEP]  step=<n> action=<action_json> reward=<0.00> done=<true|false> error=<msg|null>
[END]   success=<true|false> steps=<n> score=<score> rewards=<r1,r2,...,rn>

RL API

• reset(seed=None, task_id=None) -> initial observation
• step(action) -> next observation (reward, done included)
• state property -> full serializable internal Markov state

Observation Space

Each observation includes:

• Lane-level:
  • density in [0, 1]
  • queue_length (int)
  • avg_wait_time (float)
  • vehicle_mix (bikes, cars, trucks, sum = 1)
  • obstruction (bool)
• Intersection-level:
  • current_phase: NS_GREEN | EW_GREEN | ALL_STOP
  • phase_duration_remaining
  • compliance_rate
  • time_of_day (0-24)
  • road_type: arterial | local | highway
• Global:
  • step_count
  • episode_reward
  • task_id
  • reward_breakdown

Action Space

IndiaUnregulatedIntersectionTrafficControllerAction supports:

• actions: List[IntersectionSignalAction]
  • intersection_id
  • phase
  • green_duration in [10, 120]
  • priority_override (bool)

For single-intersection control, a single action entry is sufficient. For multi-intersection control, provide one action per intersection each step.

Environment Dynamics

• Poisson vehicle arrivals per lane (task-dependent rates)
• Rush-hour multiplier: 2.5x
• Festival surge in hard task: 3x (time-windowed)
• Clearance multipliers:
  • bikes 1.4x
  • cars 1.0x
  • trucks 0.6x
• Compliance fluctuations impact illegal crossing/conflict risk
• Random obstruction events reduce service capacity
• Hard task supports overflow pressure propagation to neighboring nodes

Reward Function

raw_reward = (
    -0.40 * normalized_wait
    -0.30 * normalized_queue
    -0.20 * conflict_risk
    +0.10 * throughput
)
reward = clip(raw_reward + 1.0, 0.0, 1.0)

Reward is clipped to [0, 1] every step.

Tasks

• easy
  • 1 intersection, lanes N/S
  • no trucks, no obstructions
  • high compliance
• medium
  • 1 intersection, lanes N/S/E/W
  • trucks present (~15%)
  • rush-hour + obstructions + emergency window
• hard
  • 4 intersections
  • partial observability noise
  • overflow propagation
  • festival surge
  • lower compliance

Episode length: 100 steps.

Grading Utilities

• grade_task(policy, task_id, seed=7)
• grade_all_tasks(policy, seed=7)

Pass thresholds:

• Easy: >= 0.60
• Medium: >= 0.50
• Hard: >= 0.40

HTTP API Endpoints

When server is running (server.app:app):

• GET / -> redirects to /docs
• GET /favicon.ico -> 204 (no content)
• GET /docs -> Swagger UI
• GET /openapi.json
• GET /health
• GET /metadata
• GET /schema
• GET /state
• POST /reset
• POST /step
• WS /ws

Environment Variables

Inference (inference.py)

Server/OpenEnv metadata

Defined in openenv.yaml:

• app: server.app:app
• port: 8000
• runtime: fastapi

Docker

Build image:

docker build -t traffic_env:latest .

Run container:

docker run --rm -p 8000:8000 traffic_env:latest

Then open:

• http://127.0.0.1:8000/docs

OpenEnv Commands

openenv validate
openenv build

Troubleshooting

GET / returns 404

Current server behavior should redirect / to /docs. If you still see 404, ensure you are running the latest code.

GET /favicon.ico returns 404

Current server returns 204 for /favicon.ico. If you still get 404, restart server and clear old process.

NameError: clea is not defined

This was caused by stale image/code. Rebuild and rerun:

docker build --no-cache -t traffic_env:latest .
docker run --rm -p 8000:8000 traffic_env:latest

Inference falls back to heuristic unexpectedly

Check:

• USE_LLM_POLICY=true
• HF_TOKEN or API_KEY is set
• API_BASE_URL and MODEL_NAME are valid

Import errors with traffic_env

Install editable package:

pip install -e .

or use uv sync from project root.