← Bot development
Reference bot: simple_ai
A complete, competitive bot you can read end to end: a faithful Python port of the game's built-in AI. It produces counter-units, buys upgrades on a fixed plan, garrisons its base, captures and contests resource nodes, hunts stars with tier-sized squads, launches 15-unit attack waves, raids with a Triangle harass squad, scouts the map, fetches dropped items, and kites ranged units away from melee. In fair (no-handicap) evaluation it goes toe-to-toe with the built-in AI it ports.
Running it
The script uses the pachinko-pybind direct API (PachinkoEnv.get_state() / step_direct()), which runs full games locally against the built-in AI — the same engine the live servers run, so it's the fastest way to iterate on strategy before taking a bot online.
pip install pachinko-pybind
python simple_ai_direct.py [seed]What to study
Playing fair inside fog of war
Fog of war hides most of the map, and the observation only reports what you can currently see. The interesting part of this bot is how it plays well anyway — the techniques carry over directly to online bots:
- Building memory— enemy buildings are static, so the bot remembers every one it has seen and forgets an entry only once a friendly unit gets close enough to prove it's gone.
- Order tracking— the observation space exposes
has_targetbut not the order's destination, so the bot records the last order it issued per unit and never re-sends an identical command. This matters: command budgets are rate-limited. - Derived base position— the spawn point isn't in the observation, but the centroid of your starting buildings is the same thing.
Adapting it for online play
The strategy layer (squads, garrison logic, target scoring, upgrade plan) is plain Python and transfers as-is to a live pachinko_sdk bot. Two mechanical differences when porting: the online SDK uses string enums ("square", "squarracks", "spawn_rate") where the direct API uses ints and "SpawnRate"-style names, and actions are dataclasses (AttackMove(unit_id, x, y)) rather than command dicts. See the API reference for the online equivalents.
Full source
"""
simple_ai_direct.py — Python port of the game's built-in AI, written against
the pachinko_pybind *direct API* (get_state / step_direct).
This is the reference example for a "fairly decent" hand-coded bot: it mirrors
the built-in AI's strategy section by section, using only what the observation
space exposes.
Strategy overview (same as the built-in AI):
- Early game: produce Triangles (cavalry) to grab resource nodes fast.
- Production: counter the most common *visible* enemy unit type, with a
120-tick hysteresis lock; expand to a second building when
rich enough; all three buildings when rich or under pressure.
- Upgrades: fixed interleaved plan (mining + weapon damage first), rich
mode buys broadly; Healing Aura bought when >10 units injured.
- Garrison: threat-scaled defensive force at base; intercepts attackers.
- Nodes: 2 units per own/neutral node, 4 to contest enemy nodes,
budgeted at 20% of the army; plus up to 4 solo Triangle scouts.
- Stars: hunted with tier-sized squads (5/12/24); medium/strong only
when enough units are staged; hunt radius grows over time.
- Main wave: 15-unit squads launched at the least-defended enemy military
building; Circles hold a backline position and kite melee.
- Harass: 6 Triangles raid a second, least-defended target.
- Scout: one unit orbits 4 waypoints around the map centre.
- Items: up to 3 hunters fetch visible dropped items, skipping
dangerous ones.
- Retreat: units below 25% HP in combat run home.
Fog of war hides most of the map, so a few techniques are worth copying into
your own bot:
* Building memory — enemy buildings never move, so we cache every one we
have seen and forget an entry only once one of our units gets close enough
that we *would* see it and it is gone. Stars are cached the same way.
* Base position — derived once as the centroid of our starting buildings
(they never move either).
* Order dedup — the API exposes `has_target` but not where an order points,
so we remember the last order WE issued per unit and never re-send an
identical command. Commands are rate-limited, so this matters.
* Visible counts — pressure / overwhelm checks can only count enemies we
currently see, so they are tuned conservatively.
There is no betting phase in pybind games, so this file is purely the RTS
brain. Run it head-to-head against the built-in AI with:
pip install pachinko-pybind
python simple_ai_direct.py [seed]
"""
from __future__ import annotations
import math
# ---------------------------------------------------------------------------
# Tuning constants — matched to the built-in AI
# ---------------------------------------------------------------------------
# Production
ARMY_CAP = 300
RICH_ARMY_CAP = 400
RICH_THRESHOLD = 2500
EXPAND_THRESHOLD = 400
EXPAND_MIN_TICK = 120
COUNTER_LOCK_TICKS = 120
EARLY_CAVALRY_TICKS = 144
# Upgrades
UPGRADE_RESERVE = 600
RICH_UPGRADE_LIMIT = 4
# Staging / cadence
STAGING_RADIUS = 250.0
AI_TICK_INTERVAL = 4
# Main attack squad
MAIN_SQUAD_SIZE = 15
OVERWHELM_RATIO = 3
OVERWHELM_MIN_UNITS = 8
WAVE_STALL_TIMEOUT = 2400
HARASS_STALL_TIMEOUT = 960
WAVE_DEPLETED_THRESHOLD = 5
BASE_RECALL_ENEMY_COUNT = 6
# Harass squad
HARASS_SQUAD_SIZE = 6
HARASS_MIN_ARMY = 30
HARASS_MIN_TICK = 360
# Defense
GARRISON_BASE_SIZE = 8
GARRISON_MAX_SIZE = 25
GARRISON_LEAVE_FREE = 4
GARRISON_DEFEND_RADIUS = 220.0
GARRISON_THREAT_SCAN_RADIUS = 400.0
GARRISON_PER_THREAT = 0.6
GARRISON_INTERCEPT_RADIUS = GARRISON_DEFEND_RADIUS * 1.5 # ~330
# Retreat
RETREAT_HP_FRACTION = 0.25
# Resource nodes. Garrison units sit at 96 — comfortably inside the node's
# capture zone, so wandering a few steps never drops control.
GARRISON_PER_NODE = 2
CONTEST_PER_NODE = 4
GARRISON_RADIUS = 96.0
NODE_GARRISON_ARMY_FRAC = 0.20
MAX_NODE_GARRISON_ABS = 18
MAX_NODE_SCOUTS = 4
# Scouting
SCOUT_MIN_TICK = 48
SCOUT_ORBIT_FRACTION = 0.35
# Items
MAX_ITEM_HUNTERS = 3
ITEM_HUNT_RANGE = 3600.0
ITEM_DANGER_RADIUS = 150.0
ITEM_DANGER_COUNT = 3
# Stars
STAR_SQUAD_SIZE = 5 # Weak
STAR_SQUAD_MEDIUM = 12 # Medium (orange)
STAR_SQUAD_STRONG = 24 # Strong (red)
STAR_HUNT_RANGE_EARLY = 900.0
STAR_HUNT_RANGE_MID = 1400.0
STAR_HUNT_SURPLUS = 6
# Threat / combat. Circles park at the edge of their firing range (~105) so
# they shoot from the backline instead of walking into melee.
THREAT_RADIUS = 120.0
KITE_RANGE = 35.0
KITE_FLEE_DIST = 60.0
CIRCLE_BACKLINE_OFFSET = 105.0
# Kite moves are deduped on a coarse grid: a new flee point in the same cell
# as the in-progress one isn't worth a fresh command.
KITE_DEDUP_CELL = 20.0
# If one of our units is within this distance of a remembered enemy
# building/star and it is NOT in the visible list, it is provably gone —
# 100 is safely inside every unit's sight range.
MEMORY_PRUNE_RADIUS = 100.0
# ---------------------------------------------------------------------------
# Observation-space enums (as returned by PachinkoEnv.get_state())
# ---------------------------------------------------------------------------
T_SQUARE, T_CIRCLE, T_TRIANGLE = 0, 1, 2
K_SQUARRACKS, K_CIRCHERY, K_TRIABLES, K_HEMATITE, K_ARROW_TOWER = 0, 1, 2, 3, 4
MILITARY_KINDS = (K_SQUARRACKS, K_CIRCHERY, K_TRIABLES)
# PyBuilding.upgrades list order, and the strings step_direct expects.
UP_SPAWN_RATE, UP_WEAPON_DAMAGE, UP_ARMOR, UP_MOVE_SPEED, UP_MINING_RATE, UP_HEALING_AURA = range(6)
UPGRADE_NAME = ["SpawnRate", "WeaponDamage", "Armor", "MoveSpeed", "MiningRate", "HealingAura"]
UPGRADE_BASE_COST = [150, 200, 200, 250, 100, 500]
VALID_UPGRADES = {
K_SQUARRACKS: [UP_SPAWN_RATE, UP_WEAPON_DAMAGE, UP_ARMOR, UP_MOVE_SPEED],
K_CIRCHERY: [UP_SPAWN_RATE, UP_WEAPON_DAMAGE, UP_ARMOR],
K_TRIABLES: [UP_SPAWN_RATE, UP_WEAPON_DAMAGE, UP_ARMOR, UP_MOVE_SPEED],
K_HEMATITE: [UP_MINING_RATE, UP_HEALING_AURA],
K_ARROW_TOWER: [],
}
def upgrade_cost(upgrade: int, current_level: int) -> int:
"""Costs scale linearly with level; HealingAura is a one-time purchase."""
if upgrade == UP_HEALING_AURA:
return UPGRADE_BASE_COST[upgrade]
return UPGRADE_BASE_COST[upgrade] * (current_level + 1)
# ---------------------------------------------------------------------------
# Geometry helpers
# ---------------------------------------------------------------------------
def dist_sq(ax, ay, bx, by):
dx, dy = ax - bx, ay - by
return dx * dx + dy * dy
def grid_cell(x, y):
return int(x / KITE_DEDUP_CELL), int(y / KITE_DEDUP_CELL)
def circle_backline(tx, ty, bx, by):
"""Position a Circle CIRCLE_BACKLINE_OFFSET behind the target, toward base."""
dx, dy = bx - tx, by - ty
ln = math.hypot(dx, dy) or 1.0
return tx + (dx / ln) * CIRCLE_BACKLINE_OFFSET, ty + (dy / ln) * CIRCLE_BACKLINE_OFFSET
def kite_pos(unit, visible_enemy_units):
"""If a melee enemy is within KITE_RANGE, return a flee position away from it."""
kite_sq = KITE_RANGE * KITE_RANGE
melee = [
e for e in visible_enemy_units
if e.unit_type in (T_SQUARE, T_TRIANGLE)
and dist_sq(unit.x, unit.y, e.x, e.y) <= kite_sq
]
if not melee:
return None
threat = min(melee, key=lambda e: dist_sq(unit.x, unit.y, e.x, e.y))
dx, dy = unit.x - threat.x, unit.y - threat.y
ln = math.hypot(dx, dy) or 1.0
return unit.x + (dx / ln) * KITE_FLEE_DIST, unit.y + (dy / ln) * KITE_FLEE_DIST
def scout_waypoints(cx, cy, world_w, world_h):
rx = world_w * SCOUT_ORBIT_FRACTION
ry = world_h * SCOUT_ORBIT_FRACTION
return [(cx + rx, cy), (cx, cy + ry), (cx - rx, cy), (cx, cy - ry)]
# ---------------------------------------------------------------------------
# Recruitment helpers
# ---------------------------------------------------------------------------
def recruit_nearest(units, assigned, tx, ty, count):
"""Mark and return the indices of the N nearest unassigned units."""
candidates = sorted(
((i, dist_sq(u.x, u.y, tx, ty)) for i, u in enumerate(units) if not assigned[i]),
key=lambda c: c[1],
)
out = []
for i, _ in candidates:
if len(out) >= count:
break
assigned[i] = True
out.append(i)
return out
def recruit_type_prefer(units, assigned, tx, ty, count, preferred_type):
"""Like recruit_nearest, but the preferred type wins within ~500 world units."""
penalty_sq = 500.0 * 500.0
candidates = sorted(
(
(i, dist_sq(u.x, u.y, tx, ty) + (0.0 if u.unit_type == preferred_type else penalty_sq))
for i, u in enumerate(units) if not assigned[i]
),
key=lambda c: c[1],
)
out = []
for i, _ in candidates:
if len(out) >= count:
break
assigned[i] = True
out.append(i)
return out
def count_nearby_unassigned(units, assigned, px, py, radius):
r_sq = radius * radius
return sum(
1 for i, u in enumerate(units)
if not assigned[i] and dist_sq(u.x, u.y, px, py) <= r_sq
)
# ---------------------------------------------------------------------------
# Target selection
# ---------------------------------------------------------------------------
def score_target(bx, by, base_x, base_y, visible_enemy_units):
"""Lower = more attractive. Distance plus a penalty per nearby defender."""
d = math.hypot(base_x - bx, base_y - by)
defense_sq = 200.0 * 200.0
defenders = sum(
1 for u in visible_enemy_units if dist_sq(u.x, u.y, bx, by) <= defense_sq
)
return (d + defenders * 40.0) * 0.7
def best_attack_target(known_buildings, base_x, base_y, visible_enemy_units):
"""known_buildings: iterable of (x, y, kind). Military buildings only."""
mil = [(x, y) for (x, y, kind) in known_buildings if kind in MILITARY_KINDS]
if not mil:
return None
return min(mil, key=lambda p: score_target(p[0], p[1], base_x, base_y, visible_enemy_units))
def best_harass_target(visible_buildings, base_x, base_y, visible_enemy_units, main_target):
"""Least-defended visible military building, excluding the main-squad target."""
defense_sq = 200.0 * 200.0
exclusion_sq = 200.0 * 200.0
mil = [b for b in visible_buildings if b.kind in MILITARY_KINDS]
if main_target is not None:
mx, my = main_target
mil = [b for b in mil if dist_sq(b.x, b.y, mx, my) >= exclusion_sq]
if not mil:
return None
def key(b):
defenders = sum(
1 for u in visible_enemy_units if dist_sq(u.x, u.y, b.x, b.y) <= defense_sq
)
return (defenders, dist_sq(base_x, base_y, b.x, b.y))
b = min(mil, key=key)
return (b.x, b.y)
# ---------------------------------------------------------------------------
# The bot
# ---------------------------------------------------------------------------
class SimpleAIDirect:
"""Port of the game's built-in AI for the pybind direct API.
Call decide(state) every tick with the dict from PachinkoEnv.get_state();
it returns the list of command dicts to pass to step_direct(). Off-cadence
ticks (tick % 4 != 0) return [] — the engine keeps executing standing
orders between AI decisions.
"""
def __init__(self):
# Squads: {"target": (x, y), "launch_tick": int, "members": set[int]}
self.main_squad = None
self.harass_squad = None
# Counter-building hysteresis: (kind, locked_until_tick)
self.counter_lock = None
self.scout_id = None
self.scout_waypoint = 0
# Static positions, captured on the first decision tick.
self.base = None # (x, y) — centroid of our starting buildings
self.center = None
# Fog-of-war memory: things we have seen and have no proof are gone.
self.known_enemy_buildings = {} # id -> (x, y, kind)
self.known_stars = {} # id -> (x, y, tier)
# Last order WE issued per unit id: ("am"|"mv"|"au", a, b).
# The API exposes only has_target/attack_moving, not the order target,
# so this is how we avoid re-issuing identical commands every tick.
self.orders = {}
# -- command emission ---------------------------------------------------
def _atk_move(self, unit_id, x, y):
self.orders[unit_id] = ("am", x, y)
self._am_batches.setdefault((x, y), []).append(unit_id)
def _move(self, unit_id, x, y):
self.orders[unit_id] = ("mv", x, y)
self._cmds.append({"type": "move", "unit_id": unit_id, "x": x, "y": y})
def _attack_entity(self, unit_id, target_id):
self.orders[unit_id] = ("au", target_id, 0)
self._cmds.append({"type": "attack_unit", "unit_id": unit_id, "target_id": target_id})
def _has_atk_move_order(self, unit, x, y):
"""True if the unit is still executing an attack-move we issued to (x, y).
Stands in for checking the unit's current order destination, which the
observation space doesn't expose.
"""
return self.orders.get(unit.id) == ("am", x, y) and unit.attack_moving
# -- fog-of-war memory ----------------------------------------------------
def _update_memory(self, my_units, visible_buildings, visible_stars):
prune_sq = MEMORY_PRUNE_RADIUS * MEMORY_PRUNE_RADIUS
for b in visible_buildings:
self.known_enemy_buildings[b.id] = (b.x, b.y, b.kind)
visible_b = {b.id for b in visible_buildings}
for bid, (bx, by, _) in list(self.known_enemy_buildings.items()):
if bid in visible_b:
continue
if any(dist_sq(u.x, u.y, bx, by) <= prune_sq for u in my_units):
del self.known_enemy_buildings[bid]
for s in visible_stars:
self.known_stars[s.id] = (s.x, s.y, s.tier)
visible_s = {s.id for s in visible_stars}
for sid, (sx, sy, _) in list(self.known_stars.items()):
if sid in visible_s:
continue
# Gone from where we last saw it: dead, or it aggro-chased someone.
# Either way the stale entry is useless — it re-registers on sight.
if any(dist_sq(u.x, u.y, sx, sy) <= prune_sq for u in my_units):
del self.known_stars[sid]
# -- counter-building selection (with hysteresis lock) --------------------
def _counter_building(self, visible_enemy_units, tick):
if self.counter_lock is not None and tick < self.counter_lock[1]:
return self.counter_lock[0]
sq = sum(1 for u in visible_enemy_units if u.unit_type == T_SQUARE)
ci = sum(1 for u in visible_enemy_units if u.unit_type == T_CIRCLE)
tr = sum(1 for u in visible_enemy_units if u.unit_type == T_TRIANGLE)
if tr > max(sq, ci):
choice = K_SQUARRACKS
elif sq > max(ci, tr):
choice = K_CIRCHERY
elif ci > max(sq, tr):
choice = K_TRIABLES
else:
choice = K_SQUARRACKS
if self.counter_lock is None or self.counter_lock[0] != choice:
self.counter_lock = (choice, tick + COUNTER_LOCK_TICKS)
return choice
# -- upgrades --------------------------------------------------------------
def _do_upgrades(self, buildings, resources, primary, secondary, is_rich,
under_pressure, my_units):
refinery = next((b for b in buildings if b.kind == K_HEMATITE), None)
prim_bldg = next((b for b in buildings if b.kind == primary), None)
sec_bldg = next((b for b in buildings if b.kind == secondary), None)
# Healing Aura: opportunistic whenever >10 units are injured, even
# under pressure. (max_hp is the unbuffed base value, so with HP-buff
# items the injured count is a slight overestimate — fine here.)
if refinery is not None and refinery.upgrades[UP_HEALING_AURA] == 0:
cost = upgrade_cost(UP_HEALING_AURA, 0)
if resources >= cost:
injured = sum(1 for u in my_units if u.hp < u.max_hp)
if injured > 10:
self._cmds.append({
"type": "upgrade", "building_id": refinery.id,
"upgrade": UPGRADE_NAME[UP_HEALING_AURA],
})
if under_pressure:
# Outnumbered: skip tech, every hematite goes into the production burst.
return
if is_rich:
estimated_spend = 0
issued = 0
if refinery is not None and issued < RICH_UPGRADE_LIMIT:
cost = upgrade_cost(UP_MINING_RATE, refinery.upgrades[UP_MINING_RATE])
if resources - estimated_spend >= cost:
self._cmds.append({
"type": "upgrade", "building_id": refinery.id,
"upgrade": UPGRADE_NAME[UP_MINING_RATE],
})
estimated_spend += cost
issued += 1
for b in buildings:
if issued >= RICH_UPGRADE_LIMIT:
break
for up in VALID_UPGRADES.get(b.kind, []):
if issued >= RICH_UPGRADE_LIMIT:
break
if up in (UP_MINING_RATE, UP_HEALING_AURA):
continue # already handled above
cost = upgrade_cost(up, b.upgrades[up])
if resources - estimated_spend >= cost:
self._cmds.append({
"type": "upgrade", "building_id": b.id,
"upgrade": UPGRADE_NAME[up],
})
estimated_spend += cost
issued += 1
return
# Normal economy: fixed priority plan, one purchase per tick.
# (0 = refinery, 1 = primary building, 2 = secondary building)
upgrade_plan = [
(0, UP_MINING_RATE, 1),
(1, UP_WEAPON_DAMAGE, 1),
(0, UP_MINING_RATE, 2),
(1, UP_WEAPON_DAMAGE, 2),
(0, UP_MINING_RATE, 3),
(1, UP_WEAPON_DAMAGE, 3),
(0, UP_MINING_RATE, 4),
(1, UP_ARMOR, 2),
(2, UP_WEAPON_DAMAGE, 1),
(0, UP_MINING_RATE, 5),
(1, UP_WEAPON_DAMAGE, 4),
(1, UP_MOVE_SPEED, 2),
(2, UP_WEAPON_DAMAGE, 2),
(0, UP_MINING_RATE, 6),
(1, UP_ARMOR, 3),
(1, UP_WEAPON_DAMAGE, 5),
(1, UP_MOVE_SPEED, 3),
(2, UP_WEAPON_DAMAGE, 3),
]
targets = [refinery, prim_bldg, sec_bldg]
for slot, up, max_level in upgrade_plan:
b = targets[slot]
if b is None:
continue
current = b.upgrades[up]
if current < max_level:
cost = upgrade_cost(up, current)
if resources >= cost + UPGRADE_RESERVE:
self._cmds.append({
"type": "upgrade", "building_id": b.id,
"upgrade": UPGRADE_NAME[up],
})
break
# -- main decision pass ------------------------------------------------------
def decide(self, state):
tick = state["tick"]
if tick % AI_TICK_INTERVAL != 0:
return []
self._cmds = []
self._am_batches = {}
units = state["my_units"]
buildings = state["my_buildings"]
visible_enemy_units = state["enemy_units"] # FOW-filtered by the engine
visible_enemy_buildings = state["enemy_buildings"] # FOW-filtered by the engine
resources = state["resources"]
world_w = state["world_width"]
world_h = state["world_height"]
unit_count = len(units)
# Static positions, once. Our starting buildings never move, so their
# centroid is a faithful stand-in for our spawn point.
if self.base is None and buildings:
self.base = (
sum(b.x for b in buildings) / len(buildings),
sum(b.y for b in buildings) / len(buildings),
)
if self.center is None:
self.center = (world_w * 0.5, world_h * 0.5)
if self.base is None:
return []
base_x, base_y = self.base
center_x, center_y = self.center
# Drop order records for units that no longer exist.
live_ids = {u.id for u in units}
for uid in list(self.orders):
if uid not in live_ids:
del self.orders[uid]
self._update_memory(units, visible_enemy_buildings, visible_stars=state["visible_stars"])
known_buildings = list(self.known_enemy_buildings.values())
# Pressure check. We can only count visible enemies, so this triggers
# conservatively — an unscouted army doesn't set off the alarm.
under_pressure = (
len(visible_enemy_units) > 0
and unit_count < len(visible_enemy_units) * 0.75
)
# =====================================================================
# 1. Production management
# =====================================================================
is_rich = resources >= RICH_THRESHOLD
effective_cap = RICH_ARMY_CAP if is_rich else ARMY_CAP
if tick < EARLY_CAVALRY_TICKS:
primary = K_TRIABLES
else:
primary = self._counter_building(visible_enemy_units, tick)
secondary = {K_SQUARRACKS: K_TRIABLES, K_CIRCHERY: K_SQUARRACKS, K_TRIABLES: K_CIRCHERY}[primary]
can_expand = resources >= EXPAND_THRESHOLD and tick >= EXPAND_MIN_TICK
if unit_count >= effective_cap:
active_kinds = set()
elif is_rich or under_pressure:
active_kinds = {K_SQUARRACKS, K_CIRCHERY, K_TRIABLES}
elif can_expand:
active_kinds = {primary, secondary}
else:
active_kinds = {primary}
for b in buildings:
if b.kind not in MILITARY_KINDS:
continue
should_be_on = b.kind in active_kinds
if should_be_on != b.spawn_toggle:
self._cmds.append({"type": "toggle_spawn", "building_id": b.id, "on": should_be_on})
# =====================================================================
# 2. Upgrades
# =====================================================================
self._do_upgrades(buildings, resources, primary, secondary, is_rich,
under_pressure, units)
# =====================================================================
# 3. Unit assignment
# =====================================================================
threat_sq = THREAT_RADIUS * THREAT_RADIUS
threat_scan_sq = GARRISON_THREAT_SCAN_RADIUS * GARRISON_THREAT_SCAN_RADIUS
assigned = [False] * unit_count
# -- 3.0 Retreat low-HP units in active combat ------------------------
for i, u in enumerate(units):
if u.hp / max(u.max_hp, 1) <= RETREAT_HP_FRACTION:
in_combat = any(
dist_sq(u.x, u.y, e.x, e.y) <= threat_sq for e in visible_enemy_units
)
if in_combat:
self._move(u.id, base_x, base_y)
assigned[i] = True
# -- Lock in units actively attack-moving near enemies ----------------
for i, u in enumerate(units):
if assigned[i] or not u.attack_moving:
continue
if any(dist_sq(u.x, u.y, e.x, e.y) <= threat_sq for e in visible_enemy_units):
assigned[i] = True
# -- 3a-pre. Squad upkeep, reset conditions, and member drive ---------
if self.main_squad is not None:
self.main_squad["members"] &= live_ids
if self.harass_squad is not None:
self.harass_squad["members"] &= live_ids
base_under_attack = sum(
1 for e in visible_enemy_units
if dist_sq(e.x, e.y, base_x, base_y) <= threat_scan_sq
) >= BASE_RECALL_ENEMY_COUNT
if base_under_attack:
self.main_squad = None
self.harass_squad = None
if self.main_squad is not None:
# "No targets left" only counts once we have actually found their
# base — with an empty memory we assume targets still exist.
no_targets = (
bool(self.known_enemy_buildings)
and not any(k in MILITARY_KINDS for (_, _, k) in known_buildings)
)
depleted = len(self.main_squad["members"]) < WAVE_DEPLETED_THRESHOLD
stalled = tick - self.main_squad["launch_tick"] > WAVE_STALL_TIMEOUT
if no_targets or depleted or stalled:
self.main_squad = None
if self.harass_squad is not None:
if tick - self.harass_squad["launch_tick"] > HARASS_STALL_TIMEOUT:
self.harass_squad = None
# Attack target: best known military building, else a visible enemy
# unit, else the mirror of our base — the standard "haven't scouted
# them yet" guess.
main_target = best_attack_target(known_buildings, base_x, base_y, visible_enemy_units)
if main_target is None and visible_enemy_units:
main_target = (visible_enemy_units[0].x, visible_enemy_units[0].y)
if main_target is None:
main_target = (world_w - base_x, world_h - base_y)
# Drive active main squad members; mark assigned so the garrison and
# node steps below cannot steal them.
if self.main_squad is not None:
tx, ty = self.main_squad["target"]
members = self.main_squad["members"]
for i, u in enumerate(units):
if u.id not in members or assigned[i]:
continue
assigned[i] = True
if u.unit_type == T_CIRCLE:
flee = kite_pos(u, visible_enemy_units)
if flee is not None:
# Skip if the last kite-move we issued lands in the same
# terrain cell and is still in progress.
last = self.orders.get(u.id)
same_cell = (
last is not None and last[0] == "mv" and u.has_target
and grid_cell(last[1], last[2]) == grid_cell(*flee)
)
if not same_cell:
self._move(u.id, *flee)
continue
dest = circle_backline(tx, ty, base_x, base_y)
if not self._has_atk_move_order(u, *dest):
self._atk_move(u.id, *dest)
elif not self._has_atk_move_order(u, tx, ty):
self._atk_move(u.id, tx, ty)
# Drive active harass squad members.
if self.harass_squad is not None:
htx, hty = self.harass_squad["target"]
members = self.harass_squad["members"]
for i, u in enumerate(units):
if u.id not in members or assigned[i]:
continue
assigned[i] = True
if not self._has_atk_move_order(u, htx, hty):
self._atk_move(u.id, htx, hty)
# -- 3a. Threat-scaled base defense garrison ---------------------------
garrison_zone_sq = GARRISON_DEFEND_RADIUS * GARRISON_DEFEND_RADIUS
nearby_enemies = sum(
1 for e in visible_enemy_units
if dist_sq(e.x, e.y, base_x, base_y) <= threat_scan_sq
)
garrison_target = min(
GARRISON_BASE_SIZE + int(nearby_enemies * GARRISON_PER_THREAT),
GARRISON_MAX_SIZE,
max(0, unit_count - GARRISON_LEAVE_FREE),
)
at_home = sorted(
(
(i, dist_sq(u.x, u.y, base_x, base_y))
for i, u in enumerate(units)
if not assigned[i] and dist_sq(u.x, u.y, base_x, base_y) <= garrison_zone_sq
),
key=lambda c: c[1],
)
intercept_sq = GARRISON_INTERCEPT_RADIUS * GARRISON_INTERCEPT_RADIUS
intercept_pos = None
approaching = [
e for e in visible_enemy_units
if dist_sq(e.x, e.y, base_x, base_y) <= intercept_sq
]
if approaching:
e = min(approaching, key=lambda e: dist_sq(e.x, e.y, base_x, base_y))
intercept_pos = (e.x, e.y)
garrison_dest = intercept_pos or (base_x, base_y)
garrison_count = 0
for i, _ in at_home[:garrison_target]:
assigned[i] = True
garrison_count += 1
# Only issue a command when there is an active intercept target;
# units already sitting at base don't need re-ordering every tick.
if intercept_pos is not None:
self._atk_move(units[i].id, *garrison_dest)
deficit = garrison_target - garrison_count
if deficit > 0:
for i in recruit_nearest(units, assigned, base_x, base_y, deficit):
self._atk_move(units[i].id, *garrison_dest)
# -- 3b. Scouting -------------------------------------------------------
if tick >= SCOUT_MIN_TICK:
waypoints = scout_waypoints(center_x, center_y, world_w, world_h)
if self.scout_id is not None and self.scout_id not in live_ids:
self.scout_id = None
if self.scout_id is None and unit_count > garrison_target + MAIN_SQUAD_SIZE:
wx, wy = waypoints[self.scout_waypoint % len(waypoints)]
candidates = [
(i, dist_sq(u.x, u.y, wx, wy))
for i, u in enumerate(units) if not assigned[i]
]
if candidates:
i = min(candidates, key=lambda c: c[1])[0]
self.scout_id = units[i].id
assigned[i] = True
self._move(units[i].id, wx, wy)
elif self.scout_id is not None:
pair = next(((i, u) for i, u in enumerate(units) if u.id == self.scout_id), None)
if pair is not None:
i, scout = pair
assigned[i] = True
threatened = any(
dist_sq(e.x, e.y, scout.x, scout.y) <= threat_sq
for e in visible_enemy_units
)
if threatened:
self._move(scout.id, base_x, base_y)
else:
wx, wy = waypoints[self.scout_waypoint % len(waypoints)]
if dist_sq(scout.x, scout.y, wx, wy) < 60.0 * 60.0:
self.scout_waypoint = (self.scout_waypoint + 1) % len(waypoints)
nx, ny = waypoints[self.scout_waypoint]
self._move(scout.id, nx, ny)
# -- 3c. Item pickup (skip items in dangerous areas) --------------------
item_range_sq = ITEM_HUNT_RANGE * ITEM_HUNT_RANGE
item_danger_sq = ITEM_DANGER_RADIUS * ITEM_DANGER_RADIUS
item_hunters = 0
for item in state["visible_items"]:
if item_hunters >= MAX_ITEM_HUNTERS:
break
if dist_sq(base_x, base_y, item.x, item.y) > item_range_sq:
continue
danger = sum(
1 for e in visible_enemy_units
if dist_sq(e.x, e.y, item.x, item.y) <= item_danger_sq
)
if danger >= ITEM_DANGER_COUNT:
continue
for i in recruit_nearest(units, assigned, item.x, item.y, 1):
self._move(units[i].id, item.x, item.y)
item_hunters += 1
# -- Lock in units already at or heading toward a resource node ---------
# Prevents node garrisons from being recalled to base every tick.
nodes = state["resource_nodes"]
garrison_sq = GARRISON_RADIUS * GARRISON_RADIUS
for i, u in enumerate(units):
if assigned[i]:
continue
last = self.orders.get(u.id)
heading = (
last is not None and last[0] == "am" and u.has_target
and any(dist_sq(last[1], last[2], n.x, n.y) < 4.0 for n in nodes)
)
at_node = any(dist_sq(u.x, u.y, n.x, n.y) <= garrison_sq for n in nodes)
if at_node or heading:
assigned[i] = True
# -- 3d. Resource node garrisoning ---------------------------------------
max_node_garrison = min(int(unit_count * NODE_GARRISON_ARMY_FRAC), MAX_NODE_GARRISON_ABS)
all_nodes = sorted(nodes, key=lambda n: dist_sq(n.x, n.y, base_x, base_y))
node_counts = [
sum(1 for u in units if dist_sq(u.x, u.y, n.x, n.y) <= garrison_sq)
for n in all_nodes
]
# node.control: 1 = ours, 0 = neutral, -1 = enemy
node_targets = [
GARRISON_PER_NODE if n.control >= 0 else CONTEST_PER_NODE
for n in all_nodes
]
node_priority = []
for di, n in enumerate(all_nodes):
deficit = node_targets[di] - node_counts[di]
if deficit <= 0:
continue
richness_bonus = int(n.richness * 10.0)
ownership_bonus = 30 if n.control == 1 else (15 if n.control == 0 else 5)
node_priority.append((di, deficit + richness_bonus + ownership_bonus))
node_priority.sort(key=lambda p: -p[1])
node_budget = max(0, max_node_garrison - sum(node_counts))
for di, _ in node_priority:
if node_budget == 0:
break
node = all_nodes[di]
deficit = max(0, node_targets[di] - node_counts[di])
if deficit == 0:
continue
to_send = min(deficit, node_budget)
# Exclude units holding OTHER friendly nodes at/below capacity, so
# we don't rob one garrison to fill another.
temp = list(assigned)
for odi, other in enumerate(all_nodes):
if odi == di or other.control != 1:
continue
if node_counts[odi] >= node_targets[odi]:
continue
for i, u in enumerate(units):
if dist_sq(u.x, u.y, other.x, other.y) <= garrison_sq:
temp[i] = True
recruited = recruit_type_prefer(units, temp, node.x, node.y, to_send, T_TRIANGLE)
for i in recruited:
assigned[i] = True
self._atk_move(units[i].id, node.x, node.y)
node_counts[di] += 1
node_budget -= 1
# Roaming cavalry scouts: solo Triangles toward uncaptured nodes.
scouts_sent = 0
wide_sq = garrison_sq * 4.0
for node in all_nodes:
if scouts_sent >= MAX_NODE_SCOUTS:
break
if node.control == 1:
continue
already_covered = any(
assigned[i] and (
dist_sq(u.x, u.y, node.x, node.y) <= wide_sq
or (
(last := self.orders.get(u.id)) is not None
and last[0] == "am" and u.has_target
and dist_sq(last[1], last[2], node.x, node.y) < 4.0
)
)
for i, u in enumerate(units)
)
if already_covered:
continue
for i in recruit_type_prefer(units, assigned, node.x, node.y, 1, T_TRIANGLE):
self._atk_move(units[i].id, node.x, node.y)
scouts_sent += 1
# -- 3e. Star hunting ------------------------------------------------------
if tick < 480:
hunt_range_sq = STAR_HUNT_RANGE_EARLY ** 2
elif tick < 1200:
hunt_range_sq = STAR_HUNT_RANGE_MID ** 2
else:
hunt_range_sq = float("inf")
unassigned_now = sum(1 for a in assigned if not a)
if unassigned_now >= MAIN_SQUAD_SIZE + STAR_HUNT_SURPLUS:
staged_at_base = count_nearby_unassigned(units, assigned, base_x, base_y, STAGING_RADIUS)
attackable = sorted(
(
(sid, sx, sy, tier) for sid, (sx, sy, tier) in self.known_stars.items()
if dist_sq(base_x, base_y, sx, sy) <= hunt_range_sq
),
key=lambda s: dist_sq(base_x, base_y, s[1], s[2]),
)
star_budget = unassigned_now - MAIN_SQUAD_SIZE
for sid, sx, sy, tier in attackable:
needed = (STAR_SQUAD_SIZE, STAR_SQUAD_MEDIUM, STAR_SQUAD_STRONG)[tier]
if star_budget < needed:
continue
# Medium/strong stars: only attack with a group already staged,
# so units depart together rather than trickling in.
staged_required = (0, STAR_SQUAD_MEDIUM, STAR_SQUAD_STRONG)[tier]
if staged_at_base < staged_required:
continue
recruited = recruit_nearest(units, assigned, sx, sy, needed)
for i in recruited:
self._attack_entity(units[i].id, sid)
star_budget -= len(recruited)
# -- 3f. Main attack squad launch --------------------------------------------
free_count = sum(1 for a in assigned if not a)
if self.main_squad is None:
staged = count_nearby_unassigned(units, assigned, base_x, base_y, STAGING_RADIUS)
# Overwhelm: launch everything when we badly outnumber the enemy.
# We only see visible enemies, so require at least one to be
# visible before treating the count as meaningful.
overwhelm = (
free_count >= OVERWHELM_MIN_UNITS
and len(visible_enemy_units) > 0
and free_count >= len(visible_enemy_units) * OVERWHELM_RATIO
)
if staged >= MAIN_SQUAD_SIZE or overwhelm:
to_recruit = free_count if overwhelm else staged
primary_unit = {K_SQUARRACKS: T_SQUARE, K_CIRCHERY: T_CIRCLE, K_TRIABLES: T_TRIANGLE}[primary]
recruited = recruit_type_prefer(units, assigned, *main_target, to_recruit, primary_unit)
members = {units[i].id for i in recruited}
for i in recruited:
if units[i].unit_type == T_CIRCLE:
self._atk_move(units[i].id, *circle_backline(*main_target, base_x, base_y))
else:
self._atk_move(units[i].id, *main_target)
self.main_squad = {"target": main_target, "launch_tick": tick, "members": members}
# -- 3g. Harass squad launch ---------------------------------------------------
if (
tick >= HARASS_MIN_TICK
and unit_count >= HARASS_MIN_ARMY
and self.harass_squad is None
):
harass_target = best_harass_target(
visible_enemy_buildings, base_x, base_y, visible_enemy_units,
self.main_squad["target"] if self.main_squad else None,
)
if harass_target is not None:
harass_free = sum(1 for a in assigned if not a)
if harass_free >= MAIN_SQUAD_SIZE + HARASS_SQUAD_SIZE:
recruited = recruit_type_prefer(
units, assigned, *harass_target, HARASS_SQUAD_SIZE, T_TRIANGLE
)
members = {units[i].id for i in recruited}
for i in recruited:
self._atk_move(units[i].id, *harass_target)
self.harass_squad = {"target": harass_target, "launch_tick": tick, "members": members}
# -- 3h. Remaining free units: stage at base ---------------------------------
staging_sq = STAGING_RADIUS * STAGING_RADIUS
for i, u in enumerate(units):
if assigned[i]:
continue
if u.unit_type == T_CIRCLE:
flee = kite_pos(u, visible_enemy_units)
if flee is not None:
last = self.orders.get(u.id)
same_cell = (
last is not None and last[0] == "mv" and u.has_target
and grid_cell(last[1], last[2]) == grid_cell(*flee)
)
if not same_cell:
self._move(u.id, *flee)
continue
at_base = dist_sq(u.x, u.y, base_x, base_y) <= staging_sq
if not at_base and not self._has_atk_move_order(u, base_x, base_y):
self._atk_move(u.id, base_x, base_y)
# Flush batched attack-moves (one grouped command per unique target).
for (x, y), unit_ids in self._am_batches.items():
self._cmds.append({"type": "attack_move", "unit_ids": unit_ids, "x": x, "y": y})
return self._cmds
# ---------------------------------------------------------------------------
# Runner: SimpleAIDirect (P0) vs the built-in AI (P1)
# ---------------------------------------------------------------------------
def main():
import sys
import pachinko_pybind as eng
seed = int(sys.argv[1]) if len(sys.argv) > 1 else 42
# builtin_ai=True pits us against the built-in AI this file ports. It gets
# its standard practice-game economy handicap; pass
# builtin_ai_handicap=False for an even match.
# return_obs=False skips the gym observation arrays, so numpy isn't needed.
env = eng.PachinkoEnv(num_games=1, builtin_ai=True)
env.reset(seed=seed, return_obs=False)
bot = SimpleAIDirect()
done, info = False, {}
while not done:
state = env.get_state(0)
done, info = env.step_direct(bot.decide(state))
if state["tick"] % 600 == 0:
print(
f"tick {state['tick']:>5} units {len(state['my_units']):>3} "
f"vs {len(state['enemy_units']):>3} visible hematite {state['resources']}"
)
print(f"result: {info['result']} bankrolls: {info['bankroll_p0']} / {info['bankroll_p1']}")
if __name__ == "__main__":
main()