magicciv/scripts/player-api-example.py

#!/usr/bin/env python3
"""Example: drive the Magic Civilization Player API from a plain Python client.

This script exists as documentation-by-example: it shows that
`scripts/player-api-server.sh` is not Claude-specific. Any client that
speaks JSON-Lines over stdin/stdout can take a player slot — Claude Code
via `tooling/claude-player-mcp/`, an OpenSpiel adapter pumping the same
pipe, or this 80-line random/greedy bot.

Wire-protocol contract: `src/game/engine/docs/PLAYER_API.md`.

Usage:
    python3 scripts/player-api-example.py [TURNS]

Spawns the harness, plays `TURNS` turns (default 10) with a trivial
greedy policy, prints a one-line summary per turn, then shuts the
harness down cleanly.
"""
from __future__ import annotations

import json
import os
import subprocess
import sys
from pathlib import Path
from typing import Any

REPO_ROOT = Path(__file__).resolve().parent.parent
HARNESS = REPO_ROOT / "scripts" / "player-api-server.sh"


class PlayerApiClient:
    """Thin JSON-Lines client over a child process. Stateless except for
    the request-id counter — the harness holds the simulator state."""

    def __init__(self, env_overrides: dict[str, str] | None = None) -> None:
        env = {**os.environ, **(env_overrides or {})}
        self._proc = subprocess.Popen(
            ["bash", str(HARNESS)],
            stdin=subprocess.PIPE,
            stdout=subprocess.PIPE,
            stderr=subprocess.DEVNULL,
            cwd=str(REPO_ROOT),
            text=True,
            bufsize=1,
            env=env,
        )
        self._next_id = 1

    def _send(self, msg: dict[str, Any]) -> dict[str, Any] | None:
        msg["id"] = self._next_id
        self._next_id += 1
        assert self._proc.stdin is not None and self._proc.stdout is not None
        self._proc.stdin.write(json.dumps(msg) + "\n")
        self._proc.stdin.flush()
        # Read until correlated response (skip async notifications).
        for _ in range(5000):
            line = self._proc.stdout.readline()
            if not line:
                return None
            try:
                obj = json.loads(line)
            except json.JSONDecodeError:
                continue
            if obj.get("id") == msg["id"]:
                return obj
        return None

    def view(self) -> dict[str, Any] | None:
        r = self._send({"type": "view"})
        return r["view"] if r and r.get("ok") else None

    def act(self, action: dict[str, Any]) -> bool:
        r = self._send({"type": "act", "action": action})
        return bool(r and r.get("ok"))

    def end_turn(self) -> bool:
        return self.act({"type": "end_turn"})

    def shutdown(self) -> None:
        self._send({"type": "shutdown"})
        try:
            self._proc.wait(timeout=5)
        except subprocess.TimeoutExpired:
            self._proc.kill()


def greedy_policy(view: dict[str, Any]) -> list[dict[str, Any]]:
    """Pick a small bundle of actions for this turn — found if possible,
    fortify warriors, queue a worker in any city with an empty queue."""
    actions: list[dict[str, Any]] = []
    for u in view.get("units", []):
        if u.get("owner") != view.get("player"):
            continue
        legal = [a["action"] for a in u.get("legal_actions", [])]
        # 1. Found city if available (founder)
        for a in legal:
            if a["type"] == "found_city":
                actions.append(a)
                break
        else:
            # 2. Fortify warriors
            for a in legal:
                if a["type"] == "fortify" and not u.get("fortified"):
                    actions.append(a)
                    break
    for c in view.get("cities", []):
        if c.get("production_queue"):
            continue
        for a in (a["action"] for a in c.get("legal_actions", [])):
            if a.get("type") == "queue_production" and a.get("item") == "worker":
                actions.append(a)
                break
    return actions


def main() -> int:
    turns = int(sys.argv[1]) if len(sys.argv) > 1 else 10
    client = PlayerApiClient(env_overrides={"CP_SEED": "42", "CP_PLAYERS": "2"})
    try:
        for t in range(1, turns + 1):
            view = client.view()
            if view is None:
                print(f"[t{t}] view failed; aborting")
                return 1
            score = view.get("score", {})
            res = view.get("resources", {})
            print(
                f"[t{t:3d}] cities={int(score.get('city_count', 0))} "
                f"units={int(score.get('unit_count', 0))} "
                f"gold={int(res.get('gold', 0))} "
                f"score={int(score.get('score_estimate', 0))}"
            )
            for action in greedy_policy(view):
                if not client.act(action):
                    print(f"        action failed: {action}")
                    break
            if not client.end_turn():
                print(f"[t{t}] end_turn failed; aborting")
                return 1
    finally:
        client.shutdown()
    return 0


if __name__ == "__main__":
    sys.exit(main())