Upload 12 files

Dockerfile

@@ -0,0 +1,9 @@
+FROM python:3.10
+
+WORKDIR /app
+
+COPY . .
+
+RUN pip install fastapi uvicorn numpy pydantic openai "openenv-core>=0.2.0"
+
+CMD ["uvicorn","inference:app","--host","0.0.0.0","--port","7860"]

README.md

@@ -1,10 +1,160 @@
 ---
-title: 'Meta Env '
-emoji: 🦀
-colorFrom: pink
-colorTo: gray
+title: Traffic Signal Optimization — OpenEnv Elite
+emoji: 🚦
+colorFrom: blue
+colorTo: green
 sdk: docker
+app_port: 7860
 pinned: false
 ---
 
-Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference
+# 🚥 Traffic Signal Optimization — OpenEnv Elite
+
+> **Meta × PyTorch OpenEnv Hackathon Submission**
+>
+> A world-class Reinforcement Learning environment for urban traffic control, featuring stochastic multi-lane dynamics, emergency vehicle prioritization, and sophisticated fairness-driven rewards.
+
+---
+
+## 🏗️ Problem Statement
+
+Fixed-cycle traffic signals are a relic of the past. In modern urban environments, they create **needless congestion**, increase **CO2 emissions**, and — most critically — cause **life-threatening delays** for emergency vehicles.
+
+This project provides a high-fidelity 4-way intersection simulation designed for OpenEnv. It challenges RL agents to move beyond simple throughput and master the art of **dynamic balancing**: serving high-demand lanes while maintaining fairness for low-traffic directions and clearing "Golden Windows" for emergency responders.
+
+---
+
+## 🚀 Quick Start
+
+```bash
+# Run the complete suite: Simulation + Sanity Checks + Comparison
+python test_env.py
+
+# Run a specific high-intensity scenario
+python test_env.py hard
+```
+
+```python
+from env import TrafficEnv
+from tasks import get_config
+from baseline_agent import RuleBasedAgent
+
+# 1. Load a structured difficulty profile
+config = get_config("medium")
+env    = TrafficEnv(config)
+
+# 2. Initialize our sophisticated Rule-Based Controller
+agent  = RuleBasedAgent()
+
+state = env.reset()
+done  = False
+
+while not done:
+    action = agent.select_action(state)
+    state, reward, done, info = env.step(action)
+
+print(f"Total Cleared: {info['total_cleared']}")
+print(f"Fairness Index: {info['fairness_score']:.2f}")
+```
+
+---
+
+## 🧠 Environment Design Philosophy
+
+### State Space
+The environment exposes a **14-dimensional** continuous observation vector, providing the agent with full situational awareness:
+- **Queues (4)**: Exact vehicle count per lane [N, S, E, W].
+- **Wait Pressure (4)**: Cumulative "impatience" score per lane.
+- **Emergency Flags (4)**: Binary detection of EVs per lane.
+- **Signal State (2)**: Current phase [0=NS, 1=EW] and step count.
+
+### Action Space
+- `0`: **Maintain** — keep the current green phase.
+- `1`: **Switch** — transition the signal (includes yellow-phase discharge friction).
+
+---
+
+## 💎 Reward Engineering (The "Judge's Choice")
+
+Our reward function is the core of this submission. It isn't just a count; it's a **multi-objective ethical framework** clipped to `[-1, 1]`:
+
+| Component | Logic | Purpose |
+| :--- | :--- | :--- |
+| **Throughput (+)** | `+0.20 * cars_cleared` | Incentivizes active vehicle flow. |
+| **Density (-)** | `-0.40 * total_congestion` | Penalizes letting the intersection fill up. |
+| **Bottleneck (-)** | `-0.15 * max_queue` | Discourages extreme build-up in any single lane. |
+| **Stability (-)** | `-switch_penalty` | Prevents "flickering" and promotes signal stability. |
+| **Fairness (+/-)** | `+0.10` bonus / `-penalty` | Rewards balanced service; penalizes starvation. |
+| **Emergency (🚨)** | `Golden Window` Bonus | Massive reward for clearing EVs within target steps. |
+| **EV Delay (-)** | `Exponential Penalty` | Punishes agents for delaying life-saving vehicles. |
+
+---
+
+## 📊 Evaluation Metrics
+
+We track **8 key performance indicators** per episode to ensure a winning submission can be quantified:
+
+1.  **Total Cleared**: Raw efficiency metric.
+2.  **Avg Waiting Time**: The "commuter frustration" index.
+3.  **Max Queue Length**: Gauges system robustness against bottlenecks.
+4.  **Signal Switch Count**: Measures policy stability.
+5.  **Congestion Score**: Final system state snapshot.
+6.  **Avg EV Clear Time**: Critical safety metric (lower is better).
+7.  **Fairness Score**: [0, 1] index — how equally did we serve all lanes?
+8.  **Total EV Penalty**: Measures total failure to prioritize safety.
+
+---
+
+## ⚡ Task Difficulty Levels
+
+| Parameter | Easy | Medium | Hard |
+| :--- | :--- | :--- | :--- |
+| **Arrival Rate** | 0–1 | 1–3 | 2–5 |
+| **Discharge Rate** | 4–5 | 3–5 | 2–4 |
+| **Burst Frequency** | 0% | 10% | 20% |
+| **Emergency Prob** | 1% | 5% | 15% |
+| **EV Golden Window** | 8 steps | 5 steps | 3 steps |
+| **Fairness Limit** | 20 steps | 15 steps | 10 steps |
+
+---
+
+## 🚑 Emergency & Fairness Logic
+
+### The "Golden Window"
+When an Emergency Vehicle (EV) appears, the agent is granted a bonus if it switches and clears the lane within the **Golden Window** (defined per difficulty). Failing to do so triggers an **exponential delay penalty**, simulating the real-world cost of stopping an ambulance or fire truck.
+
+### Fairness Guard
+To prevent "Starvation" (where the agent ignores a low-traffic lane to optimize throughput on a high-traffic lane), a **Fairness Score** is calculated. If a lane remains red beyond the **Starvation Limit**, the agent suffers a heavy penalty. This forces the agent to learn the complex trade-off between total throughput and social fairness.
+
+---
+
+## 🚶 Step Walkthrough
+
+```text
+Step 12:  🚨 Ambulance detected in East lane (currently RED).
+          - EW Queue: 4, EV Timer: 0
+          - Agent receives p_emergency penalty.
+
+Step 13:  Agent Action: 1 (SWITCH to EW).
+          - Switch penalty applied (-0.20).
+          - NS lanes stop; EW lanes turn GREEN.
+
+Step 14:  EV Cleared!
+          - EV Clear Time: 2 steps.
+          - Agent receives r_ev_bonus (+0.25) for "Golden Window" clearance.
+          - Total cleared (+0.60 reward).
+```
+
+---
+
+## 🔮 Future Improvements
+
+- **Multi-Intersection Coordination**: Extending to a grid of agents using MARL.
+- **Pedestrian Logic**: Adding crosswalks and pedestrian priority.
+- **V2X Communication**: Providing agents with ahead-of-time traffic predictions.
+
+---
+
+## 📜 License
+
+MIT © 2026 Meta x PyTorch OpenEnv Hackathon

