Skip to content

AI

Jump to bottom
Alrik Neihouser edited this page Jun 16, 2026 · 2 revisions

AI

The AI is a fully heuristic Python implementation. Each player runs as an independent OS process and connects to the server over TCP. Players have no shared memory and no external bus — the only coordination channel is the in-game Broadcast command.

Usage

# wrapper script (sets PYTHONPATH and calls main.py)
./Ai/zappy_ai -p <port> -n <team> -h <host>

# or directly
PYTHONPATH=Ai/heuristic_ai python3 Ai/heuristic_ai/main.py -p 4242 -n team1 -h localhost

The first player to start forks itself until the team is full, so launching one instance is enough.

Optional environment variables:

Variable Description
ZAPPY_AI_CONFIG=file.json Override config parameters at runtime
ZAPPY_AI_TICK=0.001 Control loop sleep duration (default ~0.01s)

File map

Ai/heuristic_ai/
├── main.py           entry point — arg parsing, handshake, fork, launch
├── client.py         non-blocking TCP socket + command pipeline
├── ai.py             PlayerAI class — main loop + state machine
├── ai_ritual.py      incantation coordination (leader/follower logic)
├── ai_comms.py       broadcast protocol parsing and sending
├── ai_inventory.py   stone collection, drop logic, deficit helpers
├── ai_vision.py      look-response parser + tile navigation helpers
├── constants.py      elevation table, resource rarity order, config defaults

Architecture

1. Connection (client.py)

A non-blocking TCP socket with a FIFO command pipeline. The Zappy server accepts up to 10 commands before answering any of them, so the client sends immediately and queues (command, callback) pairs. Each call to pump() reads available bytes, splits on \n, and dispatches each line to the oldest pending callback.

Unsolicited messages — dead, message K, text, eject: K, and Current level: K — can arrive at any time outside the normal request/response rhythm. They are caught before touching the pending queue and routed to registered slots (on_dead, on_broadcast, on_eject, on_level_up).

Incantation edge case: after Elevation underway the server may later send ko (failure) instead of Current level: K. A dedicated _incantation_end_cb slot intercepts that ko so it cannot desync the pipeline.

Handshake is the one blocking operation: the socket goes into blocking mode, reads three lines synchronously (WELCOME, slots, dimensions), then returns to non-blocking for the rest of the game.

2. Commands (client.py)

Every server command is a one-line wrapper that calls push() and converts the raw response to a typed value before calling the callback:

Function Server command Callback type
forward Forward bool
turn_right / turn_left Right / Left bool
look Look list[list[str]] tiles
inventory Inventory dict[str, int]
broadcast Broadcast TEXT bool
connect_nbr Connect_nbr int open slots
fork Fork bool
eject Eject bool
take Take RES bool
set_down Set RES bool
incantation Incantation int|None new level

3. State machine (ai.py)

Each player is always in exactly one state:

SURVIVE       food < FOOD_CRITICAL (5)  — emergency, drop everything
GATHER_FOOD   food < FOOD_SAFE          — top up before anything else
GATHER_STONES food ok, collecting stones for the current level
SEEK_TEAM     stones collected, homing toward incantation leader
WAIT_TEAM     on leader tile, waiting for enough players
INCANTATING   frozen during ritual
FORKING       brief detour to lay an egg

Transitions checked every tick:

any state     → SURVIVE        if food < FOOD_CRITICAL (except INCANTATING)
SURVIVE       → GATHER_FOOD    if food >= FOOD_CRITICAL + 2
GATHER_FOOD   → GATHER_STONES  if food >= FOOD_SAFE
GATHER_STONES → GATHER_FOOD    if food < FOOD_LOW
GATHER_STONES → WAIT_TEAM      if has_all_stones AND players_needed == 1
GATHER_STONES → SEEK_TEAM      if has_all_stones AND players_needed > 1
SEEK_TEAM     → GATHER_STONES  if seek_ticks > SEEK_TIMEOUT
WAIT_TEAM     → GATHER_STONES  if seek_ticks > WAIT_TIMEOUT
WAIT_TEAM/SEEK→ INCANTATING    on ritual fire
INCANTATING   → GATHER_FOOD    when ritual ends (success or failure)

