"""OpenRA-Bench: Agent Leaderboard for OpenRA-RL. A Gradio app that displays agent rankings, supports filtering by type and opponent difficulty, and lets users run evaluations in-browser. Run locally: python app.py Deploy on HuggingFace Spaces: Push app.py, requirements.txt, data/, and README.md to your HF Space. """ import csv import html import json import logging import os import re import time from collections import defaultdict from datetime import datetime, timezone from pathlib import Path import gradio as gr import gradio_client.utils as _gc_utils import pandas as pd _orig_json_schema_to_python_type = _gc_utils._json_schema_to_python_type def _patched_json_schema_to_python_type(schema, defs=None): if not isinstance(schema, dict): return "Any" return _orig_json_schema_to_python_type(schema, defs) _gc_utils._json_schema_to_python_type = _patched_json_schema_to_python_type from evaluate_runner import DIFFICULTY_MULTIPLIER, DEFAULT_SERVER, compute_composite_score, compute_game_metrics logger = logging.getLogger(__name__) # ── Data Loading ────────────────────────────────────────────────────────────── DATA_PATH = Path(__file__).parent / "data" / "results.csv" AGENT_TYPE_COLORS = { "Scripted": "#ffcd75", # Gold "LLM": "#7497db", # Blue "RL": "#75809c", # Gray-blue } DISPLAY_COLUMNS = [ "Rank", "Agent", "Type", "Status", "Opponent", "Games", "Win Rate (%)", "Score", "K/D Ratio", "Avg Kills", "Avg Deaths", "Avg Economy", "Avg Game Length", "Date", "Replay", ] def _safe_agent_link(name: str, url) -> str: """Render agent name, optionally as a hyperlink. HTML-escaped to prevent XSS.""" safe_name = html.escape(str(name)) if pd.notna(url) and str(url).strip(): url_str = str(url).strip() # Only allow http/https URLs — block javascript:, data:, etc. if url_str.startswith(("http://", "https://")): safe_url = html.escape(url_str, quote=True) return f'{safe_name}' return safe_name def _verified_badge(verified) -> str: """Render a Verified/Unverified HTML badge.""" if isinstance(verified, str): verified = verified.lower() in ("true", "1", "yes") if verified: return ( '' 'Verified' ) return ( '' 'Unverified' ) def _safe_replay_link(url) -> str: """Render replay download link. Filename is sanitized to prevent XSS.""" if pd.notna(url) and str(url).strip(): # Sanitize: only allow alphanumeric, dash, underscore, dot safe_name = re.sub(r"[^a-zA-Z0-9._-]", "", str(url).strip()) if safe_name: escaped = html.escape(safe_name, quote=True) return f'⬇' return "" def load_data() -> pd.DataFrame: """Load leaderboard data from CSV.""" if not DATA_PATH.exists(): return pd.DataFrame(columns=DISPLAY_COLUMNS) df = pd.read_csv(DATA_PATH) df = df.sort_values("score", ascending=False).reset_index(drop=True) df.insert(0, "Rank", range(1, len(df) + 1)) # Build agent name with optional hyperlink (XSS-safe) if "agent_url" in df.columns: df["Agent"] = df.apply( lambda r: _safe_agent_link(r.get("agent_name", ""), r.get("agent_url", "")), axis=1, ) else: df["Agent"] = df["agent_name"].apply(lambda n: html.escape(str(n))) # Build replay download link (XSS-safe) if "replay_url" in df.columns: df["Replay"] = df["replay_url"].apply(_safe_replay_link) else: df["Replay"] = "" # Verified/Unverified badge if "verified" in df.columns: df["Status"] = df["verified"].apply(_verified_badge) else: df["Status"] = _verified_badge(True) # Legacy data = verified # Rename for display df = df.rename(columns={ "agent_type": "Type", "opponent": "Opponent", "games": "Games", "win_rate": "Win Rate (%)", "score": "Score", "kd_ratio": "K/D Ratio", "avg_kills": "Avg Kills", "avg_deaths": "Avg Deaths", "avg_economy": "Avg Economy", "avg_game_length": "Avg Game Length", "timestamp": "Date", }) return df[DISPLAY_COLUMNS] def add_type_badges(df: pd.DataFrame) -> pd.DataFrame: """Add color-coded HTML badges to the Type column.""" def badge(agent_type: str) -> str: color = AGENT_TYPE_COLORS.get(agent_type, "#ccc") text_color = "#fff" if agent_type != "Scripted" else "#333" return ( f'' f"{agent_type}" ) df = df.copy() df["Type"] = df["Type"].apply(badge) return df def load_capability_leaderboard() -> pd.DataFrame: """Ranked capability leaderboard from the run_eval JSONL store (composite + Perception/Reasoning/Action + dominant weakest link).""" try: from openra_bench.leaderboard import build_table rows = build_table() except Exception: # noqa: BLE001 — never break the UI on a bad store rows = [] cols = [ "rank", "model", "episodes", "win_rate", "composite", "objective", "adversarial_rating", "perception", "reasoning", "action", "weakest_link", "reward_vector", "held_out_composite", "generalization_gap", ] if not rows: return pd.DataFrame(columns=cols) def _rv(v) -> str: if not isinstance(v, dict) or not v: return "" return " ".join(f"{k[:3]}={float(val):.2f}" for k, val in v.items()) out = [] for r in rows: row = {c: r.get(c) for c in cols} row["reward_vector"] = _rv(r.get("reward_vector")) out.append(row) return pd.DataFrame(out) # ── Battle Viewer ───────────────────────────────────────────────────────────── # Single-command playback browser: filter run → model → scenario, step # the battle turn-by-turn, and compare two models head-to-head on the # same scenario+seed. PLAYBACK_ROOT = Path( os.environ.get( "OPENRA_BENCH_PLAYBACK_ROOT", Path(__file__).parent / "playback" ) ) def _bv_scan(): try: from openra_bench.battle_viewer import scan return scan(PLAYBACK_ROOT) except Exception: # noqa: BLE001 — empty/missing root → empty viewer return [] def _bv_turn_md(v: dict, heading: str) -> str: if not v or v.get("n_turns", 0) == 0: return f"### {heading}\n\n_no episode / no turns_" m = v.get("manifest", {}) g = v.get("goal", {}) or {} lines = [ f"### {heading}", f"**{m.get('model','?')}** · run `{m.get('run_id','?')}` · " f"{m.get('scenario','?')} · seed {m.get('seed','?')} · " f"outcome **{m.get('outcome','?')}**", f"**turn {v.get('turn')} / {v['n_turns']}** · tick " f"{v.get('tick')}" + (f" · ⚡ {v['interrupt']}" if v.get("interrupt") else ""), ] if g: parts = [] for leaf in g.get("leaves", []): mark = ( "✅" if leaf.get("satisfied") else f"{float(leaf.get('ratio', 0.0)):.0%}" ) parts.append( f"{leaf['name']} {leaf.get('current')}/" f"{leaf.get('target')} {mark}" ) bars = " · ".join(parts) rv = g.get("reward_vector", {}) lines += [ f"**objective: {g.get('objective_progress',0):.0%}**" + (" ✅ WON" if g.get("won") else ""), (f"_{bars}_" if bars else ""), "reward vector: " + " ".join(f"`{k}={float(x):.2f}`" for k, x in rv.items()), ] # System prompt (the deterministic scenario knowledge the model # was given) — collapsible, shown with char count like the # training pipeline viewer. sp = str(v.get("system_prompt") or "") if sp: lines += [ "", f"
🧠 system prompt ({len(sp)} chars)" "\n\n```\n" + sp[:6000] + "\n```\n
" ] # DEBRIEF / briefing actually handed to the model this turn # (post-interrupt this is the scoped re-prompt). deb = str(v.get("debrief") or v.get("briefing") or "") if deb: tag = "⚡ DEBRIEF (interrupt)" if v.get("interrupt") else "briefing" lines += [ "", f"
{tag}\n\n```\n" + deb[:8000] + "\n```\n
" ] if v.get("reasoning"): lines += ["", "**🤔 reasoning (thinking)**", "> " + str(v["reasoning"]).replace("\n", "\n> ")] if v.get("assistant_text"): lines += ["", "**model said**", str(v["assistant_text"])] cmds = v.get("commands", []) lines += ["", "**tool calls**", "```\n" + ( "\n".join(cmds) if cmds else "(none)") + "\n```"] if v.get("tool_result"): lines += [f"tool result: `{v['tool_result']}`"] sig = v.get("signals", {}) if sig: lines += ["signals: " + " ".join( f"`{k}={sig[k]}`" for k in sig)] return "\n\n".join(s for s in lines if s != "") def _bv_b_label(e) -> str: return f"{e.run_id} / {e.model} ({e.outcome})" def bv_runs(): from openra_bench.battle_viewer import runs idx = _bv_scan() rs = runs(idx) return idx, gr.update(choices=rs, value=rs[0] if rs else None) def bv_on_run(idx, run): from openra_bench.battle_viewer import models ms = models(idx or [], run) if run else [] return gr.update(choices=ms, value=ms[0] if ms else None) def bv_on_model(idx, run, model): from openra_bench.battle_viewer import scenarios sc = scenarios(idx or [], run, model) if (run and model) else [] return gr.update(choices=sc, value=sc[0] if sc else None) def _bv_render(idx, run, model, scen, turn, compare, b_choice): from openra_bench.battle_viewer import ( compare_candidates, episode_view, find, ) idx = idx or [] a = find(idx, run, model, scen) if (run and model and scen) else None if a is None: return (None, _bv_turn_md({}, "A"), None, _bv_turn_md({}, "B"), "—", gr.update()) av = episode_view(a.dir, turn) n = av.get("n_turns", 1) ti = av.get("turn_idx", 0) cands = compare_candidates(idx, a) labels = [_bv_b_label(e) for e in cands] bv = {} if compare and b_choice: by = {_bv_b_label(e): e for e in cands} be = by.get(b_choice) if be is not None: bv = episode_view(be.dir, turn) return ( av.get("minimap_png"), _bv_turn_md(av, "A"), bv.get("minimap_png") if compare else None, _bv_turn_md(bv, "B") if compare else "_comparison off_", f"turn {ti + 1} / {n}", gr.update(choices=labels, value=b_choice if b_choice in labels else ( labels[0] if labels else None)), ) # ── Filtering ───────────────────────────────────────────────────────────────── def filter_leaderboard( search: str, agent_types: list[str], opponent: str, show_unverified: bool = True, ) -> pd.DataFrame: """Filter leaderboard by search, agent type, opponent, and verification status.""" df = load_data() # Filter by verification status if not show_unverified: df = df[df["Status"].str.contains("Verified", na=False) & ~df["Status"].str.contains("Unverified", na=False)] # Filter by agent type if agent_types: df = df[df["Type"].isin(agent_types)] # Filter by opponent if opponent and opponent != "All": df = df[df["Opponent"] == opponent] # Search by agent name (regex with fallback to literal on invalid patterns) if search and search.strip(): patterns = [p.strip() for p in search.split(",") if p.strip()] mask = pd.Series([False] * len(df), index=df.index) for pattern in patterns: try: mask |= df["Agent"].str.contains(pattern, case=False, regex=True, na=False) except re.error: mask |= df["Agent"].str.contains( re.escape(pattern), case=False, regex=True, na=False ) df = df[mask] # Re-rank after filtering df = df.reset_index(drop=True) df["Rank"] = range(1, len(df) + 1) return add_type_badges(df) # ── Result Persistence ──────────────────────────────────────────────────────── SUBMISSIONS_DIR = Path(__file__).parent / "submissions" SUBMISSIONS_DIR.mkdir(exist_ok=True) GAMES_JSONL = SUBMISSIONS_DIR / "games.jsonl" MIN_GAMES_FOR_LEADERBOARD = 5 # CommitScheduler pushes submissions to HF dataset (only on HF Spaces) _scheduler = None if os.environ.get("HF_TOKEN") and os.environ.get("SPACE_ID"): try: from huggingface_hub import CommitScheduler _scheduler = CommitScheduler( repo_id="openra-rl/bench-results", repo_type="dataset", folder_path=str(SUBMISSIONS_DIR), every=5, token=os.environ["HF_TOKEN"], ) except Exception: pass # Running locally without HF token — skip def _sanitize_csv_value(val): """Strip leading characters that trigger formula execution in spreadsheets.""" if isinstance(val, str): while val and val[0] in ("=", "+", "-", "@", "\t", "\r", "\n"): val = val[1:] val = val.replace("\n", " ").replace("\r", " ") return val # ── Rate Limiting ──────────────────────────────────────────────────────────── _submit_times: dict[str, list[float]] = defaultdict(list) MAX_SUBMITS_PER_HOUR = 20 def _check_rate_limit(identifier: str = "global") -> tuple[bool, str]: """Simple in-memory rate limiter. Returns (allowed, error_message).""" now = time.time() times = _submit_times[identifier] _submit_times[identifier] = [t for t in times if now - t < 3600] if len(_submit_times[identifier]) >= MAX_SUBMITS_PER_HOUR: return False, "Rate limit exceeded (max 20 submissions per hour). Try again later." _submit_times[identifier].append(now) return True, "" # ── HF Identity Verification ───────────────────────────────────────────────── def _verify_hf_token(token: str) -> tuple[str, str]: """Verify a HuggingFace token and return the username. Returns (hf_username, error_message). On success: ("username", ""). On failure: ("", "reason"). """ if not token or not token.strip(): return "", "no token provided" try: from huggingface_hub import HfApi api = HfApi() info = api.whoami(token=token.strip()) username = info.get("name", "") if not username: return "", "token valid but no username found" return username, "" except Exception as e: logger.debug("HF token verification failed: %s", e) return "", f"invalid token: {e}" # ── Raw Game Storage & Aggregation ──────────────────────────────────────────── def _save_raw_game(data: dict) -> None: """Append a single game result to the raw games log. Strips ``hf_token`` before writing (only ``hf_username`` is persisted). """ safe = {k: v for k, v in data.items() if k != "hf_token"} with open(GAMES_JSONL, "a") as f: f.write(json.dumps(safe) + "\n") # Also save to results.jsonl for CommitScheduler → HF dataset jsonl_path = SUBMISSIONS_DIR / "results.jsonl" with open(jsonl_path, "a") as f: f.write(json.dumps(safe) + "\n") def _load_raw_games() -> list[dict]: """Load all raw games from games.jsonl.""" if not GAMES_JSONL.exists(): return [] games = [] for line in GAMES_JSONL.read_text().splitlines(): if line.strip(): try: games.append(json.loads(line)) except json.JSONDecodeError: continue return games def _aggregate_agent_games( agent_name: str, agent_type: str, opponent: str, all_games: list[dict] | None = None, hf_username: str = "", ) -> tuple[int, dict | None]: """Aggregate all games for a specific agent+opponent pair. When *hf_username* is non-empty, only games with a matching ``hf_username`` are included. Anonymous games (empty hf_username) are never aggregated. Returns (game_count, aggregated_row_or_None). aggregated_row is None if game_count < MIN_GAMES_FOR_LEADERBOARD. """ if all_games is None: all_games = _load_raw_games() if not hf_username: # Anonymous games are not aggregated return 0, None matching = [ g for g in all_games if g.get("agent_name") == agent_name and g.get("agent_type") == agent_type and g.get("opponent") == opponent and g.get("hf_username") == hf_username ] count = len(matching) if count < MIN_GAMES_FOR_LEADERBOARD: return count, None game_results = [] for g in matching: game_results.append({ "win": g.get("win", g.get("result") == "win"), "kills_cost": g.get("kills_cost", 0), "deaths_cost": g.get("deaths_cost", 0), "assets_value": g.get("assets_value", 0), "ticks": g.get("ticks", 0), }) raw_score = compute_composite_score(game_results) multiplier = DIFFICULTY_MULTIPLIER.get(opponent, 1.0) total_kills = sum(g["kills_cost"] for g in game_results) total_deaths = sum(g["deaths_cost"] for g in game_results) return count, { "agent_name": agent_name, "agent_type": agent_type, "opponent": opponent, "difficulty": opponent, "games": count, "win_rate": round(100.0 * sum(1 for g in game_results if g["win"]) / count, 1), "score": round(raw_score * multiplier, 1), "avg_kills": round(total_kills / count), "avg_deaths": round(total_deaths / count), "kd_ratio": round(total_kills / max(total_deaths, 1), 2), "avg_economy": round(sum(g["assets_value"] for g in game_results) / count), "avg_game_length": round(sum(g["ticks"] for g in game_results) / count), "timestamp": max((g.get("timestamp", "")[:10] for g in matching), default=""), "replay_url": next( (g.get("replay_url", "") for g in reversed(matching) if g.get("replay_url")), "", ), "agent_url": next( (g.get("agent_url", "") for g in reversed(matching) if g.get("agent_url")), "", ), "hf_username": hf_username, "verified": True, } def _single_game_row(game: dict) -> dict: """Build a leaderboard row from a single anonymous game.""" game_results = [{ "win": game.get("win", game.get("result") == "win"), "kills_cost": game.get("kills_cost", 0), "deaths_cost": game.get("deaths_cost", 0), "assets_value": game.get("assets_value", 0), "ticks": game.get("ticks", 0), }] raw_score = compute_composite_score(game_results) opponent = game.get("opponent", "Normal") multiplier = DIFFICULTY_MULTIPLIER.get(opponent, 1.0) kills = game.get("kills_cost", 0) deaths = game.get("deaths_cost", 0) return { "agent_name": game.get("agent_name", ""), "agent_type": game.get("agent_type", ""), "opponent": opponent, "difficulty": opponent, "games": 1, "win_rate": round(100.0 * int(game_results[0]["win"]), 1), "score": round(raw_score * multiplier, 1), "avg_kills": kills, "avg_deaths": deaths, "kd_ratio": round(kills / max(deaths, 1), 2), "avg_economy": game.get("assets_value", 0), "avg_game_length": game.get("ticks", 0), "timestamp": game.get("timestamp", "")[:10], "replay_url": game.get("replay_url", ""), "agent_url": game.get("agent_url", ""), "hf_username": "", "verified": False, } def _rebuild_leaderboard() -> None: """Rebuild leaderboard CSV from raw games. Verified users (non-empty hf_username) are aggregated by (hf_username, agent_name, agent_type, opponent) with a minimum of 5 games to appear. Anonymous games (empty hf_username) appear as individual rows marked as unverified. """ all_games = _load_raw_games() if not all_games: return # No games yet, keep existing CSV as-is rows = [] # 1. Aggregate verified games verified_groups = set() for g in all_games: hf_user = g.get("hf_username", "") if hf_user: key = (hf_user, g.get("agent_name", ""), g.get("agent_type", ""), g.get("opponent", "")) verified_groups.add(key) for hf_user, name, atype, opp in verified_groups: count, agg = _aggregate_agent_games(name, atype, opp, all_games, hf_username=hf_user) if agg is not None: rows.append(agg) # 2. Add anonymous games as individual rows for g in all_games: if not g.get("hf_username"): rows.append(_single_game_row(g)) if not rows: return # No qualifying entries rows.sort(key=lambda r: r.get("score", 0), reverse=True) fieldnames = LEADERBOARD_FIELDNAMES with open(DATA_PATH, "w", newline="") as f: writer = csv.DictWriter(f, fieldnames=fieldnames) writer.writeheader() for row in rows: writer.writerow({k: _sanitize_csv_value(row.get(k, "")) for k in fieldnames}) LEADERBOARD_FIELDNAMES = [ "agent_name", "agent_type", "opponent", "difficulty", "games", "win_rate", "score", "avg_kills", "avg_deaths", "kd_ratio", "avg_economy", "avg_game_length", "timestamp", "replay_url", "agent_url", "hf_username", "verified", ] # ── Submission Handling ─────────────────────────────────────────────────────── MAX_REPLAY_SIZE = 10 * 1024 * 1024 # 10 MB VALID_OPPONENTS = {"Beginner", "Easy", "Medium", "Normal", "Hard"} VALID_AGENT_TYPES = {"Scripted", "LLM", "RL"} REQUIRED_FIELDS = [ "agent_name", "agent_type", "opponent", "result", "ticks", "kills_cost", "deaths_cost", "assets_value", ] def validate_submission(data: dict) -> tuple[bool, str]: """Validate an uploaded JSON submission. Returns (is_valid, error_message). """ for field in REQUIRED_FIELDS: if field not in data: return False, f"Missing required field: {field}" if data["agent_type"] not in VALID_AGENT_TYPES: return False, ( f"Invalid agent_type: {data['agent_type']}. " f"Must be one of: {', '.join(sorted(VALID_AGENT_TYPES))}" ) if data["opponent"] not in VALID_OPPONENTS: return False, ( f"Invalid opponent: {data['opponent']}. " f"Must be one of: {', '.join(sorted(VALID_OPPONENTS))}" ) # Type checks for numeric fields for field in ("ticks", "kills_cost", "deaths_cost", "assets_value"): if not isinstance(data[field], (int, float)): return False, f"Field '{field}' must be a number" # String length limits if len(str(data["agent_name"])) > 100: return False, "agent_name must be 100 characters or fewer" # agent_url: optional, but must be http(s) if provided agent_url = str(data.get("agent_url", "")).strip() if agent_url and not agent_url.startswith(("http://", "https://")): return False, "agent_url must be an HTTP(S) URL" if len(agent_url) > 500: return False, "agent_url must be 500 characters or fewer" return True, "" def handle_upload(json_file, replay_file) -> tuple[str, pd.DataFrame]: """Process an uploaded bench submission JSON + optional replay.""" if json_file is None: return "Please upload a JSON file.", add_type_badges(load_data()) allowed, err = _check_rate_limit() if not allowed: return err, add_type_badges(load_data()) try: with open(json_file.name) as f: data = json.load(f) except (json.JSONDecodeError, Exception) as e: return f"Invalid JSON: {e}", add_type_badges(load_data()) is_valid, error = validate_submission(data) if not is_valid: return f"Validation error: {error}", add_type_badges(load_data()) hf_username, anon_warning = _process_identity(data) # Save replay if provided if replay_file is not None: import shutil orig = Path(replay_file.name) if orig.stat().st_size > MAX_REPLAY_SIZE: return "Replay file too large (max 10 MB).", add_type_badges(load_data()) ts = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ") slug = re.sub(r"[^a-zA-Z0-9_-]", "", data["agent_name"].replace("/", "_").replace(" ", "_"))[:30] replay_name = f"replay-{slug}-{ts}.orarep" shutil.copy2(str(orig), SUBMISSIONS_DIR / replay_name) data["replay_url"] = replay_name _save_raw_game(data) _rebuild_leaderboard() agent_name = data["agent_name"] opponent = data["opponent"] if not hf_username: msg = ( f"Recorded anonymous game for **{agent_name}** vs {opponent}. " f"Add an HF token to aggregate games and track progress." ) if anon_warning: msg = f"{anon_warning} {msg}" else: count, agg = _aggregate_agent_games( agent_name, data["agent_type"], opponent, hf_username=hf_username, ) if count < MIN_GAMES_FOR_LEADERBOARD: remaining = MIN_GAMES_FOR_LEADERBOARD - count msg = ( f"Recorded game {count}/{MIN_GAMES_FOR_LEADERBOARD} for " f"**{agent_name}** vs {opponent}. " f"Play {remaining} more game{'s' if remaining != 1 else ''} " f"to appear on the leaderboard!" ) else: msg = ( f"**{agent_name}** vs {opponent} updated \u2014 " f"{count} games, score **{agg['score']}** (win rate {agg['win_rate']}%)" ) return msg, add_type_badges(load_data()) def _process_identity(data: dict) -> tuple[str, str]: """Verify HF token if present, set hf_username on data. Returns (hf_username, warning_message). """ token = data.pop("hf_token", "") if token: hf_username, err = _verify_hf_token(token) if hf_username: data["hf_username"] = hf_username return hf_username, "" else: data["hf_username"] = "" return "", f"HF token verification failed ({err}). Submitted as anonymous." data.setdefault("hf_username", "") return "", "" def _build_response(agent_name: str, agent_type: str, opponent: str, hf_username: str, anonymous_warning: str, all_games: list[dict] | None = None) -> str: """Build a response message after saving a game.""" parts = [] if anonymous_warning: parts.append(anonymous_warning) if not hf_username: # Anonymous: not aggregated parts.append( f"OK: recorded anonymous game for {agent_name} vs {opponent}. " f"Add an HF token to aggregate games and track progress." ) return " ".join(parts) count, agg = _aggregate_agent_games( agent_name, agent_type, opponent, all_games=all_games, hf_username=hf_username, ) if count < MIN_GAMES_FOR_LEADERBOARD: remaining = MIN_GAMES_FOR_LEADERBOARD - count parts.append( f"OK: recorded game {count}/{MIN_GAMES_FOR_LEADERBOARD} for " f"{agent_name} vs {opponent}. " f"Play {remaining} more game{'s' if remaining != 1 else ''} " f"to appear on the leaderboard!" ) else: parts.append( f"OK: {agent_name} vs {opponent} updated \u2014 " f"{count} games, score {agg['score']} (win rate {agg['win_rate']}%)" ) return " ".join(parts) def handle_api_submit(json_data: str) -> str: """API endpoint: accept JSON string submission. Used by CLI auto-upload.""" allowed, err = _check_rate_limit() if not allowed: return err try: data = json.loads(json_data) except (json.JSONDecodeError, Exception) as e: return f"Invalid JSON: {e}" is_valid, error = validate_submission(data) if not is_valid: return f"Validation error: {error}" hf_username, anon_warning = _process_identity(data) _save_raw_game(data) _rebuild_leaderboard() return _build_response( data["agent_name"], data["agent_type"], data["opponent"], hf_username, anon_warning, ) def handle_api_submit_with_replay(json_data: str, replay_file) -> str: """API endpoint: accept JSON + replay file. Used by CLI with --replay.""" allowed, err = _check_rate_limit() if not allowed: return err try: data = json.loads(json_data) except (json.JSONDecodeError, Exception) as e: return f"Invalid JSON: {e}" is_valid, error = validate_submission(data) if not is_valid: return f"Validation error: {error}" hf_username, anon_warning = _process_identity(data) # Save replay if provided if replay_file is not None: import shutil orig = Path(replay_file) if isinstance(replay_file, str) else Path(replay_file.name) if orig.exists() and orig.stat().st_size > MAX_REPLAY_SIZE: return "Replay file too large (max 10 MB)" ts = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ") slug = re.sub(r"[^a-zA-Z0-9_-]", "", data["agent_name"].replace("/", "_").replace(" ", "_"))[:30] replay_name = f"replay-{slug}-{ts}.orarep" shutil.copy2(str(orig), SUBMISSIONS_DIR / replay_name) data["replay_url"] = replay_name _save_raw_game(data) _rebuild_leaderboard() return _build_response( data["agent_name"], data["agent_type"], data["opponent"], hf_username, anon_warning, ) # ── UI ──────────────────────────────────────────────────────────────────────── ABOUT_MD = """ ## What is OpenRA-Bench? **OpenRA-Bench** is a standardized benchmark for evaluating AI agents playing [Red Alert](https://www.openra.net/) through the [OpenRA-RL](https://openra-rl.dev) environment. ### Evaluation Protocol - **Game**: Red Alert (OpenRA engine) - **Format**: 1v1 agent vs built-in AI - **Opponents**: Beginner, Easy, Medium, Normal, Hard difficulty - **Games per entry**: Minimum 5 games per configuration - **Metrics**: Win rate, composite score, K/D ratio, economy ### Composite Score The benchmark score combines four components, scaled by opponent difficulty: | Component | Weight | Description | |-----------|--------|-------------| | Win Rate | 50% | Percentage of games won | | Military Efficiency | 20% | Kill/death cost ratio (0 if no combat) | | Economy | 20% | Final asset value (normalized) | | Speed | 10% | Faster decisive games score higher | **Difficulty multiplier**: Beginner (0.5x), Easy (0.7x), Medium (0.85x), Normal (1.0x), Hard (1.2x) **Minimum games**: 5 games required per agent+opponent to appear on the leaderboard (verified users only). ### Identity & Verification - **Verified**: Include your HuggingFace token (`hf_token`) in submissions. Games are aggregated by HF username + agent name + opponent. - **Anonymous**: No token required. Games appear individually with an "Unverified" badge and are not aggregated across sessions. ### Agent Types - **Scripted**: Rule-based bots with hardcoded strategies - **LLM**: Language model agents (Claude, GPT, etc.) - **RL**: Reinforcement learning policies (PPO, SAC, etc.) ### Links - [OpenRA-RL Documentation](https://openra-rl.dev) - [GitHub Repository](https://github.com/yxc20089/OpenRA-RL) - [OpenRA-Bench Source](https://github.com/yxc20089/OpenRA-Bench) - [OpenEnv Framework](https://huggingface.co/openenv) - [HuggingFace Space](https://huggingface.co/spaces/openra-rl/OpenRA-Bench) """ SUBMIT_MD = """ --- ## Other Submission Methods ### CLI Auto-Upload Set `BENCH_URL` and optionally `HF_TOKEN` in your OpenRA-RL config. Results upload automatically after each game. With a HF token, games are aggregated under your verified username: ```yaml # config.yaml agent: bench_url: "https://openra-rl-openra-bench.hf.space" hf_token: "hf_..." # Optional: enables verified aggregation ``` ### CLI Manual Upload Upload a previously exported bench JSON: ```bash python -m openra_env.bench_submit ~/.openra-rl/bench-exports/bench-*.json ``` ### Batch Evaluation (5+ games) ```bash git clone https://github.com/yxc20089/OpenRA-Bench.git cd OpenRA-Bench pip install -r requirements.txt pip install openra-rl openra-rl-util python evaluate.py \\ --agent scripted \\ --agent-name "MyBot-v1" \\ --agent-type Scripted \\ --opponent Normal \\ --games 10 \\ --server http://localhost:8000 ``` ### Evaluation Parameters | Parameter | Description | |-----------|-------------| | `--agent` | Agent type: `scripted`, `llm`, `mcp`, `custom` | | `--agent-name` | Display name on the leaderboard | | `--agent-type` | Category: `Scripted`, `LLM`, `RL` | | `--opponent` | AI difficulty: `Beginner`, `Easy`, `Medium`, `Normal`, `Hard` | | `--games` | Number of games (minimum 5) | | `--server` | OpenRA-RL server URL (local or HuggingFace-hosted) | ### Custom Agents Implement the standard `reset/step` loop: ```python from openra_env.client import OpenRAEnv from openra_env.models import OpenRAAction async with OpenRAEnv("http://localhost:8000") as env: obs = await env.reset() while not obs.done: action = your_agent.decide(obs) obs = await env.step(action) ``` Then run `evaluate.py --agent custom` with your agent integrated. """ # ── Scenarios tab (interactive catalog) ──────────────────────────────────────── _CAP_COLORS = { "perception": "#7497db", "reasoning": "#9b8cce", "action": "#5fae7a", "adversarial": "#d2683c", } _translate_cache: dict[str, str] = {} def _google_translate_zh(text: str) -> str: """Translate English text to Simplified Chinese via Google Translate.""" if not text or not text.strip(): return text if text in _translate_cache: return _translate_cache[text] import urllib.parse import urllib.request url = ( "https://translate.googleapis.com/translate_a/single" "?client=gtx&sl=en&tl=zh-CN&dt=t&q=" + urllib.parse.quote(text) ) try: req = urllib.request.Request(url, headers={"User-Agent": "Mozilla/5.0"}) with urllib.request.urlopen(req, timeout=10) as resp: data = json.loads(resp.read().decode()) result = "".join(seg[0] for seg in data[0] if seg[0]) _fixups = [ ("游戏前勾选", "游戏刻"), ("游戏勾选", "游戏刻"), ("游戏滴答", "游戏刻"), ("游戏刻度", "游戏刻"), ("游戏蜱虫", "游戏刻"), ("游戏壁虱", "游戏刻"), ("游戏报价", "游戏刻"), ("游戏打勾", "游戏刻"), ("决策轮次", "决策回合"), ("决策转弯", "决策回合"), ("勾号", "刻"), ] for wrong, right in _fixups: result = result.replace(wrong, right) _translate_cache[text] = result return result except Exception: return text def _scenarios_catalog_df() -> pd.DataFrame: """Load every active scenario pack into a DataFrame for the catalog.""" try: from openra_bench.scenarios import discover_packs except Exception: # noqa: BLE001 return pd.DataFrame(columns=[ "ID", "Title", "Capability", "Map", "Real-World Meaning", "Robotics Analogue", "Benchmark Anchor", ]) rows = [] for p in discover_packs(): if p.meta.status != "active": continue anchors = ", ".join(p.meta.benchmark_anchor) if p.meta.benchmark_anchor else "" rows.append({ "ID": p.meta.id, "Title": p.meta.title, "Capability": p.meta.capability, "Map": p.base_map if isinstance(p.base_map, str) else "generated", "Real-World Meaning": p.meta.real_world_meaning, "Robotics Analogue": p.meta.robotics_analogue, "Benchmark Anchor": anchors, }) return pd.DataFrame(rows) def _scenarios_filter(search: str, capabilities: list[str]) -> pd.DataFrame: """Filter the scenario catalog by search term and capability.""" df = _scenarios_catalog_df() if not len(df): return df df = df[df["Capability"].isin(capabilities or [])] if search and search.strip(): q = search.strip().lower() mask = ( df["ID"].str.lower().str.contains(q, na=False) | df["Title"].str.lower().str.contains(q, na=False) | df["Real-World Meaning"].str.lower().str.contains(q, na=False) ) df = df[mask] return df.reset_index(drop=True) def _scenarios_detail_md(pack_id: str, lang: str = "en") -> str: """Render full detail for one scenario pack as Markdown. When lang='zh', all objectives are shown in Chinese via Google Translate. """ if not pack_id or not pack_id.strip(): return "_Select a scenario from the table above to see details._" pack_id = pack_id.strip() try: from openra_bench.game_knowledge import objective_brief from openra_bench.scenarios import load_pack from openra_bench.scenarios.loader import PACKS_DIR path = PACKS_DIR / f"{pack_id}.yaml" if not path.exists(): return f"Pack `{pack_id}` not found." pack = load_pack(path) except Exception as e: # noqa: BLE001 return f"Error loading `{pack_id}`: {e}" cap = pack.meta.capability color = _CAP_COLORS.get(cap, "#666") anchors = ", ".join(pack.meta.benchmark_anchor) if pack.meta.benchmark_anchor else "none" rwm = pack.meta.real_world_meaning rob = pack.meta.robotics_analogue if lang == "zh": rwm = _google_translate_zh(rwm) rob = _google_translate_zh(rob) why_label = "为什么有这个场景：" if lang == "zh" else "Why this exists:" robo_label = "机器人类比：" if lang == "zh" else "Robotics analogue:" anchor_label = "基准锚点：" if lang == "zh" else "Benchmark anchors:" levels_label = "### 难度等级" if lang == "zh" else "### Levels" lines = [ f"## {_md_escape(pack.meta.title)}", f"**ID:** `{pack.meta.id}` | **Capability:** " f"{cap} | " f"**Map:** `{pack.base_map if isinstance(pack.base_map, str) else 'generated'}`", "", f"**{why_label}** {_md_escape(rwm)}", "", f"**{robo_label}** {_md_escape(rob)}", "", f"**{anchor_label}** {_md_escape(anchors)}", "", "---", "", levels_label, ] cells = [] if pack.configs: for c in pack.configs: try: cl = pack.compile_config(c.name) cells.append((c.name, cl)) except Exception as e: # noqa: BLE001 cells.append((c.name, e)) else: for lv in ("easy", "medium", "hard"): try: cl = pack.compile(lv) cells.append((lv, cl)) except Exception as e: # noqa: BLE001 cells.append((lv, e)) diff_zh = {"easy": "简单", "medium": "中等", "hard": "困难"} for label, cl in cells: if isinstance(cl, Exception): lines.append(f"\n**{label}** — compile error: {cl}") continue fog = getattr(cl, "fog_mode", "vision") cash_str = f" | cash: {cl.starting_cash}" if cl.starting_cash is not None else "" display_label = diff_zh.get(label, label) if lang == "zh" else label lines.append( f"\n**{display_label}** (level {cl.level} | fog: {fog} | " f"turns: {cl.max_turns}{cash_str})" ) try: ob = objective_brief( cl.scenario.description, cl.win_condition, cl.fail_condition, cl.max_turns, getattr(cl, "objective_coords", "exact"), ) if lang == "zh": ob = _google_translate_zh(ob) lines.append(f"```\n{ob}\n```") except Exception as e: # noqa: BLE001 lines.append(f"_(objective error: {e})_") return "\n".join(lines) # ── Play tab (human-labeling machine) ───────────────────────────────────────── # Lets a human play the exact scenarios LLM agents are scored on, by # clicking the minimap — the Phase 2 human-labeling machine. Backed by # openra_bench.human_labeling.InteractiveSession (turn-steppable) so a # human's run is scored by the identical rules as a model's. _PLAY_LEVELS = ["easy", "medium", "hard"] _PLAY_UPSCALE = 5 # tactical-minimap cell scale for the Play tab _PLAY_KEYBOARD_JS = r""" () => { if (window.__openraBenchPlayEnterBound) return; window.__openraBenchPlayEnterBound = true; document.addEventListener("keydown", (event) => { if (event.key !== "Enter" || event.shiftKey || event.metaKey || event.ctrlKey || event.altKey || event.repeat) { return; } const target = event.target; const tag = (target && target.tagName || "").toLowerCase(); if (["input", "textarea", "select", "button"].includes(tag) || (target && target.isContentEditable)) { return; } const root = document.getElementById("play-end-turn-btn"); if (!root) return; const rect = root.getBoundingClientRect(); if (rect.width === 0 && rect.height === 0) return; const button = root.tagName && root.tagName.toLowerCase() === "button" ? root : root.querySelector("button"); if (!button || button.disabled) return; event.preventDefault(); button.click(); }, true); } """ def _play_scenarios() -> list[str]: """Active pack ids playable in the Play tab.""" try: from openra_bench.scenarios import load_pack from openra_bench.scenarios.loader import PACKS_DIR out = [] for f in sorted(PACKS_DIR.glob("*.yaml")): if f.name.startswith(("_", "TEMPLATE")): continue try: if load_pack(f).meta.status == "active": out.append(f.stem) except Exception: # noqa: BLE001 continue return out except Exception: # noqa: BLE001 return [] def _md_escape(text: str) -> str: """Escape Markdown-significant characters so scenario prose renders literally — e.g. the `~` in 'NE ~110,6' must not become strikethrough.""" text = text.replace("\\", "\\\\") for ch in ("~", "*", "_", "`", "#"): text = text.replace(ch, "\\" + ch) return text def _play_minimap(render_state: dict, sel=None, queue=None): """The Play-tab minimap — `render_tactical_minimap` plus a white boundary on the selected units and movement arrows. An arrow points to a unit's queued destination this turn; if it has no queued order but is already moving, the arrow points to its in-engine target.""" try: from openra_bench.minimap import render_tactical_minimap sel_ids = {str(s) for s in (sel or [])} # A queued move/attack this turn overrides the in-flight target. queued_dest: dict = {} for a in queue or []: if getattr(a, "mode", "") in ( "move", "attack", "attack_move" ) and getattr(a, "target", None): for uid in a.units: queued_dest[str(uid)] = (a.target[0], a.target[1]) arrows = [] for u in render_state.get("units_summary", []) or []: if not isinstance(u, dict): continue uid = str(u.get("id", "")) fx, fy = u.get("cell_x"), u.get("cell_y") if fx is None or fy is None: continue if uid in queued_dest: tx, ty = queued_dest[uid] arrows.append((fx, fy, tx, ty, "queued")) elif u.get("activity") == "moving" and ( u.get("target_x") is not None ): arrows.append( (fx, fy, u["target_x"], u["target_y"], "enroute") ) return render_tactical_minimap( render_state, scale=_PLAY_UPSCALE, grid=True, legend=True, selected=sel_ids, arrows=arrows, ) except Exception: # noqa: BLE001 return None def _play_render_state(sess, show_objectives: bool = False) -> dict: rs = sess.render_state() if not show_objectives and "objective_regions" in rs: rs = dict(rs) rs.pop("objective_regions", None) return rs def _play_objective_md(sess) -> str: """The scenario objective — what the human must do to WIN.""" if sess is None: return "" obj = (getattr(sess, "objective", "") or "").strip() if not obj: return "" return f"### 🎯 Objective\n{_md_escape(obj)}" _PLAY_UNIT_COLS = ["sel", "unit", "type", "cell", "hp", "status"] def _play_units_df(sess, sel): """Table of the human's own units. Selected units are marked '▶' and sorted to the top so the current selection is obvious. `hp` is the 0-1 fraction the engine reports, shown as a percentage.""" if sess is None: return pd.DataFrame(columns=_PLAY_UNIT_COLS) try: rs = sess.render_state() except Exception: # noqa: BLE001 return pd.DataFrame(columns=_PLAY_UNIT_COLS) selset = {str(s) for s in (sel or [])} rows = [] for u in rs.get("units_summary", []) or []: if not isinstance(u, dict): continue uid = str(u.get("id", "")) try: hp_txt = f"{int(round(float(u.get('hp', 1.0)) * 100))}%" except (TypeError, ValueError): hp_txt = "?" is_sel = uid in selset rows.append({ "sel": "▶" if is_sel else "", "unit": uid, "type": u.get("type") or u.get("actor_type") or "?", "cell": f"({u.get('cell_x')}, {u.get('cell_y')})", "hp": hp_txt, "status": u.get("activity", "") or "idle", "_sel": is_sel, }) df = pd.DataFrame(rows, columns=_PLAY_UNIT_COLS + ["_sel"]) # Selected units float to the top so the selection is unmistakable. df = ( df.sort_values("_sel", ascending=False, kind="stable") .drop(columns="_sel") .reset_index(drop=True) ) return df def _play_status_md(sess) -> str: if sess is None: return "_No active session — pick a scenario and click **Start**._" st = sess.status() line = f"**Turn {st['turn']}/{st['max_turns']}** · tick {st['tick']}" if st["done"]: line += f" · **{st['outcome'].upper()}** — game over" if st.get("save_path"): line += ( f"\n\n_Run saved (standard playback format): " f"`{st['save_path']}`_" ) return line def _play_briefing_md(sess, sel, queue, note: str = "") -> str: """The turn panel: current selection + queued orders FIRST, then the exact text briefing the model is given for this turn.""" if sess is None: return "" try: from openra_bench.human_labeling import HumanController brief = HumanController._briefing(sess.render_state()) except Exception: # noqa: BLE001 brief = "" sel_txt = ", ".join(sel) if sel else "(none)" q_txt = "; ".join(a.describe() for a in queue) if queue else "(none)" head = f"{note}\n\n" if note else "" return ( f"{head}" f"**Selected units:** {sel_txt} \n" f"**Queued this turn:** {q_txt}\n\n" f"**Turn briefing — exactly what the model sees:**\n" f"```\n{brief}\n```" ) def _play_render(sess, sel, queue, show_objectives=False): img = ( _play_minimap(_play_render_state(sess, show_objectives), sel, queue) if sess is not None else None ) return ( img, _play_briefing_md(sess, sel, queue), _play_status_md(sess), _play_units_df(sess, sel), ) def _play_start(prev_sess, pack, level, seed, show_objectives=False): # Release any prior session's engine env before opening a new one. if prev_sess is not None: try: prev_sess.close() except Exception: # noqa: BLE001 pass empty_units = _play_units_df(None, []) if not pack: return ( None, [], [], "", None, "", "_pick a scenario first_", empty_units, ) try: from openra_bench.human_labeling import InteractiveSession sess = InteractiveSession.from_pack( pack, level or "easy", int(seed or 1) ) except Exception as e: # noqa: BLE001 return ( None, [], [], "", None, f"⚠️ {e}", "_start failed_", empty_units, ) img, brief, status, units = _play_render(sess, [], [], show_objectives) return ( sess, [], [], _play_objective_md(sess), img, brief, status, units, ) def _play_click(sess, sel, queue, show_objectives, evt: gr.SelectData): """Contextual minimap click — classic RTS interaction, no mode: * click a cell holding YOUR unit(s) → select/deselect them (toggle); * click an enemy with units selected → queue an ATTACK on it; * click empty ground with a selection → queue a MOVE there. So 'select a unit, then click where to send it' just works.""" if sess is None or evt is None or evt.index is None: return ( sel, queue, _play_briefing_md(sess, sel, queue), _play_units_df(sess, sel), None, ) note = "" try: from openra_bench.human_labeling import ( HumanAction, enemy_at_cell, minimap_click_to_cell, own_units_at_cell, ) px, py = evt.index # pixels in the displayed (upscaled) image rs = sess.render_state() rows = [r for r in (rs.get("minimap") or "").split("\n") if r] if not rows: return ( sel, queue, _play_briefing_md(sess, sel, queue), _play_units_df(sess, sel), None, ) h = len(rows) w = max(len(r) for r in rows) img_w = w * 6 * _PLAY_UPSCALE img_h = h * 6 * _PLAY_UPSCALE cx, cy = minimap_click_to_cell(px, py, img_w, img_h, w, h) note = f"🖱 Cell **({cx}, {cy})**" own_here = own_units_at_cell(rs, cx, cy, radius=0) enemy_here = enemy_at_cell(rs, cx, cy, radius=0) sel = list(sel) if own_here: # Toggle-select your own unit(s) on this cell. for uid in own_here: if uid in sel: sel.remove(uid) else: sel.append(uid) note += ( f" — selected unit {', '.join(own_here)} · " f"**{len(sel)} selected**" ) elif sel and enemy_here: queue = queue + [ HumanAction( mode="attack", units=list(sel), target_id=enemy_here, target=(cx, cy), ) ] note += ( f" — queued **attack** on enemy {enemy_here} " f"({len(sel)} unit(s))" ) elif sel: queue = queue + [ HumanAction(mode="move", units=list(sel), target=(cx, cy)) ] note += f" — queued **move** of {len(sel)} unit(s) here" else: note += " — empty (select one of your units first)" except Exception as e: # noqa: BLE001 logger.warning("play click failed: %s", e) note = "" # Re-render the minimap so the selection boundary + move arrows # update live as the player clicks. img = ( _play_minimap(_play_render_state(sess, show_objectives), sel, queue) if sess is not None else None ) return ( sel, queue, _play_briefing_md(sess, sel, queue, note), _play_units_df(sess, sel), img, ) def _play_end_turn(sess, sel, queue, show_objectives=False): if sess is not None and not sess.done: try: sess.submit_turn(list(queue)) except Exception as e: # noqa: BLE001 logger.warning("play submit_turn failed: %s", e) img, brief, status, units = _play_render(sess, [], [], show_objectives) return sess, [], [], img, brief, status, units def _play_clear_queue(sess, sel, show_objectives=False): """Cancel queued orders this turn (keeps the unit selection).""" img = ( _play_minimap(_play_render_state(sess, show_objectives), sel, []) if sess is not None else None ) return ( [], _play_briefing_md(sess, sel, []), _play_units_df(sess, sel), img, ) def _play_clear_selection(sess, queue, show_objectives=False): """Cancel the current unit selection (keeps queued orders).""" img = ( _play_minimap(_play_render_state(sess, show_objectives), [], queue) if sess is not None else None ) return ( [], _play_briefing_md(sess, [], queue), _play_units_df(sess, []), img, ) def _play_toggle_objectives(sess, sel, queue, show_objectives): if sess is None: return None return _play_minimap( _play_render_state(sess, show_objectives), sel, queue ) def _play_build_item(sess, sel, queue, item, show_objectives=False): """Queue a production order for the current turn (Play tab build queue).""" note = "" if sess is not None: item = str(item or "").strip().lower() if item: try: from openra_bench.human_labeling import HumanAction queue = queue + [HumanAction(mode="build", unit_type=item)] note = f"Queued **build {item}**" except Exception as e: # noqa: BLE001 logger.warning("play build queue failed: %s", e) note = "" img = ( _play_minimap(_play_render_state(sess, show_objectives), sel, queue) if sess is not None else None ) return ( queue, "", _play_briefing_md(sess, sel, queue, note), _play_units_df(sess, sel), img, ) # ── Human-study mode ───────────────────────────────────────────────── # Walks a recruited player through the fixed 24-pack study subset under # 3 conditions (72 games, per-player counterbalanced). Every game saves # to the standard Playback format — apples-to-apple with model runs. def _study_progress_md(st: dict) -> str: pl = st.get("playlist", []) i = st.get("idx", 0) if not pl: return "_Enter your name and click **Begin study**._" if i >= len(pl): return ( f"**✅ Study complete** — all {len(pl)} games done. " f"Thank you, `{st['player']}`!" ) pack, level, cond = pl[i] return ( f"**Study — game {i + 1} / {len(pl)}** · player `{st['player']}`\n\n" f"`{pack}` [{level}] · condition: **{cond}** \n" f"_Play to game-over, then click **Next scenario ▶**._" ) def _study_render(st: dict, prev_sess): """Open the study session for st's current cell and render it.""" if prev_sess is not None: try: prev_sess.close() except Exception: # noqa: BLE001 pass empty = _play_units_df(None, []) pl = st.get("playlist", []) if st.get("idx", 0) >= len(pl): return (None, [], [], "", None, "", "", empty, st, _study_progress_md(st)) pack, level, cond = pl[st["idx"]] try: from openra_bench.human_study import open_study_session sess = open_study_session( pack, level, cond, player=st["player"], seed=1 ) except Exception as e: # noqa: BLE001 return (None, [], [], "", None, f"⚠️ {e}", "_study load failed_", empty, st, _study_progress_md(st)) img, brief, status, units = _play_render(sess, [], [], False) return (sess, [], [], _play_objective_md(sess), img, brief, status, units, st, _study_progress_md(st)) def _study_begin(prev_sess, player): import hashlib from openra_bench.human_study import study_playlist player = (player or "").strip() or "anon" # Per-player counterbalancing — a stable seed from the name. seed = int(hashlib.md5(player.encode()).hexdigest()[:8], 16) st = {"player": player, "playlist": study_playlist(seed), "idx": 0} return _study_render(st, prev_sess) def _study_next(prev_sess, st): if not st or "playlist" not in st: return _study_render({"player": "anon", "playlist": []}, prev_sess) st = dict(st) st["idx"] = st.get("idx", 0) + 1 return _study_render(st, prev_sess) # ── Playlist mode (cold-start non-gamer UX) ────────────────────────── # A separate, simpler tab from `Play` (sandbox/debug) and the 24-pack # `Study` accordion (recruited tester). The Playlist tab walks a # non-gamer through `NOVICE_PLAYLIST` — a curated 20-pack list whose # objectives are visually obvious and whose tool surface fits the # reduced palette (move/attack/build/end-turn). Plain-language # objective above the minimap, jargon-substituted everywhere, auto- # advance on game-over, progress bar at the top, summary table at the # end. The underlying engine session is the same `InteractiveSession` # the `Play` tab uses (and the 24-pack study uses), so a Playlist run # is still apples-to-apple with a model run on the same scenario. def _playlist_state(player: str = "anon") -> dict: """Initial Playlist tab state — the curated 20-pack list, idx 0, no completed games yet, no game-over timestamp.""" from openra_bench.playlist import NOVICE_PLAYLIST return { "player": (player or "").strip() or "anon", "playlist": list(NOVICE_PLAYLIST), "idx": 0, "results": [], # list of session_summary_row dicts "done_at": None, # wall-clock when current game ended "submitted": False, # baseline already posted } def _playlist_progress_md(st: dict) -> str: """The progress bar at the top of the Playlist tab — '1 of 20', plus the text bar so a non-gamer can see at-a-glance how far they've come.""" from openra_bench.playlist import playlist_progress_bar pl = st.get("playlist") or [] total = len(pl) idx = int(st.get("idx", 0) or 0) if total == 0: return "_Enter your name and click **Start playlist** to begin._" if idx >= total: return f"### ✅ Playlist complete — {total} of {total}\n\n`{'▮' * 20} 100%`" bar = playlist_progress_bar(idx, total) return f"### Scenario {idx + 1} of {total} · player `{st.get('player', 'anon')}`\n\n`{bar}`" def _playlist_objective_md(sess) -> str: """The plain-English 1-2 sentence objective shown above the minimap. Drops jargon and the structured WIN/LOSE machine block — those live behind the **Details** expand instead.""" if sess is None: return "" from openra_bench.playlist import simplify_objective raw = (getattr(sess, "objective", "") or "").strip() if not raw: return "" plain = simplify_objective(raw) if not plain: plain = simplify_objective(raw, max_chars=600) return f"### 🎯 Your goal\n{_md_escape(plain)}" def _playlist_details_md(sess) -> str: """The full briefing (jargon, win clauses, fail clauses) tucked behind the **Details** expand. Same content `_play_briefing_md` shows in the Play tab — an interested tester can still drill in.""" if sess is None: return "" try: from openra_bench.human_labeling import HumanController from openra_bench.playlist import simplify_text brief = HumanController._briefing(sess.render_state()) except Exception: # noqa: BLE001 brief = "" plain_brief = simplify_text(brief) obj = (getattr(sess, "objective", "") or "").strip() plain_obj = simplify_text(obj) return ( f"**Full objective**\n\n{plain_obj}\n\n" f"**Per-turn briefing (model sees the same):**\n```\n" f"{plain_brief}\n```" ) def _playlist_status_md(sess, st: dict) -> str: """Status line under the minimap. After a game ends, append the auto-advance countdown so the player knows the next scenario is coming.""" base = _play_status_md(sess) if sess is None or not getattr(sess, "done", False): return base from openra_bench.playlist import AUTO_ADVANCE_WAIT_SECONDS pl = st.get("playlist") or [] idx = int(st.get("idx", 0) or 0) n = len(pl) outcome = (getattr(sess, "outcome", "draw") or "draw").upper() if idx + 1 >= n: tail = "Final scenario complete — see your **Session summary** below." else: tail = ( f"**Game {idx + 1} of {n}: {outcome}** — next scenario in " f"{int(AUTO_ADVANCE_WAIT_SECONDS)}s. Click **Skip wait ▶** to go now." ) return f"{base}\n\n{tail}" def _playlist_should_show_build(sess) -> bool: """Hide the Build textbox unless the active pack actually exposes a build verb — keeps the UI sparse for the move-and-shoot majority. """ if sess is None: return False try: from openra_bench.playlist import needs_build_tool compiled = getattr(sess, "compiled", None) meta = getattr(compiled, "meta", None) if compiled else None # Pack-level `tools:` lives on compile_config / pack.base. tools = ( getattr(compiled, "tools", None) or getattr(getattr(compiled, "scenario", None), "tools", None) or [] ) return needs_build_tool(tools or []) and not getattr(sess, "done", False) except Exception: # noqa: BLE001 return False def _playlist_simplified_units_df(sess, sel): """Same as `_play_units_df` but with the `type` column run through the jargon dictionary so the table shows 'medium tank' instead of '2tnk'. Engine-side ids are unchanged — the click-to-select path still works.""" df = _play_units_df(sess, sel) if df is None or not len(df) or "type" not in df.columns: return df try: from openra_bench.playlist import simplify_text df = df.copy() df["type"] = df["type"].map( lambda t: simplify_text(str(t)) if t is not None else t ) return df except Exception: # noqa: BLE001 return df def _playlist_summary_df(st: dict): """Session-end summary table — one row per game played so far. Used in the 'Session summary' panel that appears once the playlist is complete (or the player clicks 'End session').""" rows = list(st.get("results") or []) if not rows: return pd.DataFrame(columns=[ "Game", "Scenario", "Level", "Outcome", "Turns", "Max Turns", ]) return pd.DataFrame(rows) def _playlist_render(st: dict, prev_sess): """Open a Playlist session for `st`'s current cell and render it. Mirrors `_study_render` but produces the simplified Playlist panel set.""" if prev_sess is not None: try: prev_sess.close() except Exception: # noqa: BLE001 pass empty_units = _play_units_df(None, []) pl = st.get("playlist") or [] idx = int(st.get("idx", 0) or 0) if idx >= len(pl): # Playlist complete — return the summary view, no session. return ( None, [], [], "", None, "", "_session complete_", empty_units, st, _playlist_progress_md(st), _playlist_summary_df(st), gr.update(visible=True), gr.update(visible=False), ) pack, level = pl[idx] try: from openra_bench.human_labeling import InteractiveSession sess = InteractiveSession.from_pack( pack, level, seed=1, player=st.get("player", "anon"), ) except Exception as e: # noqa: BLE001 # Engine wheel missing — friendly message, never crash. msg = ( f"⚠️ Could not start `{pack}:{level}` — `{e}`.\n\n" "Tip: from the repo root, run " "`cd OpenRA-Rust && PATH=$HOME/.cargo/bin:/opt/anaconda3/bin:$PATH " "maturin develop --release` first to install the engine wheel." ) return ( None, [], [], msg, None, "", "_engine wheel not installed — see the message above_", empty_units, st, _playlist_progress_md(st), _playlist_summary_df(st), gr.update(visible=False), gr.update(visible=False), ) img, _brief, _status, _units = _play_render(sess, [], [], False) return ( sess, [], [], _playlist_objective_md(sess), img, _playlist_details_md(sess), _playlist_status_md(sess, st), _playlist_simplified_units_df(sess, []), st, _playlist_progress_md(st), _playlist_summary_df(st), gr.update(visible=False), # summary panel hidden during play gr.update(visible=_playlist_should_show_build(sess)), ) def _playlist_start(prev_sess, player): """The single 'Start playlist' click — set up state, open game 1.""" st = _playlist_state(player) return _playlist_render(st, prev_sess) def _playlist_record_outcome(sess, st: dict) -> dict: """If the current session has finished and we haven't recorded its result yet, append it to `st['results']`. Returns the updated state dict (may be the same object).""" if sess is None: return st if not getattr(sess, "done", False): return st pl = st.get("playlist") or [] idx = int(st.get("idx", 0) or 0) results = list(st.get("results") or []) if idx >= len(pl) or idx < len(results): return st # already recorded for this idx pack, level = pl[idx] from openra_bench.playlist import session_summary_row results.append(session_summary_row( idx=idx, pack=pack, level=level, outcome=str(getattr(sess, "outcome", "draw") or "draw"), turns=int(getattr(sess, "turn", 0) or 0), max_turns=int(getattr(sess, "max_turns", 0) or 0), )) new_st = dict(st) new_st["results"] = results if new_st.get("done_at") is None: new_st["done_at"] = time.time() return new_st def _playlist_click(sess, sel, queue, show_obj, evt): """Wrapper around `_play_click` that swaps the unit-table output for the jargon-simplified version. Matches `_play_click`'s 5-output signature so it drops in to the same `pl_img.select` wiring.""" new_sel, new_queue, brief, _units, img = _play_click( sess, sel, queue, show_obj, evt ) return ( new_sel, new_queue, brief, _playlist_simplified_units_df(sess, new_sel), img, ) def _playlist_clear_selection(sess, queue, show_obj): sel, brief, _units, img = _play_clear_selection(sess, queue, show_obj) return sel, brief, _playlist_simplified_units_df(sess, sel), img def _playlist_clear_queue(sess, sel, show_obj): queue, brief, _units, img = _play_clear_queue(sess, sel, show_obj) return queue, brief, _playlist_simplified_units_df(sess, sel), img def _playlist_build_item(sess, sel, queue, item, show_obj): queue, item_out, brief, _units, img = _play_build_item( sess, sel, queue, item, show_obj ) return ( queue, item_out, brief, _playlist_simplified_units_df(sess, sel), img, ) def _playlist_end_turn(sess, sel, queue, st): """End-turn for Playlist mode. After advancing the engine, if the game just ended, stamp `done_at` so the auto-advance countdown can fire.""" if sess is not None and not sess.done: try: sess.submit_turn(list(queue)) except Exception as e: # noqa: BLE001 logger.warning("playlist submit_turn failed: %s", e) img, _brief, _status, _units = _play_render(sess, [], [], False) new_st = _playlist_record_outcome(sess, st or {}) return ( sess, [], [], img, _playlist_details_md(sess), _playlist_status_md(sess, new_st), _playlist_simplified_units_df(sess, []), new_st, gr.update(visible=_playlist_should_show_build(sess)), ) def _playlist_advance(prev_sess, st): """Move to the next pack — fired by the auto-advance Timer or the manual 'Skip wait ▶' button.""" if not st or "playlist" not in st: return _playlist_render(_playlist_state(), prev_sess) new_st = dict(st) new_st["idx"] = int(new_st.get("idx", 0) or 0) + 1 new_st["done_at"] = None return _playlist_render(new_st, prev_sess) _PL_TICK_NOOP_OUTPUTS = 13 # length of _pl_render_outs def _playlist_tick(sess, st): """Auto-advance Timer tick. If the current game has been over for >= AUTO_ADVANCE_WAIT_SECONDS seconds, advance to the next pack; otherwise return state unchanged. The Timer fires every 1s while the tab is active.""" from openra_bench.playlist import ( AUTO_ADVANCE_WAIT_SECONDS, playlist_should_advance, ) noop = tuple(gr.skip() for _ in range(_PL_TICK_NOOP_OUTPUTS)) if not st or not isinstance(st, dict): return noop if sess is None or not getattr(sess, "done", False): return noop done_at = st.get("done_at") if not playlist_should_advance( True, done_at, time.time(), AUTO_ADVANCE_WAIT_SECONDS ): return noop return _playlist_advance(sess, st) def _playlist_submit_baseline(st): """Post the playlist's per-game outcomes to the bench raw-games log as `agent_type=Human`. Idempotent — a second click is a no-op.""" if not st: return "_no session to submit_", st if st.get("submitted"): return "_already submitted_", st rows = list(st.get("results") or []) if not rows: return "_no completed games yet — play at least one scenario_", st player = (st.get("player") or "anon").strip() or "anon" timestamp = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") saved = 0 for r in rows: outcome = str(r.get("Outcome", "DRAW")).lower() # The Submit tab consumes a richer JSON blob than we have here; # mirror its expected shape so the leaderboard pipeline picks # the rows up identically to a model run. try: _save_raw_game({ "agent_name": player, "agent_type": "Human", "opponent": "Beginner", "scenario": f"{r.get('Scenario')}:{r.get('Level')}", "result": outcome, "win": outcome == "win", "kills_cost": 0, "deaths_cost": 0, "assets_value": 0, "ticks": int(r.get("Turns", 0) or 0) * 90, "timestamp": timestamp, "replay_url": "", "agent_url": "", "hf_username": "", }) saved += 1 except Exception as e: # noqa: BLE001 logger.warning("playlist submit row failed: %s", e) new_st = dict(st) new_st["submitted"] = True return ( f"✅ Submitted {saved} game(s) for `{player}` as **Human** baseline. " f"They will appear in the leaderboard once aggregated.", new_st, ) def build_app() -> gr.Blocks: """Build the Gradio leaderboard app.""" initial_df = add_type_badges(load_data()) with gr.Blocks(title="OpenRA-Bench") as app: gr.Markdown( "# OpenRA-Bench\n" "**Agent Leaderboard for OpenRA-RL** — " "Train AI to Play Real-Time Strategy" ) with gr.Tabs(): # ── Leaderboard Tab ─────────────────────────────────────────── with gr.Tab("Leaderboard"): with gr.Row(): search_box = gr.Textbox( label="Search agents", placeholder="Search by name (supports regex, comma-separated)...", scale=3, ) type_filter = gr.CheckboxGroup( choices=["Scripted", "LLM", "RL"], value=["Scripted", "LLM", "RL"], label="Agent Type", scale=2, ) opponent_filter = gr.Dropdown( choices=["All", "Beginner", "Easy", "Medium", "Normal", "Hard"], value="All", label="Opponent", scale=1, ) show_unverified = gr.Checkbox( label="Show unverified", value=True, scale=1, ) leaderboard = gr.Dataframe( value=initial_df, datatype=[ "number", # Rank "html", # Agent (may contain hyperlink) "html", # Type (badge) "html", # Status (verified badge) "str", # Opponent "number", # Games "number", # Win Rate "number", # Score "number", # K/D Ratio "number", # Avg Kills "number", # Avg Deaths "number", # Avg Economy "number", # Avg Game Length "str", # Date "html", # Replay (download link) ], interactive=False, show_label=False, ) # Wire up filters filter_inputs = [search_box, type_filter, opponent_filter, show_unverified] for component in filter_inputs: component.change( fn=filter_leaderboard, inputs=filter_inputs, outputs=leaderboard, ) # ── Capability Leaderboard Tab ──────────────────────────────── # run_eval reports (composite + Perception/Reasoning/Action + # weakest link) published via `run_eval --leaderboard`. with gr.Tab("Capability Leaderboard"): gr.Markdown( "Models on customized scenarios, scored on the " "Perception→Reasoning→Action chain. **weakest_link** " "shows the dominant failure mode." ) cap_df = gr.Dataframe( value=load_capability_leaderboard(), interactive=False, wrap=True, ) refresh_cap = gr.Button("Refresh") refresh_cap.click(load_capability_leaderboard, outputs=cap_df) # ── Scenarios Tab ───────────────────────────────────────────── with gr.Tab("Scenarios"): gr.Markdown( "Browse every active scenario pack. Each pack tests " "one capability at three difficulty levels with " "identical win/fail rules used to score LLM agents. " "Switch to **中文** for Chinese translations." ) with gr.Row(): scen_search = gr.Textbox( label="Search", placeholder="Filter by id, title, or meaning...", scale=3, ) scen_cap_filter = gr.CheckboxGroup( choices=["perception", "reasoning", "action", "adversarial"], value=["perception", "reasoning", "action", "adversarial"], label="Capability", scale=3, ) scen_lang = gr.Radio( choices=["English", "中文"], value="English", label="Language", scale=1, ) scen_table = gr.Dataframe( value=_scenarios_filter("", [ "perception", "reasoning", "action", "adversarial" ]), interactive=False, wrap=True, show_label=False, ) scen_filter_inputs = [scen_search, scen_cap_filter] for comp in scen_filter_inputs: comp.change( fn=_scenarios_filter, inputs=scen_filter_inputs, outputs=scen_table, ) gr.Markdown("---") scen_id_input = gr.Textbox( label="Pack ID (click a row above or type)", placeholder="e.g. combat-focus-fire-priority", ) scen_detail = gr.Markdown( "_Select a scenario from the table above to see " "details._" ) def _scen_detail_with_lang(pack_id, lang_choice): lang = "zh" if lang_choice == "中文" else "en" return _scenarios_detail_md(pack_id, lang) scen_id_input.change( fn=_scen_detail_with_lang, inputs=[scen_id_input, scen_lang], outputs=scen_detail, ) scen_lang.change( fn=_scen_detail_with_lang, inputs=[scen_id_input, scen_lang], outputs=scen_detail, ) def _scen_row_select(evt: gr.SelectData, df, lang_choice): if evt is None or df is None or not len(df): return gr.update(), gr.update() try: row_idx = evt.index[0] pack_id = str(df.iloc[row_idx]["ID"]) lang = "zh" if lang_choice == "中文" else "en" return pack_id, _scenarios_detail_md(pack_id, lang) except Exception: # noqa: BLE001 return gr.update(), gr.update() scen_table.select( fn=_scen_row_select, inputs=[scen_table, scen_lang], outputs=[scen_id_input, scen_detail], ) # ── Battle Viewer Tab ───────────────────────────────────────── # Browse saved playbacks: filter run → model → scenario, # step the battle with ◀ / ▶, and compare two models # head-to-head on the same scenario+seed. with gr.Tab("Battle Viewer"): gr.Markdown( "Pick a **run → model → scenario**, then step the " f"battle. Playback root: `{PLAYBACK_ROOT}` " "(set `OPENRA_BENCH_PLAYBACK_ROOT` to change)." ) bv_idx = gr.State([]) bv_turn = gr.State(0) with gr.Row(): bv_run = gr.Dropdown(label="Run", scale=2) bv_model = gr.Dropdown(label="Model", scale=2) bv_scen = gr.Dropdown(label="Scenario @ seed", scale=3) bv_refresh = gr.Button("⟳ Rescan", scale=1) with gr.Row(): bv_compare = gr.Checkbox(label="Compare mode", value=False) bv_bsel = gr.Dropdown( label="B: run / model (same scenario+seed)", scale=3 ) with gr.Row(): bv_prev = gr.Button("◀ Prev turn") bv_pos = gr.Markdown("—") bv_next = gr.Button("Next turn ▶") with gr.Row(): with gr.Column(): bv_a_img = gr.Image( label="A minimap", height=320, show_label=True, interactive=False ) bv_a_md = gr.Markdown() with gr.Column(): bv_b_img = gr.Image( label="B minimap", height=320, show_label=True, interactive=False ) bv_b_md = gr.Markdown() _render_outs = [ bv_a_img, bv_a_md, bv_b_img, bv_b_md, bv_pos, bv_bsel ] _sel = [bv_run, bv_model, bv_scen] def _bv_go(idx, run, model, scen, turn, comp, b, delta=0): turn = max(0, (turn or 0) + delta) *outs, bupd = _bv_render( idx, run, model, scen, turn, comp, b ) return (*outs, bupd, turn) bv_refresh.click( bv_runs, outputs=[bv_idx, bv_run] ).then( bv_on_run, [bv_idx, bv_run], bv_model ).then( bv_on_model, [bv_idx, bv_run, bv_model], bv_scen ).then( _bv_go, [bv_idx, bv_run, bv_model, bv_scen, bv_turn, bv_compare, bv_bsel], [*_render_outs, bv_turn], ) bv_run.change(bv_on_run, [bv_idx, bv_run], bv_model).then( bv_on_model, [bv_idx, bv_run, bv_model], bv_scen ) bv_model.change( bv_on_model, [bv_idx, bv_run, bv_model], bv_scen ) for comp in (bv_scen, bv_compare, bv_bsel): comp.change( lambda i, r, m, s, c, b: _bv_go( i, r, m, s, 0, c, b), [bv_idx, bv_run, bv_model, bv_scen, bv_compare, bv_bsel], [*_render_outs, bv_turn], ) bv_prev.click( lambda i, r, m, s, t, c, b: _bv_go( i, r, m, s, t, c, b, -1), [bv_idx, bv_run, bv_model, bv_scen, bv_turn, bv_compare, bv_bsel], [*_render_outs, bv_turn], ) bv_next.click( lambda i, r, m, s, t, c, b: _bv_go( i, r, m, s, t, c, b, +1), [bv_idx, bv_run, bv_model, bv_scen, bv_turn, bv_compare, bv_bsel], [*_render_outs, bv_turn], ) app.load(bv_runs, outputs=[bv_idx, bv_run]).then( bv_on_run, [bv_idx, bv_run], bv_model ).then( bv_on_model, [bv_idx, bv_run, bv_model], bv_scen ) # ── Play Tab (human-labeling machine) ───────────────────────── # Play the exact scenarios LLM agents are scored on, by # clicking the minimap. Backed by InteractiveSession so a # human's run is graded by the identical win/fail rules — # human-vs-LLM comparison on one bench. with gr.Tab("Play"): gr.Markdown( "Play a scenario yourself — the same scenarios LLM " "agents are scored on. Pick a **scenario → level → " "seed**, click **Start**. Then, on the minimap: " "**click your own unit** to select it (click again " "to deselect, click several to build a group); with " "units selected, **click empty ground to move** " "them there, or **click an enemy to attack** it. " "**End Turn** advances. You are graded by the " "identical win/fail rules as the models." ) play_sess = gr.State(None) play_sel = gr.State([]) play_queue = gr.State([]) study_state = gr.State({}) with gr.Accordion( "📋 Human-study mode — 24-pack subset, 3 conditions", open=False, ): gr.Markdown( "For the **human-baseline study**. Enter your name " "and click **Begin study** — you'll be walked " "through 72 games (24 scenarios × fog / no-fog / " "handoff-deficit), counterbalanced per player. " "Play each to game-over, then **Next scenario ▶**. " "Every game auto-saves apples-to-apple with the " "model runs." ) with gr.Row(): study_player = gr.Textbox( label="Your name / id", scale=2, ) study_begin_btn = gr.Button("Begin study", scale=1) study_next_btn = gr.Button( "Next scenario ▶", variant="primary", scale=1, ) study_progress = gr.Markdown() with gr.Row(): play_scen = gr.Dropdown( choices=_play_scenarios(), label="Scenario", scale=3, ) play_level = gr.Dropdown( choices=_PLAY_LEVELS, value="easy", label="Level", scale=1, ) play_seed = gr.Number( value=1, label="Seed", precision=0, scale=1, ) play_start = gr.Button("▶ Start", scale=1) play_show_objectives = gr.Checkbox( label="Show objective rings", value=False, info="Only available when the scenario already reveals exact coordinates.", ) gr.Markdown( "_Press **Enter** to end the turn when focus is outside " "inputs._" ) play_objective = gr.Markdown() play_status = gr.Markdown(_play_status_md(None)) # Minimap on its own full-width row. No fixed height — # it renders at its natural aspect so there is no # letterbox band wasting space. play_img = gr.Image( label="Minimap — click your unit, then click where " "to send it", interactive=False, show_label=True, ) play_brief = gr.Markdown() play_units = gr.Dataframe( label="Your units (▶ = selected, shown at top)", headers=_PLAY_UNIT_COLS, interactive=False, wrap=True, ) with gr.Row(): play_build_item = gr.Textbox( label="Build item", value="e1", placeholder="e1, pbox, proc, powr, ...", scale=2, ) play_build_btn = gr.Button("Queue build", scale=1) with gr.Row(): play_clearsel_btn = gr.Button( "✖ Clear selected units", scale=1 ) play_clear_btn = gr.Button( "Cancel queued orders", scale=1 ) play_end_btn = gr.Button( "End Turn (Enter) ▶", variant="primary", scale=2, elem_id="play-end-turn-btn", ) play_start.click( _play_start, inputs=[ play_sess, play_scen, play_level, play_seed, play_show_objectives, ], outputs=[ play_sess, play_sel, play_queue, play_objective, play_img, play_brief, play_status, play_units, ], ) play_img.select( _play_click, inputs=[ play_sess, play_sel, play_queue, play_show_objectives, ], outputs=[ play_sel, play_queue, play_brief, play_units, play_img, ], ) play_build_btn.click( _play_build_item, inputs=[ play_sess, play_sel, play_queue, play_build_item, play_show_objectives, ], outputs=[ play_queue, play_build_item, play_brief, play_units, play_img, ], ) _study_outputs = [ play_sess, play_sel, play_queue, play_objective, play_img, play_brief, play_status, play_units, study_state, study_progress, ] study_begin_btn.click( _study_begin, inputs=[play_sess, study_player], outputs=_study_outputs, ) study_next_btn.click( _study_next, inputs=[play_sess, study_state], outputs=_study_outputs, ) play_end_btn.click( _play_end_turn, inputs=[ play_sess, play_sel, play_queue, play_show_objectives, ], outputs=[ play_sess, play_sel, play_queue, play_img, play_brief, play_status, play_units, ], ) play_clearsel_btn.click( _play_clear_selection, inputs=[play_sess, play_queue, play_show_objectives], outputs=[ play_sel, play_brief, play_units, play_img, ], ) play_clear_btn.click( _play_clear_queue, inputs=[play_sess, play_sel, play_show_objectives], outputs=[ play_queue, play_brief, play_units, play_img, ], ) play_show_objectives.change( _play_toggle_objectives, inputs=[ play_sess, play_sel, play_queue, play_show_objectives, ], outputs=play_img, ) app.load(fn=None, js=_PLAY_KEYBOARD_JS) # ── Playlist Tab (cold-start non-gamer UX) ─────────────────── # The Play tab is a sandbox / power-user UI: 210 packs in a # dropdown, full briefings, full tool palette. The Playlist # tab is the COLD-START non-gamer UX: one click to start, a # curated 20-pack list (no dropdown — the player never picks # a pack), a 1-2 sentence plain-English objective above the # minimap (no jargon), only move/attack/build/end-turn, an # auto-advance countdown when a game ends, and a session # summary with a 'Submit baseline' button at the end. # # Underneath this is the same `InteractiveSession` the Play # tab uses (and the 24-pack study uses), so the playback / # leaderboard format is apples-to-apple with model runs on # the same scenarios. with gr.Tab("Playlist"): gr.Markdown( "## Beginner Playlist — 20 scenarios, ~60-90 minutes\n\n" "A curated tour of the bench, hand-picked for " "non-gamers: visible green tanks, visible red " "enemies, drive units to yellow rings or attack " "what's in front of you. **No game knowledge " "needed.** Enter your name, hit **Start**, and play " "each scenario to game-over — the next one loads " "automatically.\n\n" "**How to play.** Click your green unit to select " "it (click again to deselect; click several to " "build a group). With units selected, click empty " "ground to **move** there, or click a red enemy " "to **attack** it. **End Turn** advances time. The " "minimap shows yellow rings where you need to go." ) pl_sess = gr.State(None) pl_sel = gr.State([]) pl_queue = gr.State([]) pl_state = gr.State({}) # A constant 'show objectives = False' state — reused by # `_play_click` / `_play_clear_*` so the Playlist tab can # share Play-tab handlers without the objective-rings # toggle (objectives are surfaced via the plain-English # text above the minimap, not as on-map rings). pl_show_obj_const = gr.State(False) pl_progress = gr.Markdown(_playlist_progress_md({})) with gr.Row(): pl_player = gr.Textbox( label="Your name", placeholder="e.g. Alex", scale=3, ) pl_start_btn = gr.Button( "▶ Start playlist", variant="primary", scale=1, ) pl_skip_btn = gr.Button( "Skip wait ▶", scale=1, visible=True, ) pl_objective = gr.Markdown() pl_status = gr.Markdown(_play_status_md(None)) pl_img = gr.Image( label="Minimap — click your green unit, then click " "where to send it (move) or click a red enemy (attack)", interactive=False, show_label=True, ) pl_units = gr.Dataframe( label="Your units (▶ = selected, shown at top)", headers=_PLAY_UNIT_COLS, interactive=False, wrap=True, ) with gr.Accordion("Details (advanced)", open=False): pl_details = gr.Markdown() with gr.Row(visible=False) as pl_build_row: pl_build_item = gr.Textbox( label="Build item", placeholder="e.g. e1, pbox", scale=2, ) pl_build_btn = gr.Button("Queue build", scale=1) with gr.Row(): pl_clearsel_btn = gr.Button( "✖ Clear selected units", scale=1, ) pl_clear_btn = gr.Button( "Cancel queued orders", scale=1, ) pl_end_btn = gr.Button( "End Turn ▶", variant="primary", scale=2, ) # Session-end summary panel — hidden until the playlist # finishes (or `_playlist_render` toggles it). with gr.Group(visible=False) as pl_summary_panel: gr.Markdown("### 📊 Session summary") pl_summary_df = gr.Dataframe( headers=["Game", "Scenario", "Level", "Outcome", "Turns", "Max Turns"], interactive=False, wrap=True, ) pl_submit_btn = gr.Button( "Submit baseline", variant="primary", ) pl_submit_msg = gr.Markdown() # Background timer — fires every 1s while the tab is # active. When the current game is over and the # 5-second wait has elapsed, advance to the next pack. pl_timer = gr.Timer(1.0, active=True) _pl_render_outs = [ pl_sess, pl_sel, pl_queue, pl_objective, pl_img, pl_details, pl_status, pl_units, pl_state, pl_progress, pl_summary_df, pl_summary_panel, pl_build_row, ] pl_start_btn.click( _playlist_start, inputs=[pl_sess, pl_player], outputs=_pl_render_outs, ) pl_skip_btn.click( _playlist_advance, inputs=[pl_sess, pl_state], outputs=_pl_render_outs, ) pl_timer.tick( _playlist_tick, inputs=[pl_sess, pl_state], outputs=_pl_render_outs, ) # Click on minimap → reuse the Play-tab handler (same # selection / attack / move semantics). The Playlist tab # never shows objective rings (objectives are surfaced # via the plain-English text above the map). pl_img.select( _playlist_click, inputs=[pl_sess, pl_sel, pl_queue, pl_show_obj_const], outputs=[pl_sel, pl_queue, pl_details, pl_units, pl_img], ) pl_end_btn.click( _playlist_end_turn, inputs=[pl_sess, pl_sel, pl_queue, pl_state], outputs=[ pl_sess, pl_sel, pl_queue, pl_img, pl_details, pl_status, pl_units, pl_state, pl_build_row, ], ) pl_clearsel_btn.click( _playlist_clear_selection, inputs=[pl_sess, pl_queue, pl_show_obj_const], outputs=[pl_sel, pl_details, pl_units, pl_img], ) pl_clear_btn.click( _playlist_clear_queue, inputs=[pl_sess, pl_sel, pl_show_obj_const], outputs=[pl_queue, pl_details, pl_units, pl_img], ) pl_build_btn.click( _playlist_build_item, inputs=[ pl_sess, pl_sel, pl_queue, pl_build_item, pl_show_obj_const, ], outputs=[ pl_queue, pl_build_item, pl_details, pl_units, pl_img, ], ) pl_submit_btn.click( _playlist_submit_baseline, inputs=[pl_state], outputs=[pl_submit_msg, pl_state], ) # ── About Tab ───────────────────────────────────────────────── with gr.Tab("About"): gr.Markdown(ABOUT_MD) # ── Submit Tab ──────────────────────────────────────────────── with gr.Tab("Submit"): gr.Markdown( "## Upload Results\n\n" "Upload a bench export JSON from your OpenRA-RL game. " "After each game, the agent saves a JSON file to " "`~/.openra-rl/bench-exports/`." ) with gr.Row(): json_upload = gr.File( label="Bench export JSON", file_types=[".json"], scale=3, ) replay_upload = gr.File( label="Replay file (optional)", file_types=[".orarep"], scale=2, ) submit_btn = gr.Button("Submit Results", variant="primary") submit_output = gr.Markdown() submit_btn.click( fn=handle_upload, inputs=[json_upload, replay_upload], outputs=[submit_output, leaderboard], ) # API endpoint for CLI auto-upload (JSON only) api_json_input = gr.Textbox(visible=False) api_result = gr.Textbox(visible=False) api_btn = gr.Button(visible=False) api_btn.click( fn=handle_api_submit, inputs=[api_json_input], outputs=[api_result], api_name="submit", ) # API endpoint for CLI upload with replay api_json_input2 = gr.Textbox(visible=False) api_replay_input = gr.File(visible=False) api_result2 = gr.Textbox(visible=False) api_btn2 = gr.Button(visible=False) api_btn2.click( fn=handle_api_submit_with_replay, inputs=[api_json_input2, api_replay_input], outputs=[api_result2], api_name="submit_with_replay", ) gr.Markdown(SUBMIT_MD) return app if __name__ == "__main__": app = build_app() app.launch( allowed_paths=[str(SUBMISSIONS_DIR), str(PLAYBACK_ROOT)], )