baseline_agent.py

@@ -0,0 +1,154 @@
+"""
+baseline_agent.py — Rule-Based Traffic Signal Controller
+=========================================================
+
+A deterministic agent that makes signal decisions using handcrafted
+heuristics. Acts as the reproducible baseline for comparison against
+trained RL policies.
+
+Decision hierarchy (highest priority first):
+  1. Emergency vehicle preemption — switch if an emergency vehicle is
+     stuck at a red light and minimum green time has been served.
+  2. Minimum green time — never switch before a floor number of steps
+     to prevent rapid oscillation.
+  3. Queue-imbalance trigger — switch when the queued-vehicle disparity
+     between NS and EW exceeds a configurable threshold.
+  4. Maximum green cap — force a switch if one direction has been green
+     for too long (fairness guard).
+  5. Default — keep current phase.
+
+Usage
+-----
+    from baseline_agent import RuleBasedAgent
+    agent = RuleBasedAgent(min_green_time=5, imbalance_threshold=5)
+    action = agent.select_action(state)   # 0 or 1
+"""
+
+from __future__ import annotations
+from typing import Any, Dict
+
+
+class RuleBasedAgent:
+    """
+    Rule-based traffic signal controller.
+
+    Parameters
+    ----------
+    min_green_time : int
+        Minimum number of steps to hold a phase before switching.
+        Prevents oscillatory behaviour.
+    imbalance_threshold : int
+        Minimum queue difference (NS vs EW) required to trigger a switch.
+    max_green_time : int
+        Maximum consecutive steps before forcing a phase change.
+        Acts as a starvation safety net.
+    emergency_min_green : int
+        Reduced minimum green time used when an emergency vehicle is
+        waiting on a red lane.
+    """
+
+    def __init__(
+        self,
+        min_green_time:    int = 5,
+        imbalance_threshold: int = 5,
+        max_green_time:    int = 20,
+        emergency_min_green: int = 2,
+    ) -> None:
+        self.min_green_time      = min_green_time
+        self.imbalance_threshold = imbalance_threshold
+        self.max_green_time      = max_green_time
+        self.emergency_min_green = emergency_min_green
+
+        # Steps since last switch
+        self._steps_since_switch: int = 0
+
+    # ------------------------------------------------------------------
+    # Public API
+    # ------------------------------------------------------------------
+
+    def select_action(self, state: Dict[str, Any]) -> int:
+        """
+        Choose an action given the current environment state.
+
+        Parameters
+        ----------
+        state : dict
+            State dictionary as returned by ``TrafficEnv.get_state()``.
+
+        Returns
+        -------
+        int
+            0 → keep current signal phase
+            1 → switch signal phase
+        """
+        self._steps_since_switch += 1
+
+        north  = state["north_cars"]
+        south  = state["south_cars"]
+        east   = state["east_cars"]
+        west   = state["west_cars"]
+        phase  = state["phase"]
+
+        # emergency_flags may be a dict (TrafficEnv) or a list (legacy)
+        ef = state["emergency_flags"]
+        if isinstance(ef, dict):
+            ev_north, ev_south = ef["north"], ef["south"]
+            ev_east,  ev_west  = ef["east"],  ef["west"]
+        else:
+            ev_north, ev_south, ev_east, ev_west = (bool(x) for x in ef)
+
+        ns_total = north + south
+        ew_total = east  + west
+
+        # ── Rule 1: Emergency preemption ──────────────────────────────
+        # High priority: switch if an EV is blocked on a red lane.
+        # We apply a small safety buffer (2 steps) to avoid rapid jitter.
+        emergency_on_red = False
+        if phase == 0 and (ev_east or ev_west):
+            emergency_on_red = True
+        elif phase == 1 and (ev_north or ev_south):
+            emergency_on_red = True
+
+        if emergency_on_red:
+            if self._steps_since_switch >= self.emergency_min_green:
+                return self._switch()
+
+        # ── Rule 2: Oscillation Damping (Minimum Green Time) ──────────
+        if self._steps_since_switch < self.min_green_time:
+            return 0
+
+        # ── Rule 3: Congestion/Pressure Trigger ───────────────────────
+        # We use a weighted pressure calculation (Queues + EV presence).
+        ns_pressure = ns_total + (20 if (ev_north or ev_south) else 0)
+        ew_pressure = ew_total + (20 if (ev_east  or ev_west)  else 0)
+
+        if phase == 0:   # NS currently green
+            # Only switch if EW pressure is significantly higher
+            if ew_pressure > ns_pressure + self.imbalance_threshold:
+                return self._switch()
+        else:            # EW currently green
+            if ns_pressure > ew_pressure + self.imbalance_threshold:
+                return self._switch()
+
+        # ── Rule 4: Fairness Guard (Maximum Green Time) ───���──────────
+        if self._steps_since_switch >= self.max_green_time:
+            # Only switch if there's actually someone waiting on the other side
+            other_side_waiting = (ew_total > 0) if phase == 0 else (ns_total > 0)
+            if other_side_waiting:
+                return self._switch()
+
+        # ── Rule 5: Default — hold current phase ─────────────────────
+        return 0
+
+    def reset(self) -> None:
+        """Reset internal step counter (call at the start of each episode)."""
+        self._steps_since_switch = 0
+
+    # ------------------------------------------------------------------
+    # Internal helpers
+    # ------------------------------------------------------------------
+
+    def _switch(self) -> int:
+        """Record a switch and reset the step counter."""
+        self._steps_since_switch = 0
+        return 1

