147 lines
5 KiB
Python
Executable file
147 lines
5 KiB
Python
Executable file
#!/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())
|