engine.py

#!/usr/bin/env python3
"""
LightCycles Game Engine
=======================

A deterministic, tick-based Tron light-cycles arena for CodeClash.

Each player drives a cycle around a bordered grid, leaving a solid trail behind
it. Every tick, all cycles move one cell simultaneously. If a cycle would move
into a wall, any trail (its own or an opponent's), off the grid, or into the same
cell as another cycle (a head-on), it crashes and is eliminated. The last cycle
riding wins; if everyone still alive crashes on the same tick, it's a draw.

The simulation is fully deterministic given (seed, bot code) and the whole grid
state is handed to each bot every tick, so a strong bot can look ahead / flood-fill
to find the move that keeps the most space open. Strategy lives in the code.

Usage:
    python engine.py /path/to/p1/main.py /path/to/p2/main.py -r NUM_GAMES -o OUTPUT_DIR

The bot interface (see README.md and main.py):

    def get_move(obs: dict) -> str:
        # return one of: "N" "S" "E" "W"  (up / down / right / left)
"""

from __future__ import annotations

import argparse
import importlib.util
import json
import math
import os
import random
import signal
import sys
from dataclasses import dataclass
from typing import Callable

# --------------------------------------------------------------------------------------
# Game constants
# --------------------------------------------------------------------------------------

WIDTH = 48                   # grid width in cells
HEIGHT = 36                  # grid height in cells
MAX_TICKS = 2000             # safety cap; games almost always end earlier
TURN_TIMEOUT = 0.10          # seconds a single get_move call may take before -> continue

ROCK_DENSITY = 0.04          # fraction of the board seeded with rock obstacles (0 = open)
ROCK_CLEARANCE = 3           # keep rocks at least this many cells from every spawn

EMPTY = -1                   # an empty grid cell
ROCK = -2                    # a static rock obstacle (solid, like a wall)

# Direction vectors (origin top-left; N = up = -y). Also the set of valid actions.
DIRS: dict[str, tuple[int, int]] = {"N": (0, -1), "S": (0, 1), "E": (1, 0), "W": (-1, 0)}
OPPOSITE = {"N": "S", "S": "N", "E": "W", "W": "E"}


@dataclass
class Cycle:
    pid: int
    x: int
    y: int
    dir: str
    alive: bool = True


# --------------------------------------------------------------------------------------
# Bot loading + sandboxed invocation
# --------------------------------------------------------------------------------------


class TimeoutError_(Exception):
    pass


def _on_alarm(signum, frame):  # noqa: ARG001
    raise TimeoutError_()


_HAS_ALARM = hasattr(signal, "SIGALRM")
if _HAS_ALARM:
    signal.signal(signal.SIGALRM, _on_alarm)

_module_counter = 0


def load_bot(path: str) -> Callable:
    """Import a bot module and return its get_move function."""
    global _module_counter
    _module_counter += 1
    module_name = f"bot_module_{_module_counter}"
    spec = importlib.util.spec_from_file_location(module_name, path)
    module = importlib.util.module_from_spec(spec)
    sys.modules[module_name] = module
    spec.loader.exec_module(module)
    if not hasattr(module, "get_move"):
        raise ValueError(f"Bot module {path} must define a get_move(obs) function")
    return module.get_move


def call_bot(fn: Callable, obs: dict, current_dir: str) -> str:
    """Call a bot's get_move with a crash/timeout guard. Any failure, an invalid move,
    or a reversal (into your own neck) falls back to continuing in the current direction."""
    if _HAS_ALARM:
        signal.setitimer(signal.ITIMER_REAL, TURN_TIMEOUT)
    try:
        move = fn(obs)
    except Exception:
        return current_dir
    finally:
        if _HAS_ALARM:
            signal.setitimer(signal.ITIMER_REAL, 0)

    if isinstance(move, str):
        move = move.strip().upper()
        if move in DIRS and move != OPPOSITE[current_dir]:
            return move
    return current_dir


# --------------------------------------------------------------------------------------
# Simulation
# --------------------------------------------------------------------------------------