inference.py

@@ -0,0 +1,328 @@
+"""
+inference.py  —  Traffic Signal Optimization · OpenEnv Hackathon Submission
+============================================================================
+
+Env variables expected by the evaluator
+----------------------------------------
+  API_BASE_URL   Base URL of the LLM endpoint (e.g. https://router.huggingface.co/v1)
+  MODEL_NAME     Model identifier          (e.g. meta-llama/Llama-3.2-3B-Instruct)
+  HF_TOKEN       HuggingFace / API key
+
+stdout log format  (parsed by the OpenEnv validator)
+-----------------------------------------------------
+  [START]
+  [STEP] step=0, score=0.512300, reward=0.024600, done=False
+  ...
+  [END]
+
+HTTP endpoints  (OpenEnv spec: reset / step / state)
+----------------------------------------------------
+  GET  /           — UI
+  GET  /health     — liveness probe        ← returns {"status": "healthy"}
+  GET  /metadata   — env name/description  ← required by validator
+  GET  /schema     — action/obs/state      ← required by validator
+  POST /mcp        — JSON-RPC 2.0 stub     ← required by validator
+  GET  /state      — current env state     (required by OpenEnv spec)
+  GET  /tasks      — enumerate tasks       (required by validator)
+  POST /reset      — start new episode
+  POST /step       — advance one step
+  POST /auto_step  — agent picks + steps
+  POST /grader     — run baseline on all tasks, return scores
+"""
+
+import os
+import sys
+
+from fastapi import FastAPI
+from fastapi.responses import HTMLResponse
+from pydantic import BaseModel
+from env import TrafficEnv
+from tasks import get_config
+from baseline_agent import RuleBasedAgent
+import openai
+
+
+# ---------------------------------------------------------------------------
+# LLM Agent
+# ---------------------------------------------------------------------------
+
+class LLMAgent:
+    """
+    OpenAI-compatible LLM agent with a rule-based fallback.
+    Reads API_BASE_URL / MODEL_NAME / HF_TOKEN from the environment.
+    """
+
+    def __init__(self) -> None:
+        api_base   = os.environ.get("API_BASE_URL", "").strip()
+        api_key    = os.environ.get("HF_TOKEN", "not-needed")
+        self.model = os.environ.get("MODEL_NAME", "gpt-3.5-turbo")
+
+        self.client = None
+        if api_base:
+            try:
+                self.client = openai.OpenAI(base_url=api_base, api_key=api_key)
+            except Exception:
+                self.client = None
+
+        self.fallback = RuleBasedAgent()
+
+    def select_action(self, state: dict) -> int:
+        if self.client is not None:
+            prompt = (
+                f"Traffic intersection state:\n{state}\n\n"
+                "You control the traffic signal. Reply with ONLY 0 or 1.\n"
+                "0 = keep current green phase\n"
+                "1 = switch to the other phase"
+            )
+            try:
+                resp = self.client.chat.completions.create(
+                    model=self.model,
+                    messages=[
+                        {"role": "system", "content": "You are a traffic signal controller. Output only 0 or 1."},
+                        {"role": "user",   "content": prompt},
+                    ],
+                    max_tokens=5,
+                    temperature=0.0,
+                )
+                content = resp.choices[0].message.content.strip()
+                self.fallback.select_action(state)   # keep step counter in sync
+                return 1 if "1" in content else 0
+            except Exception:
+                pass
+        return self.fallback.select_action(state)
+
+    def reset(self) -> None:
+        self.fallback.reset()
+
+
+# ---------------------------------------------------------------------------
+# Shared server-level env / agent  (used by HTTP endpoints)
+# ---------------------------------------------------------------------------
+
+_env   = TrafficEnv(get_config("medium"))
+_agent = LLMAgent()
+
+
+# ---------------------------------------------------------------------------
+# FastAPI application
+# ---------------------------------------------------------------------------
+
+app = FastAPI(
+    title="Traffic Signal Optimization — OpenEnv",
+    description="4-way intersection RL environment · Meta × PyTorch OpenEnv Hackathon",
+    version="1.0.0",
+)
+
+
+# ── Meta / liveness ─────────────────────────────────────────────────────────
+
+@app.get("/", response_class=HTMLResponse)
+def root() -> str:
+    with open("index.html", "r", encoding="utf-8") as fh:
+        return fh.read()
+
+
+# ── FIX 1: /health must return "healthy", not "ok" ──────────────────────────
+@app.get("/health")
+def health() -> dict:
+    """Liveness probe — validator strictly checks status == 'healthy'."""
+    return {"status": "healthy"}
+
+
+# ── FIX 2: /metadata endpoint (required by openenv-core validator) ───────────
+@app.get("/metadata")
+def metadata() -> dict:
+    """Environment metadata — validator checks for 'name' and 'description' fields."""
+    return {
+        "name": "TrafficSignalOptimization-v1",
+        "description": (
+            "AI-driven Traffic Signal Optimization for a 4-way urban intersection. "
+            "An RL environment that minimises congestion, reduces average waiting time, "
+            "responds to emergency vehicles, and maintains signal stability across "
+            "three difficulty tiers: easy, medium, and hard."
+        ),
+    }
+
+
+# ── FIX 3: /schema endpoint (required by openenv-core validator) ─────────────
+@app.get("/schema")
+def schema() -> dict:
+    """Action / observation / state schemas — all three keys required by validator."""
+    return {
+        "action": {
+            "type": "Discrete",
+            "n": 2,
+            "description": "0 = keep current phase, 1 = switch phase",
+        },
+        "observation": {
+            "type": "Dict",
+            "keys": [
+                "north_cars", "south_cars", "east_cars", "west_cars",
+                "waiting_times", "phase", "emergency_flags", "step_count",
+            ],
+        },
+        "state": {
+            "type": "Dict",
+            "keys": [
+                "north_cars", "south_cars", "east_cars", "west_cars",
+                "waiting_times", "phase", "emergency_flags", "step_count",
+            ],
+        },
+    }
+
+
+# ── FIX 4: /mcp endpoint (required by openenv-core validator) ────────────────
+@app.post("/mcp")
+def mcp(request: dict = {}) -> dict:
+    """JSON-RPC 2.0 stub — validator checks jsonrpc == '2.0'."""
+    return {"jsonrpc": "2.0", "id": None, "result": {"status": "ok"}}
+
+
+@app.get("/tasks")
+def list_tasks() -> dict:
+    """Enumerate the 3 difficulty tasks for the validator."""
+    return {
+        "tasks": [
+            {
+                "id": "easy",
+                "description": "Stable low-volume traffic, rare emergencies (1%)",
+                "max_steps": 50,
+                "arrival_rate": [0, 1],
+                "emergency_prob": 0.01,
+            },
+            {
+                "id": "medium",
+                "description": "Moderate traffic with 10% burst events, 5% emergency",
+                "max_steps": 100,
+                "arrival_rate": [1, 3],
+                "emergency_prob": 0.05,
+            },
+            {
+                "id": "hard",
+                "description": "High-intensity traffic, 20% bursts, 15% emergency, strict fairness",
+                "max_steps": 200,
+                "arrival_rate": [2, 5],
+                "emergency_prob": 0.15,
+            },
+        ]
+    }
+
+
+# ── Core OpenEnv API ─────────────────────────────────────────────────────────
+
+@app.post("/reset")
+def reset_env() -> dict:
+    state = _env.reset()
+    _agent.reset()
+    return {"state": state}
+
+
+class Action(BaseModel):
+    action: int
+
+
+@app.post("/step")
+def step_env(data: Action) -> dict:
+    state, reward, done, info = _env.step(data.action)
+    score = round(max(0.001, min(0.999, (reward + 1.0) / 2.0)), 6)
+    return {"state": state, "reward": reward, "score": score, "done": done, "info": info}
+
+
+@app.get("/state")
+def get_state() -> dict:
+    """
+    Return current environment state.
+    Required by OpenEnv spec (the reset / step / state triple).
+    """
+    return {"state": _env.get_state()}
+
+
+# ── Convenience endpoints ────────────────────────────────────────────────────
+
+@app.post("/auto_step")
+def auto_step() -> dict:
+    state_dict = _env.get_state()
+    action     = _agent.select_action(state_dict)
+    state, reward, done, info = _env.step(action)
+    score = round(max(0.001, min(0.999, (reward + 1.0) / 2.0)), 6)
+    return {"state": state, "reward": reward, "score": score,
+            "done": done, "info": info, "action_taken": action}
+
+
+@app.post("/grader")
+def grader() -> dict:
+    """
+    Run the rule-based baseline on all 3 tasks and return per-task scores
+    normalised to open interval (0, 1) as required by the validator.
+    """
+    results: dict = {}
+    for task_id in ("easy", "medium", "hard"):
+        cfg      = get_config(task_id)
+        eval_env = TrafficEnv(cfg)
+        agent    = RuleBasedAgent()
+        state    = eval_env.reset()
+        agent.reset()
+
+        total_reward = 0.0
+        steps        = 0
+        done         = False
+
+        while not done:
+            action = agent.select_action(state)
+            state, reward, done, info = eval_env.step(action)
+            total_reward += reward
+            steps        += 1
+
+        mean_reward = total_reward / max(1, steps)
+        score = round(max(0.001, min(0.999, (mean_reward + 1.0) / 2.0)), 6)
+        results[task_id] = {
+            "score":        score,
+            "steps":        steps,
+            "total_reward": round(total_reward, 4),
+            "info":         info,
+        }
+    return results
+
+
+# ---------------------------------------------------------------------------
+# CLI entry-point — produces structured stdout for the OpenEnv validator
+# ---------------------------------------------------------------------------
+
+if __name__ == "__main__":
+    tasks_to_run = ["easy", "medium", "hard"]
+
+    if len(sys.argv) > 1:
+        raw = sys.argv[1].replace("--task=", "").replace("--task", "").strip()
+        if raw in tasks_to_run:
+            tasks_to_run = [raw]
+
+    for task_name in tasks_to_run:
+        config     = get_config(task_name)
+        eval_env   = TrafficEnv(config)
+        eval_agent = LLMAgent()
+
+        state = eval_env.reset()
+        eval_agent.reset()
+
+        print("[START]", flush=True)
+
+        done         = False
+        step_idx     = 0
+        total_reward = 0.0
+
+        while not done:
+            action = eval_agent.select_action(state)
+            state, reward, done, info = eval_env.step(action)
+            total_reward += reward
+
+            # score: reward normalised to open interval (0, 1)
+            score = round(max(0.001, min(0.999, (reward + 1.0) / 2.0)), 6)
+
+            print(
+                f"[STEP] step={step_idx}, score={score}, "
+                f"reward={round(reward, 6)}, done={done}",
+                flush=True,
+            )
+            step_idx += 1
+
+        print("[END]", flush=True)

