跳到主要内容

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()

这里有两点要记住:

  1. move()harvest() 只会修改内存里正在构造的计划,不会发网络请求。
  2. 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: ")))

同步和异步客户端的状态、单位接口完全一样,只有迭代、提交和关闭的写法不同:

同步异步
ArenaHeroClientAsyncArenaHeroClient
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.coreNone。给 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",
)

接下来读什么