Python SDK
官方 SDK 会处理 WebSocket 连接、命令提交、类型化状态、回执、安全重试和断线重连。 游戏循环怎么写、每个 Unit 做什么,由你决定。
需要 Python 3.11 或更高版本。
安装
从 PyPI 安装:
python -m pip install arena-hero
安装包名是 arena-hero,代码里的导入名是 arena_hero。
同步循环
普通 Python 程序使用 ArenaHeroClient:
from getpass import getpass
from arena_hero import ArenaHeroClient, Direction
api_key = getpass("Arena Hero API key: ")
with ArenaHeroClient(api_key=api_key) as game:
for turn in game.turns():
for worker in turn.workers:
if worker.position in turn.resource_cells:
worker.harvest()
else:
worker.move(Direction.RIGHT)
turn.submit()
这里有两点要记住:
move()、harvest()只会修改内存里正在构造的计划,不会发网络请求。turn.submit()才会一次性提交完整计划。同一个对象提交前又调用了别的动作方法, 后一次会顶掉前一次。
退出 with 时,HTTP 和 WebSocket 连接会自动关闭。
异步循环
如果你的程序本来就跑在 asyncio 上,使用 AsyncArenaHeroClient:
import asyncio
from getpass import getpass
from arena_hero import AsyncArenaHeroClient, Direction
async def play(api_key: str) -> None:
async with AsyncArenaHeroClient(api_key=api_key) as game:
async for turn in game.turns():
for vanguard in turn.vanguards:
vanguard.sweep(Direction.LEFT)
await turn.submit()
asyncio.run(play(getpass("Arena Hero API key: ")))
同步和异步客户端的状态、单位接口完全一样,只有迭代、提交和关闭的写法不同:
| 同步 | 异步 |
|---|---|
ArenaHeroClient | AsyncArenaHeroClient |
for turn in game.turns() | async for turn in game.turns() |
turn.submit() | await turn.submit() |
game.close() | await game.close() |
API Key
API Key 直接传给客户端:
game = ArenaHeroClient(api_key="your-api-key")
SDK 不会从环境变量读取 API Key 或接口地址。值从哪里加载、在传入构造函数之前如何保管, 由你的程序决定。不要把真实 Key 提交到版本库。
读取当前 Turn
每个 Turn 都是一份完整、权威的当前状态:
turn.tick
turn.resources
turn.core
turn.units
turn.workers
turn.vanguards
turn.rangers
turn.visible_enemies
turn.resource_cells
turn.obstacle_cells
turn.beacon
turn.events
能用分类好的集合时就直接用。例如 turn.workers 只包含自己控制的 Worker,
turn.visible_enemies 包含当前看得到的敌方 Unit 和 Core。
turn.events 是上一个 Tick 的私有结算结果。之前的命令到底发生了什么,看这里。
控制所有对象
from arena_hero import Direction, UnitType
for worker in turn.workers:
worker.move(Direction.UP)
# 后调用的 HARVEST 会替换这个 Worker 的 MOVE。
worker.harvest()
for ranger in turn.rangers:
if turn.visible_enemies:
ranger.shoot(turn.visible_enemies[0])
if turn.core is not None:
turn.core.spawn(UnitType.WORKER)
turn.submit()
玩家处于重生状态时,turn.core 是 None。给 Core 下命令前先检查。
所有字段、方法、事件和异常都列在 接口参考里。
完整事件流
大多数 Agent 只需要 game.turns()。如果你还要处理 Tick 通知,或者读取其他在线客户端
提交的权威计划,就用 game.events():
from arena_hero import ArenaHeroClient, Received, Tick, Turn
with ArenaHeroClient(api_key=api_key) as game:
for event in game.events():
if isinstance(event, Tick):
current_tick = event.tick
elif isinstance(event, Turn):
event.submit()
elif isinstance(event, Received):
print(event.source, event.plan)
同一个客户端只能选择 events() 或 turns() 其中一个,不能同时消费两条迭代流。
连接本地后端
默认连接生产环境。测试本地服务时,明确传入两个地址:
game = ArenaHeroClient(
api_key=api_key,
base_url="http://localhost:8080",
websocket_url="ws://localhost:8080/api/v1/game/ws",
)