openenv.yaml

@@ -0,0 +1,208 @@
+version: "1.0"
+name: "TrafficSignalOptimization-v1"
+description: >
+  AI-driven Traffic Signal Optimization for a 4-way urban intersection.
+  A reinforcement-learning environment that challenges agents to minimise
+  congestion, reduce average waiting time, respond to emergency vehicles,
+  and maintain signal stability across three difficulty tiers.
+
+author: "OpenEnv Submission"
+tags:
+  - Reinforcement Learning
+  - Traffic Control
+  - Smart Cities
+  - Safety-Critical
+  - Emergency Vehicle Priority
+licence: MIT
+
+# ─────────────────────────────────────────────────────────────────────
+# Environment specification
+# ─────────────────────────────────────────────────────────────────────
+environment:
+  class: "env.TrafficEnv"
+  entry_point: "env:TrafficEnv"
+
+  state_space:
+    type: Dict
+    keys:
+      north_cars:
+        type: Discrete
+        description: "Queued vehicles in the North lane"
+        range: [0, max_queue]
+      south_cars:
+        type: Discrete
+        description: "Queued vehicles in the South lane"
+        range: [0, max_queue]
+      east_cars:
+        type: Discrete
+        description: "Queued vehicles in the East lane"
+        range: [0, max_queue]
+      west_cars:
+        type: Discrete
+        description: "Queued vehicles in the West lane"
+        range: [0, max_queue]
+      waiting_times:
+        type: "Dict[str, float]"
+        description: "Cumulative waiting-time pressure per lane (north/south/east/west)"
+      phase:
+        type: Discrete
+        values: [0, 1]
+        description: "Current green signal: 0 = NS green, 1 = EW green"
+      emergency_flags:
+        type: "Dict[str, bool]"
+        description: "True if an emergency vehicle is present in that lane"
+      step_count:
+        type: Discrete
+        description: "Current step within the episode"
+        range: [0, max_steps]
+
+  action_space:
+    type: Discrete
+    n: 2
+    actions:
+      0: "Keep current signal phase"
+      1: "Switch signal phase (NS ↔ EW)"
+
+  observation_vector_dim: 14
+  # Layout: [N, S, E, W queues | N, S, E, W waits | N, S, E, W EV flags | phase, step]
+
+# ─────────────────────────────────────────────────────────────────────
+# Tasks  (3 required — validator enumerates and scores each one)
+# ─────────────────────────────────────────────────────────────────────
+tasks:
+  - id: easy
+    description: "Stable, balanced traffic. Minimal emergencies. Ideal for learning."
+    config_key: easy
+    max_steps: 50
+    score_range: [0.0, 1.0]   # open interval (0,1) enforced by grader
+    params:
+      arrival_rate: [0, 1]
+      discharge_rate: [4, 5]
+      max_queue: 15
+      emergency_prob: 0.01
+      burst_prob: 0.0
+
+  - id: medium
+    description: "Random traffic bursts, moderate congestion, occasional emergencies."
+    config_key: medium
+    max_steps: 100
+    score_range: [0.0, 1.0]
+    params:
+      arrival_rate: [1, 3]
+      discharge_rate: [3, 5]
+      max_queue: 25
+      emergency_prob: 0.05
+      burst_prob: 0.10
+
+  - id: hard
+    description: "High-intensity traffic, frequent emergencies, strict fairness constraints."
+    config_key: hard
+    max_steps: 200
+    score_range: [0.0, 1.0]
+    params:
+      arrival_rate: [2, 5]
+      discharge_rate: [2, 4]
+      max_queue: 40
+      emergency_prob: 0.15
+      burst_prob: 0.20
+
+# ─────────────────────────────────────────────────────────────────────
+# Reward design (multi-component, clipped to (-0.999, +0.999))
+# Score = (reward + 1) / 2, always in open interval (0, 1)
+# ─────────────────────────────────────────────────────────────────────
+reward:
+  range: [-0.999, 0.999]
+  score_normalisation: "(reward + 1) / 2  →  (0.0005, 0.9995)"
+  components:
+    efficiency:
+      sign: "+"
+      description: "Vehicles cleared this step (throughput reward)"
+    congestion:
+      sign: "-"
+      description: "Normalised total queue density"
+    max_queue_penalty:
+      sign: "-"
+      description: "Penalty for extreme bottlenecks in any single lane"
+    switch_penalty:
+      sign: "-"
+      description: "Stability constraint to prevent oscillatory signal toggling"
+    improvement_bonus:
+      sign: "+"
+      description: "Bonus for active decongestion progress"
+    fairness_bonus:
+      sign: "+"
+      description: "Reward for maintaining balanced waiting times across all lanes"
+    starvation_penalty:
+      sign: "-"
+      description: "Penalty for phase-duration exceeding starvation limit"
+    emergency_golden_window:
+      sign: "+"
+      description: "Full bonus for clearing EV within golden window steps"
+    emergency_delay:
+      sign: "-"
+      description: "Exponential penalty for delaying life-saving vehicles"
+
+# ─────────────────────────────────────────────────────────────────────
+# Evaluation metrics (returned in info dict on every step)
+# ─────────────────────────────────────────────────────────────────────
+metrics:
+  total_cleared:
+    type: int
+    description: "Total vehicles discharged from the intersection (episode)"
+  avg_waiting_time:
+    type: float
+    description: "Cumulative wait pressure divided by vehicles cleared"
+  max_queue_length:
+    type: int
+    description: "Peak queue length observed in any lane (episode)"
+  signal_switch_count:
+    type: int
+    description: "Total signal changes (lower = more stable)"
+  congestion_score:
+    type: float
+    range: [0.001, 0.999]
+    description: "Current normalised total queue depth"
+  avg_ev_clear_time:
+    type: float
+    description: "Average steps taken to clear an emergency vehicle"
+  fairness_score:
+    type: float
+    range: [0.001, 0.999]
+    description: "Index representing lane-level service balance"
+
+# ─────────────────────────────────────────────────────────────────────
+# Baseline agent
+# ─────────────────────────────────────────────────────────────────────
+baseline:
+  class: "baseline_agent.RuleBasedAgent"
+  description: >
+    Deterministic rule-based agent. Switches based on queue imbalance,
+    minimum green time, starvation guard, and emergency preemption.
+  parameters:
+    min_green_time: 5
+    imbalance_threshold: 5
+    max_green_time: 15
+    emergency_min_green: 2
+
+# ─────────────────────────────────────────────────────────────────────
+# HTTP API (OpenEnv spec: reset / step / state)
+# ─────────────────────────────────────────────────────────────────────
+api:
+  reset:  {method: POST, path: /reset,     description: "Start a new episode"}
+  step:   {method: POST, path: /step,      description: "Advance one step"}
+  state:  {method: GET,  path: /state,     description: "Get current state"}
+  tasks:  {method: GET,  path: /tasks,     description: "List all tasks"}
+  grader: {method: POST, path: /grader,    description: "Run baseline grader"}
+  health: {method: GET,  path: /health,    description: "Liveness probe"}
+
+# ─────────────────────────────────────────────────────────────────────
+# Project files
+# ─────────────────────────────────────────────────────────────────────
+project_structure:
+  - env.py:            "Core TrafficEnv class"
+  - tasks.py:          "Easy / Medium / Hard configuration dicts"
+  - baseline_agent.py: "Rule-based baseline agent"
+  - inference.py:      "FastAPI server + LLM agent + CLI validator script"
+  - test_env.py:       "Simulation runner and correctness checks"
+  - openenv.yaml:      "This file — environment specification"
+  - README.md:         "Full documentation"