A _coord_cooldown prevents immediately re-entering the coordination dance after a failed attempt.

4. Food thresholds (dynamic, map-aware)

Thresholds are computed at startup from map dimensions and slot count so they scale correctly from small to large maps:

total_food  = 0.5 * world_x * world_y
food_pp     = total_food / max(6, slots + 1)

FOOD_SAFE = clamp(
    food_pp * food_safe_mult + (world_x + world_y) * food_travel_factor,
    food_safe_min, food_safe_max
)
FOOD_LOW  = clamp(food_pp * food_low_mult, food_low_min, food_low_max)
FOOD_JOIN = FOOD_CRITICAL + int((world_x + world_y) * 0.5)

FOOD_JOIN (minimum food to accept a leader's call) is kept deliberately below FOOD_LOW — joining a nearby leader is cheap and gating it behind FOOD_LOW caused rituals to never fire on food-scarce maps.

5. Resource collection (ai_inventory.py)

Stone collection priority is rarest-first (RARITY_ORDER): thystame → phiras → mendiane → sibur → deraumere → linemate. This avoids accumulating plenty of common stones while the one thystame on the map remains uncollected.

6. Navigation (ai_vision.py)

Known-resource navigation: if the resource is on tile 0 → take it immediately. If it appears in the look grid → convert tile index to moves. If not visible → wander.

tile_to_moves(index, level): the look grid grows one row per vision level; row r starts at index r² and spans 2r+1 tiles. If the resource is mostly ahead (lateral offset < row depth), returns up to 3 Forward commands; otherwise turns once then steps.

Coverage wander: instead of a random walk, the player tracks visited tiles via dead reckoning and scores neighbours by how long ago they were visited. Least-recently-visited tile wins. This cuts search time on sparse maps.

Dead reckoning: position and facing are updated at command-issue time (not on response) because Forward/Left/Right never fail in Zappy (toroidal map, no walls). An Eject event can desync the stored position but only hurts wander efficiency, not correctness.

7. Broadcast protocol (ai_comms.py)

All team messages use the prefix ZAPPY: to be silently ignored by other teams. Format: ZAPPY:<TYPE>:<PAYLOAD>.

Type Payload Meaning
NEED_INC "level:uid" Leader calling same-level teammates
IM_COMING "leader_uid" Follower acknowledging the call
IM_READY "leader_uid" Follower arrived on leader's tile
START "leader_uid" Leader firing the incantation
LVL "new_level" Ritual finished (success or failure)
FORK "team_name" Player announcing a fork

Sound-based homing (moves_toward): broadcasts arrive with a direction integer k (0 = same tile, 1–8 = adjacent sectors per the spec grid). The function converts k to at most two primitives (a turn + one forward step) so the follower makes one step of progress per ping and re-aims on the next. Pure rotation without a step would cause the follower to orbit the target and never arrive.

  k layout:
    2  1  8
    3  @  7
    4  5  6

  k=0 → []                   (already there)
  k=1 → [Forward]            (straight ahead)
  k=2,3,4 → [Left, Forward]  (leader on the left)
  k=5 → [Left,Left,Forward]  (directly behind)
  k=6,7,8 → [Right,Forward]  (leader on the right)

8. Incantation coordination (ai_ritual.py)

The protocol is asymmetric: one player is the leader, the rest are followers.

Leader flow:

  1. Finishes stone collection → SEEK_TEAM (players_needed > 1).
  2. Listens LISTEN_BEFORE_LEAD ticks for an existing NEED_INC from another player.
  3. No reply → broadcasts NEED_INC("level:uid"), moves to WAIT_TEAM.
  4. In WAIT_TEAM: drops stones on tile, re-broadcasts NEED_INC every BCAST_INTERVAL ticks.
  5. When confirmed >= needed (or 30 ticks with enough bodies on tile) AND stones_ok: broadcasts START(uid), sends Incantation.
  6. Ritual ends → broadcasts LVL(new_level), goes to GATHER_FOOD.

Follower flow:

  1. Hears NEED_INC at same level, food ≥ FOOD_JOIN, no current leader → broadcasts IM_COMING(leader_uid), moves to SEEK_TEAM homing toward leader.
  2. On each new broadcast ping, recalculates one move via moves_toward(k).
  3. k == 0 (arrived) → broadcasts IM_READY(leader_uid), WAIT_TEAM.
  4. In WAIT_TEAM: re-broadcasts IM_READY every BCAST_INTERVAL (handles lost acks).
  5. Hears START(leader_uid) → INCANTATING (does not send Incantation itself; server auto-includes all players on the tile).
  6. Success: receives Current level: K → GATHER_FOOD. Failure: LVL broadcast from leader → GATHER_FOOD.

Leader election: if two players both become leaders simultaneously (race on LISTEN_BEFORE_LEAD), the higher-UID one yields — it switches to follower and homes toward the lower-UID leader. Yield only triggers when _ready_count == 0; a leader with committed followers stays put.

Forking: done opportunistically during GATHER_FOOD / GATHER_STONES every fork_interval ticks. Connect_nbr is checked first; if slots are already open, the fork is skipped (saves the 42/f cost).

Configuration (constants.py)

All tunable parameters live in DEFAULTS and can be overridden via a JSON file:

ZAPPY_AI_CONFIG=my_params.json ./Ai/zappy_ai -p 4242 -n team1

Key parameters:

Key Default Description
food_critical 5 Drop everything and find food now
food_safe_mult 1.2 Multiplier on food-per-player for FOOD_SAFE
food_safe_min/max 15/50 Clamp on computed FOOD_SAFE
food_travel_factor 0.3 Extra food reserve per unit of map span
food_low_mult 0.9 Multiplier for FOOD_LOW (re-enter food gathering)
join_food_buffer 3 Extra food above FOOD_CRITICAL to accept a call
seek_timeout 500 Ticks before giving up homing to a leader
wait_timeout_mult 2.0 Leader wait = seek_timeout × this
listen_before_lead 30 Ticks to listen before self-leading
bcast_interval 2 Ticks between leader re-broadcasts / follower IM_READY
fork_min_food 30 Minimum food before forking
fork_interval 150 Ticks between fork attempts

seek_timeout and wait_timeout are automatically scaled by map span ((world_x + world_y) / 20) so the same config works on both 10×10 and 20×20 maps.

Known issues and limitations

The level 3 stall (structural)

Level 1→2 needs 1 player, 2→3 needs 2, 3→4 jumps to 4 players — the first ritual that tests coordination at scale. With a 6-player team: all 6 reach level 2, pair into 3 simultaneous 2→3 rituals, and all 6 reach level 3. The first successful 4-player 3→4 ritual permanently strands the remaining 2 players at level 3 — they can never do 3→4 alone and have no same-level partners. They burn food until death.

_present_ticks fallback fires with wrong-level players

The leader fires an incantation after 30 consecutive ticks with enough bodies on the tile, even if IM_READY confirmations haven't arrived. count_players_on_tile counts any player, regardless of level. Wrong-level players drifting through the leader's tile can trigger a premature ko. Fix: gate the fallback on _ready_count >= needed - 1.

Dropped stones not reclaimed after failed ritual

When entering WAIT_TEAM a player calls _drop_for_ritual(), setting stone inventory to 0. On ritual failure the player transitions to GATHER_STONES and must re-collect all stones from scratch — the dropped stones are left on the old tile. On resource-scarce maps this can deplete the world.

Competing leaders split the recruitment pool

Multiple players finishing stone collection at similar times can both become leaders simultaneously. The UID yield guard only works when _ready_count == 0 — a leader with even one committed follower will not yield, causing the other leader to time out with too few players. Both cooldown and the cycle repeats.

Single on_level_up callback slot

Connection has only one on_level_up slot. cmd.incantation() overwrites the slot on every incantation and clears it when the ritual ends. Followers who miss the leader's LVL_UP broadcast stay frozen in INCANTATING until WAIT_TIMEOUT. Fix: use a list of handlers instead of a single slot.

parse_look in ai_vision.py is dead code

ai_vision.py exports parse_look(raw) but client.py's look() wrapper does the same parsing inline. parse_look is never called.

Dead reckoning desyncs on eject

When ejected, _pos and _facing are not updated. The wander coverage map becomes inaccurate until the player moves enough to re-sync naturally.

Clone this wiki locally