class Game:
    def __init__(self, num_players: int, seed: int):
        self.n = num_players
        self.rng = random.Random(seed)
        self.tick = 0
        # grid[y][x] = owning player id (its trail/head), or EMPTY
        self.grid: list[list[int]] = [[EMPTY] * WIDTH for _ in range(HEIGHT)]

        # Start the cycles spread evenly around a circle, with a small seeded jitter so
        # sims aren't identical, each facing a seeded random direction.
        self.cycles: list[Cycle] = []
        cx, cy = WIDTH / 2.0, HEIGHT / 2.0
        rx, ry = WIDTH * 0.30, HEIGHT * 0.30
        for i in range(num_players):
            angle = 2 * math.pi * i / num_players
            x = int(round(cx + math.cos(angle) * rx + self.rng.uniform(-2, 2)))
            y = int(round(cy + math.sin(angle) * ry + self.rng.uniform(-2, 2)))
            x = max(1, min(WIDTH - 2, x))
            y = max(1, min(HEIGHT - 2, y))
            direction = self.rng.choice(list(DIRS.keys()))
            self.cycles.append(Cycle(pid=i, x=x, y=y, dir=direction))
            self.grid[y][x] = i

        self.rocks: list[tuple[int, int]] = []
        self._place_rocks()

    def _place_rocks(self) -> None:
        """Scatter rock obstacles with 180-degree rotational symmetry (fair for the
        head-to-head default), keeping clear of every spawn, and only accept a layout
        that leaves all spawns mutually reachable so nobody starts boxed in. Seeded, so
        each sim gets a different-but-reproducible map. Density 0 leaves the board open."""
        if ROCK_DENSITY <= 0:
            return
        spawn_clear = {
            (c.x + dx, c.y + dy)
            for c in self.cycles
            for dx in range(-ROCK_CLEARANCE, ROCK_CLEARANCE + 1)
            for dy in range(-ROCK_CLEARANCE, ROCK_CLEARANCE + 1)
        }
        interior = [
            (x, y)
            for x in range(2, WIDTH - 2)
            for y in range(2, HEIGHT - 2)
            if (x, y) not in spawn_clear
        ]
        target = int(ROCK_DENSITY * WIDTH * HEIGHT)

        for _ in range(20):  # retry until we get a connected (fair) layout
            candidates = interior[:]
            self.rng.shuffle(candidates)
            rocks: set[tuple[int, int]] = set()
            for x, y in candidates:
                if len(rocks) >= target:
                    break
                mx, my = WIDTH - 1 - x, HEIGHT - 1 - y  # 180-degree partner
                if (x, y) in rocks or (mx, my) in rocks or (mx, my) in spawn_clear:
                    continue
                rocks.add((x, y))
                rocks.add((mx, my))
            if self._all_spawns_connected(rocks):
                break
        else:
            rocks = set()  # give up on rocks rather than ship an unfair/boxed-in map

        self.rocks = sorted(rocks)
        for x, y in self.rocks:
            self.grid[y][x] = ROCK

    def _all_spawns_connected(self, rocks: set[tuple[int, int]]) -> bool:
        """Flood-fill from the first spawn over non-rock cells; every spawn must be reachable."""
        start = (self.cycles[0].x, self.cycles[0].y)
        seen = {start}
        stack = [start]
        while stack:
            x, y = stack.pop()
            for dx, dy in DIRS.values():
                nx, ny = x + dx, y + dy
                if 0 <= nx < WIDTH and 0 <= ny < HEIGHT and (nx, ny) not in rocks and (nx, ny) not in seen:
                    seen.add((nx, ny))
                    stack.append((nx, ny))
        return all((c.x, c.y) in seen for c in self.cycles)

    def observation(self, me: int) -> dict:
        return {
            "tick": self.tick,
            "max_ticks": MAX_TICKS,
            "width": WIDTH,
            "height": HEIGHT,
            "you": me,
            "players": [
                {"id": c.pid, "x": c.x, "y": c.y, "dir": c.dir, "alive": c.alive}
                for c in self.cycles
            ],
            # grid[y][x]: owning player id (>=0), -1 empty, or -2 rock. Off-grid is a wall.
            "grid": [row[:] for row in self.grid],
        }

    def alive_cycles(self) -> list[Cycle]:
        return [c for c in self.cycles if c.alive]

    def step(self, moves: dict[int, str]) -> None:
        """Apply one simultaneous tick. `moves` maps player id -> chosen direction."""
        targets: dict[int, tuple[int, int]] = {}
        for c in self.alive_cycles():
            c.dir = moves[c.pid]
            dx, dy = DIRS[c.dir]
            targets[c.pid] = (c.x + dx, c.y + dy)

        # Count target cells to detect head-on collisions (2+ cycles into one cell).
        counts: dict[tuple[int, int], int] = {}
        for t in targets.values():
            counts[t] = counts.get(t, 0) + 1

        crashed: set[int] = set()
        for pid, (tx, ty) in targets.items():
            off_grid = tx < 0 or tx >= WIDTH or ty < 0 or ty >= HEIGHT
            # grid already holds every head + trail, so moving into a cell an opponent is
            # vacating (it leaves a trail) or swapping places both count as a crash here.
            into_solid = (not off_grid) and self.grid[ty][tx] != EMPTY
            if off_grid or into_solid or counts[(tx, ty)] > 1:
                crashed.add(pid)

        for c in self.alive_cycles():
            if c.pid in crashed:
                c.alive = False
            else:
                c.x, c.y = targets[c.pid]
                self.grid[c.y][c.x] = c.pid

    def territory(self) -> dict[int, int]:
        counts = {i: 0 for i in range(self.n)}
        for row in self.grid:
            for v in row:
                if v >= 0:  # player trails only (not EMPTY, not ROCK)
                    counts[v] += 1
        return counts

    def frame(self) -> dict:
        """A compact replay frame: just each cycle's head + alive flag. The renderer
        rebuilds the trails by accumulating head positions across frames."""
        return {
            "t": self.tick,
            "heads": [[c.x, c.y, 1 if c.alive else 0] for c in self.cycles],
        }