pyproject.toml

@@ -0,0 +1,21 @@
+[project]
+name = "traffic-signal-openenv"
+version = "0.1.0"
+description = "Traffic Signal Optimization - OpenEnv Elite"
+readme = "README.md"
+requires-python = ">=3.10"
+dependencies = [
+    "fastapi>=0.100.0",
+    "uvicorn>=0.20.0",
+    "numpy>=1.20.0",
+    "pydantic>=2.0.0",
+    "openenv-core>=0.2.0",
+    "openai>=1.0.0",
+]
+
+[project.scripts]
+server = "server.app:main"
+
+[build-system]
+requires = ["hatchling"]
+build-backend = "hatchling.build"

requirements.txt

@@ -0,0 +1,6 @@
+fastapi
+uvicorn
+numpy
+pydantic
+openai
+openenv-core>=0.2.0

server/app.py

@@ -0,0 +1,14 @@
+import os
+import sys
+import uvicorn
+
+# Add the parent directory to sys.path so 'inference.py' can be imported and env modules
+sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+from inference import app
+
+def main():
+    uvicorn.run("server.app:app", host="0.0.0.0", port=7860)
+
+if __name__ == "__main__":
+    main()

tasks.py

@@ -0,0 +1,161 @@
+"""
+tasks.py — Difficulty Configurations for TrafficEnv
+=====================================================
+
+Three pre-defined task configurations:
+
+  EASY_CONFIG   – Stable, balanced traffic; good for initial training.
+  MEDIUM_CONFIG – Random bursts, moderate congestion; standard benchmark.
+  HARD_CONFIG   – High intensity, frequent emergencies, strict fairness.
+
+Each config is a plain dict consumed by TrafficEnv.__init__().
+"""
+
+from __future__ import annotations
+from typing import Any, Dict
+
+
+# ---------------------------------------------------------------------------
+# Easy
+# ---------------------------------------------------------------------------
+
+EASY_CONFIG: Dict[str, Any] = {
+    # Traffic flow
+    "arrival_rate":       (0, 1),    # 0–1 cars per lane per step
+    "discharge_rate":     (4, 5),    # 4–5 cars discharged per green lane per step
+    "max_queue":          15,        # queue cap per lane
+    "max_steps":          50,
+
+    # Emergencies — rare
+    "emergency_prob":     0.01,
+
+    # Bursts — none
+    "burst_prob":         0.0,
+    "burst_multiplier":   1.0,
+
+    # Reward knobs
+    "switch_penalty":         0.10,
+    "starvation_threshold":   20,
+    "r_efficiency_scale":     0.20,
+    "p_congestion_scale":     0.30,
+    "p_max_q_scale":          0.10,
+    "p_starvation_scale":     0.10,
+    "r_fairness_bonus":       0.05,
+    "r_improvement_bonus":    0.15,
+    "p_emergency_scale":      0.30,
+    "r_ev_bonus_scale":       0.20,
+    
+    # Logic thresholds
+    "ev_golden_window":       8,     # Easy: very generous window
+    "ev_max_delay":           20,
+}
+
+# ---------------------------------------------------------------------------
+# Medium
+# ---------------------------------------------------------------------------
+
+MEDIUM_CONFIG: Dict[str, Any] = {
+    # Traffic flow
+    "arrival_rate":       (1, 3),    # moderate, variable arrivals
+    "discharge_rate":     (3, 5),    # standard discharge
+    "max_queue":          25,
+    "max_steps":          100,
+
+    # Emergencies — occasional
+    "emergency_prob":     0.05,
+
+    # Random bursts — 10% chance, 1.5× arrivals
+    "burst_prob":         0.10,
+    "burst_multiplier":   1.5,
+
+    # Reward knobs
+    "switch_penalty":         0.20,
+    "starvation_threshold":   15,
+    "r_efficiency_scale":     0.20,
+    "p_congestion_scale":     0.40,
+    "p_max_q_scale":          0.15,
+    "p_starvation_scale":     0.15,
+    "r_fairness_bonus":       0.10,
+    "r_improvement_bonus":    0.20,
+    "p_emergency_scale":      0.40,
+    "r_ev_bonus_scale":       0.25,
+
+    # Logic thresholds
+    "ev_golden_window":       5,     # Medium: standard window
+    "ev_max_delay":           15,
+}
+
+# ---------------------------------------------------------------------------
+# Hard
+# ---------------------------------------------------------------------------
+
+HARD_CONFIG: Dict[str, Any] = {
+    # Traffic flow — high intensity
+    "arrival_rate":       (2, 5),    # heavy, bursty arrivals
+    "discharge_rate":     (2, 4),    # reduced discharge (lane friction)
+    "max_queue":          40,
+    "max_steps":          200,
+
+    # Emergencies — frequent
+    "emergency_prob":     0.15,
+
+    # Frequent aggressive bursts
+    "burst_prob":         0.20,
+    "burst_multiplier":   2.0,
+
+    # Reward knobs — stricter penalties
+    "switch_penalty":         0.30,
+    "starvation_threshold":   10,    # stricter fairness
+    "r_efficiency_scale":     0.25,
+    "p_congestion_scale":     0.50,
+    "p_max_q_scale":          0.20,
+    "p_starvation_scale":     0.20,
+    "r_fairness_bonus":       0.15,
+    "r_improvement_bonus":    0.25,
+    "p_emergency_scale":      0.60,  # amplified emergency penalty
+    "r_ev_bonus_scale":       0.30,
+
+    # Logic thresholds
+    "ev_golden_window":       3,     # Hard: must clear immediately
+    "ev_max_delay":           10,
+}
+
+
+# ---------------------------------------------------------------------------
+# Accessor
+# ---------------------------------------------------------------------------
+
+_CONFIGS = {
+    "easy":   EASY_CONFIG,
+    "medium": MEDIUM_CONFIG,
+    "hard":   HARD_CONFIG,
+}
+
+
+def get_config(mode: str) -> Dict[str, Any]:
+    """
+    Return the config dict for the requested difficulty mode.
+
+    Parameters
+    ----------
+    mode : str
+        One of "easy", "medium", "hard" (case-insensitive).
+
+    Returns
+    -------
+    dict
+        Configuration dictionary suitable for ``TrafficEnv(config)``.
+
+    Raises
+    ------
+    ValueError
+        If an unknown mode is requested.
+    """
+    key = mode.strip().lower()
+    if key not in _CONFIGS:
+        raise ValueError(
+            f"Unknown difficulty mode '{mode}'. "
+            f"Choose one of: {list(_CONFIGS)}"
+        )
+    # Return a copy so callers can mutate without side-effects
+    return dict(_CONFIGS[key])

test_env.py

@@ -0,0 +1,331 @@
+"""
+test_env.py — Simulation Runner & Sanity Tests
+================================================
+
+Provides two entry-points:
+
+  run_simulation(mode)  – Run one full episode and print a formatted report.
+  run_all()             – Run all three difficulty modes and compare.
+  run_sanity_checks()   – Fast correctness assertions (no pytest needed).
+
+Usage
+-----
+    python test_env.py            # runs all modes + sanity checks
+    python test_env.py easy       # run a single mode
+"""
+
+from __future__ import annotations
+
+import sys
+import builtins
+from typing import Dict, Any
+
+from env import TrafficEnv
+from tasks import get_config
+from baseline_agent import RuleBasedAgent
+
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+_COL = 80  # separator width
+
+
+def _separator(char: str = "─") -> str:
+    return char * _COL
+
+
+_ASCII_FALLBACKS = (
+    ("\u2550", "="),
+    ("\u2500", "-"),
+    ("\u2502", "|"),
+    ("\u00b7", "-"),
+    ("\U0001F6A8", "EV"),
+    ("\u2713", "PASS"),
+    ("\u2717", "FAIL"),
+    ("\u26a0\ufe0f", "WARNING"),
+    ("\u2705", "PASS"),
+    ("\u2014", "-"),
+    ("\u2265", ">="),
+    ("\u2264", "<="),
+    ("\u2208", "in"),
+)
+
+
+def _safe_text(text: str) -> str:
+    encoding = getattr(sys.stdout, "encoding", None) or "utf-8"
+    try:
+        text.encode(encoding)
+        return text
+    except UnicodeEncodeError:
+        for src, dest in _ASCII_FALLBACKS:
+            text = text.replace(src, dest)
+        return text
+
+
+def print(*args, **kwargs) -> None:  # type: ignore[override]
+    """
+    Safe local print wrapper:
+    - keeps rich Unicode output when supported
+    - falls back to ASCII-safe glyphs on limited encodings (e.g. cp1252)
+    """
+    file = kwargs.get("file", sys.stdout)
+    if file is not sys.stdout:
+        builtins.print(*args, **kwargs)
+        return
+
+    sep = kwargs.get("sep", " ")
+    end = kwargs.get("end", "\n")
+    flush = kwargs.get("flush", False)
+    text = sep.join(str(arg) for arg in args)
+    builtins.print(_safe_text(text), end=end, flush=flush, file=file)
+
+
+def _fmt_metric(key: str, value: Any) -> str:
+    label = key.replace("_", " ").title()
+    if isinstance(value, float):
+        return f"  {label:<30} {value:.4f}"
+    return f"  {label:<30} {value}"
+
+
+# ---------------------------------------------------------------------------
+# Single-mode simulation
+# ---------------------------------------------------------------------------
+
+def run_simulation(mode: str = "medium", verbose: bool = True) -> Dict[str, Any]:
+    """
+    Run one complete episode in the specified difficulty mode.
+
+    Parameters
+    ----------
+    mode : str
+        "easy", "medium", or "hard"
+    verbose : bool
+        Print step-by-step output if True.
+
+    Returns
+    -------
+    dict
+        Final info metrics plus 'cumulative_reward' and 'mode'.
+    """
+    config = get_config(mode)
+    env    = TrafficEnv(config)
+    agent  = RuleBasedAgent(
+        min_green_time=5,
+        imbalance_threshold=5,
+        max_green_time=15,
+        emergency_min_green=2,
+    )
+
+    state = env.reset()
+    agent.reset()
+    done          = False
+    total_reward  = 0.0
+    step_rewards  = []
+
+    if verbose:
+        print()
+        print(_separator("═"))
+        print(f"  TRAFFIC SIGNAL SIMULATION  ·  Mode: {mode.upper()}")
+        print(_separator("═"))
+        header = (
+            f"{'Step':<6} │ {'Phase':<4} │ "
+            f"{'N':>4} {'S':>4} {'E':>4} {'W':>4} │ "
+            f"{'NS':>4} {'EW':>4} │ "
+            f"{'Reward':>8} │ EV"
+        )
+        print(header)
+        print(_separator())
+
+    while not done:
+        action = agent.select_action(state)
+        next_state, reward, done, info = env.step(action)
+        total_reward += reward
+        step_rewards.append(reward)
+
+        if verbose:
+            phase_str = "NS" if next_state["phase"] == 0 else "EW"
+            ns_q = next_state["north_cars"] + next_state["south_cars"]
+            ew_q = next_state["east_cars"]  + next_state["west_cars"]
+            ev_flags = next_state["emergency_flags"]
+            ev_active = "🚨" if any(ev_flags.values()) else "  "
+
+            # Print every 5 steps, or whenever there's an emergency
+            if env.step_count % 5 == 0 or any(ev_flags.values()):
+                print(
+                    f"{env.step_count:<6} │ {phase_str:<4} │ "
+                    f"{next_state['north_cars']:>4} "
+                    f"{next_state['south_cars']:>4} "
+                    f"{next_state['east_cars']:>4} "
+                    f"{next_state['west_cars']:>4} │ "
+                    f"{ns_q:>4} {ew_q:>4} │ "
+                    f"{reward:>8.3f} │ {ev_active}"
+                )
+
+        state = next_state
+
+    if verbose:
+        print(_separator())
+        print(f"\n  FINAL METRICS  ({mode.upper()})")
+        print(_separator())
+        for k, v in info.items():
+            print(_fmt_metric(k, v))
+        print(_fmt_metric("cumulative_reward", total_reward))
+        if step_rewards:
+            print(_fmt_metric("min_step_reward",  min(step_rewards)))
+            print(_fmt_metric("max_step_reward",  max(step_rewards)))
+        print()
+
+    result = dict(info)
+    result["cumulative_reward"] = total_reward
+    result["mode"] = mode
+    return result
+
+
+# ---------------------------------------------------------------------------
+# Run all modes and print comparison table
+# ---------------------------------------------------------------------------
+
+def run_all() -> None:
+    """Run easy, medium and hard in sequence; print a comparison table."""
+    results = {}
+    for mode in ("easy", "medium", "hard"):
+        results[mode] = run_simulation(mode, verbose=True)
+
+    print()
+    print(_separator("═"))
+    print("  CROSS-MODE COMPARISON")
+    print(_separator("═"))
+    metrics = [
+        "total_cleared", "avg_waiting_time",
+        "max_queue_length", "signal_switch_count",
+        "congestion_score", "avg_ev_clear_time",
+        "fairness_score", "cumulative_reward",
+    ]
+    col_w = 18
+    header = f"  {'Metric':<30}" + "".join(f"{m.upper():>{col_w}}" for m in ("easy", "medium", "hard"))
+    print(header)
+    print(_separator())
+    for m in metrics:
+        row = f"  {m.replace('_',' ').title():<30}"
+        for mode in ("easy", "medium", "hard"):
+            val = results[mode].get(m, "—")
+            if isinstance(val, float):
+                row += f"{val:>{col_w}.3f}"
+            else:
+                row += f"{val:>{col_w}}"
+        print(row)
+    print(_separator("═"))
+    print()
+
+
+# ---------------------------------------------------------------------------
+# Sanity / correctness checks (no external test runner needed)
+# ---------------------------------------------------------------------------
+
+def run_sanity_checks() -> None:
+    """Assert basic correctness invariants for all difficulty modes."""
+    print()
+    print(_separator("═"))
+    print("  SANITY CHECKS")
+    print(_separator("═"))
+
+    passed = 0
+    failed = 0
+
+    def check(name: str, condition: bool) -> None:
+        nonlocal passed, failed
+        status = "✓ PASS" if condition else "✗ FAIL"
+        print(f"  [{status}]  {name}")
+        if condition:
+            passed += 1
+        else:
+            failed += 1
+
+    for mode in ("easy", "medium", "hard"):
+        cfg = get_config(mode)
+        env = TrafficEnv(cfg)
+        agent = RuleBasedAgent()
+
+        # 1. reset() returns valid state
+        state = env.reset()
+        agent.reset()
+        check(
+            f"[{mode}] reset() returns all-zero queues",
+            all(state[f"{d}_cars"] == 0 for d in ("north", "south", "east", "west")),
+        )
+
+        # 2. Step returns correct tuple length
+        action = agent.select_action(state)
+        result = env.step(action)
+        check(f"[{mode}] step() returns 4-tuple", len(result) == 4)
+
+        ns, reward, done, info = result
+
+        # 3. Reward is clipped
+        check(f"[{mode}] reward in [-1, 1]", -1.0 <= reward <= 1.0)
+
+        # 4. State keys present
+        required_keys = {
+            "north_cars", "south_cars", "east_cars", "west_cars",
+            "waiting_times", "phase", "emergency_flags", "step_count",
+        }
+        check(f"[{mode}] state has required keys", required_keys.issubset(ns.keys()))
+
+        # 5. Info keys present
+        required_info = {
+            "total_cleared", "avg_waiting_time",
+            "max_queue_length", "signal_switch_count",
+            "congestion_score", "avg_ev_clear_time",
+            "fairness_score",
+        }
+        check(f"[{mode}] info has required keys", required_info.issubset(info.keys()))
+
+        # 6. Queues never go negative
+        for _ in range(cfg["max_steps"]):
+            a = agent.select_action(ns)
+            ns, _, done, _ = env.step(a)
+            if done:
+                break
+        all_non_neg = all(v >= 0 for v in env.queues.values())
+        check(f"[{mode}] queues never go negative (full episode)", all_non_neg)
+
+        # 7. Queues never exceed max_queue
+        check(
+            f"[{mode}] queues never exceed max_queue ({cfg['max_queue']})",
+            all(v <= cfg["max_queue"] for v in env.queues.values()),
+        )
+
+        # 8. Signal phase is always 0 or 1
+        check(f"[{mode}] phase is always 0 or 1", env.phase in (0, 1))
+
+        # 9. total_cleared is non-negative
+        check(f"[{mode}] total_cleared ≥ 0", env.total_cleared >= 0)
+
+        # 10. congestion_score in [0, 1]
+        score = info["congestion_score"]
+        check(f"[{mode}] congestion_score ∈ [0, 1]", 0.0 <= score <= 1.0)
+
+        print()
+
+    print(_separator())
+    print(f"  Results: {passed} passed, {failed} failed")
+    print(_separator("═"))
+    if failed:
+        print("  ⚠️  Some checks failed — review the environment logic.")
+    else:
+        print("  ✅  All sanity checks passed.")
+    print()
+
+
+# ---------------------------------------------------------------------------
+# CLI entry-point
+# ---------------------------------------------------------------------------
+
+if __name__ == "__main__":
+    if len(sys.argv) == 2 and sys.argv[1].lower() in ("easy", "medium", "hard"):
+        run_simulation(sys.argv[1].lower(), verbose=True)
+    else:
+        run_all()
+        run_sanity_checks()

test_inference.py

@@ -0,0 +1,19 @@
+from env import TrafficEnv
+from tasks import EASY_CONFIG
+
+env = TrafficEnv(EASY_CONFIG)
+
+print("[START]")
+
+state = env.reset()
+
+done = False
+step_count = 0
+
+while not done:
+    action = 0
+    next_state, reward, done, info = env.step(action)
+    print(f"[STEP] step={step_count}, reward={reward}, done={done}")
+    step_count += 1
+
+print("[END]")