def run_game(bot_paths: list[str], seed: int) -> dict:
    """Run a single game. Returns winner (player id or None for draw) + replay."""
    n = len(bot_paths)

    bots: list[Callable | None] = []
    load_errors: dict[int, str] = {}
    for i, path in enumerate(bot_paths):
        try:
            bots.append(load_bot(path))
        except Exception as e:
            bots.append(None)
            load_errors[i] = str(e)

    game = Game(num_players=n, seed=seed)
    frames = [game.frame()]
    winner: int | None = None

    # A bot that failed to import can't steer -> it just goes straight and soon crashes.
    for t in range(1, MAX_TICKS + 1):
        game.tick = t
        moves: dict[int, str] = {}
        for c in game.alive_cycles():
            fn = bots[c.pid]
            moves[c.pid] = c.dir if fn is None else call_bot(fn, game.observation(c.pid), c.dir)
        game.step(moves)
        frames.append(game.frame())

        alive = game.alive_cycles()
        if len(alive) <= 1:
            if len(alive) == 1:
                winner = alive[0].pid
            # 0 alive -> everyone crashed together -> draw (winner stays None)
            break
    else:
        # Hit the tick cap with multiple survivors: most territory wins, ties draw.
        terr = game.territory()
        alive_ids = [c.pid for c in game.alive_cycles()]
        best = max((terr[i] for i in alive_ids), default=0)
        leaders = [i for i in alive_ids if terr[i] == best]
        winner = leaders[0] if len(leaders) == 1 else None

    return {
        "winner": winner,
        "load_errors": load_errors,
        "num_ticks": game.tick,
        "territory": game.territory(),
        "replay": {
            "width": WIDTH,
            "height": HEIGHT,
            "num_players": n,
            "rocks": [list(r) for r in game.rocks],
            "frames": frames,
        },
    }


# --------------------------------------------------------------------------------------
# CLI / tournament driver
# --------------------------------------------------------------------------------------


def write_replay(result: dict, game_num: int, bot_paths: list[str], output_dir: str) -> None:
    rp = dict(result["replay"])
    rp["names"] = [
        os.path.basename(os.path.dirname(p)) or f"player{i + 1}" for i, p in enumerate(bot_paths)
    ]
    rp["winner"] = result["winner"]
    with open(os.path.join(output_dir, f"sim_{game_num}.json"), "w") as f:
        json.dump(rp, f)


def main() -> None:
    parser = argparse.ArgumentParser(description="LightCycles Game Engine")
    parser.add_argument("bots", nargs="+", help="Paths to bot files (main.py)")
    parser.add_argument("-r", "--rounds", type=int, default=10, help="Number of games")
    parser.add_argument("-o", "--output-dir", type=str, default=None, help="Replay output dir")
    parser.add_argument("-s", "--seed", type=int, default=0, help="Base RNG seed")
    args = parser.parse_args()

    bot_paths = args.bots
    n = len(bot_paths)
    names = [
        os.path.basename(os.path.dirname(p)) or f"player{i + 1}" for i, p in enumerate(bot_paths)
    ]

    wins = {i: 0 for i in range(n)}
    draws = 0

    print(f"Running {args.rounds} games between:")
    for i, p in enumerate(bot_paths):
        print(f"  Player {i + 1}: {p}  ({names[i]})")
    print()

    if args.output_dir:
        os.makedirs(args.output_dir, exist_ok=True)

    for g in range(args.rounds):
        result = run_game(bot_paths, seed=args.seed + g)
        if result["winner"] is None:
            draws += 1
            wtxt = "draw"
        else:
            wins[result["winner"]] += 1
            wtxt = f"Player {result['winner'] + 1}"
        print(f"Game {g + 1}: {wtxt} (after {result['num_ticks']} ticks)")
        if result["load_errors"]:
            for i, err in result["load_errors"].items():
                print(f"    (Player {i + 1} failed to load: {err})")
        if args.output_dir:
            write_replay(result, g, bot_paths, args.output_dir)

    print()
    print("FINAL_RESULTS")
    for i in range(n):
        print(f"Bot_{i + 1}: {wins[i]} games won ({names[i]})")
    print(f"Draws: {draws}")


if __name__ == "__main__":
    main()