Map scanning

client.map.scan_kingdom finds a kingdom's castles, outposts, capitals and other map objects. It discovers the kingdom breadth-first from your castle's position, and a full scan can take a few minutes.

from empire_core import Kingdom, MapItemType

result = client.map.scan_kingdom(Kingdom.GREEN, item_types=[MapItemType.CASTLE])
print(f"{len(result.items)} items, {len(result.failed_chunks)} failed chunks")

Always check failed_chunks

A partial scan is not an empty kingdom, and only this field tells them apart.

What a scan returns

A ScanResult holds:

Field	What it is
items	Every MapAreaItem found, read by its area type's layout; empty with on_chunk.
objects	The owner records (MapObject) the chunks sent, by player id; empty with on_chunk.
kingdom	The kingdom scanned; every item carries it too.
failed_chunks	Chunks that still failed after their retries, or were left unscanned by a timeout, a cancel or a dropped connection.
content_chunks	Chunks that answered and held items.

An item's owner_id is below 0 for an NPC owner. NPCOwner (in empire_core.enums) names those ids, e.g. NPCOwner.ROBBER_BARON or NPCOwner.OUTPOST for an unclaimed outpost.

Requests go out back to back, as the game client sends its map requests; a live scan of 289 chunks at about 17 requests a second ran without a refusal. A chunk that times out, or is refused with a cooldown, is retried after a short backoff. chunk_delay adds a fixed wait before every request if you want one.

A session that drops mid-scan ends the scan with what it has, the rest in failed_chunks. The drop, a close by the server or a reset connection, is logged once, as a warning naming the client and the cause, and reported by client.on_disconnect; the scan adds one INFO line with the number of chunks it left, and the chunks themselves only at DEBUG.

Re-scanning cheaply

Feed content_chunks back into scan_chunks() to re-scan a known region without paying for the breadth-first discovery again, which takes roughly a third fewer requests. Run a full scan_kingdom() now and then to pick up content in chunks that were empty before.

discovery = client.map.scan_kingdom(Kingdom.GREEN, item_types=[MapItemType.CASTLE])

fresh = client.map.scan_chunks(
    Kingdom.GREEN, list(discovery.content_chunks), item_types=[MapItemType.CASTLE]
)

flowchart LR
    A[scan_kingdom<br/>breadth-first discovery] -->|content_chunks| B[scan_chunks<br/>known chunks only]
    B -->|often| B
    B -.->|now and then| A

Scanning with several accounts

The server limits the request rate per account, and each client keeps one request in flight, so one client scans a kingdom at roughly the pace of its round trips. scan_kingdom_with spreads one scan over several logged-in clients, each on its own thread. Lease them from an AccountPool:

from contextlib import ExitStack

from empire_core import Kingdom
from empire_core.map import scan_kingdom_with

with ExitStack() as leases:
    clients = [leases.enter_context(pool.leased(tag="scanner")) for _ in range(4)]
    result = scan_kingdom_with(clients, Kingdom.GREEN)
    # or re-scan known chunks: scan_kingdom_with(clients, Kingdom.GREEN, chunks=discovery.content_chunks)
print(f"{len(result.items)} items, {len(result.failed_chunks)} failed chunks")

Without chunks the clients discover the kingdom together from the castle there of the first client that has one; with chunks they scan those. Pass each client once. Every other argument means what it means for scan_kingdom(). Each client takes the next chunk nobody has taken yet, so the chunks interleave and a slow client takes fewer. The result merges every client's chunks; items and content_chunks come in the order the chunks answered.

A chunk that fails on one client is asked again by a client that has not tried it, and a client whose session drops, or whose account has not unlocked the kingdom, leaves the scan while the others take over its chunk. Only a chunk every remaining client has failed, or one left when no client remains, a timeout passes or cancel is set, ends in failed_chunks; when every client left because it has not unlocked the kingdom, that CommandError is raised to you. on_chunk is called from the clients' threads, one call at a time, so it needs no lock of its own. An exception it raises, or a KeyboardInterrupt while you wait, stops every client after the request it has in flight and is raised to you; no chunk reaches on_chunk after it.

Parsing a chunk's reply holds the GIL, so threads in one process overlap their waits for replies but not their parsing: one process never scans faster than the parse time of every chunk added up (scripts/bench_map_parse.py times one chunk). scripts/bench_multi_scan.py measures it offline with made-up 50 ms replies and 289 live-shaped chunks:

Clients	Wall time
1	16.7 s
8, threads in one process	2.4 s
8, separate processes	2.1 s

With replies at 20 ms the one process stays at about 2.2 s, the parse time, while separate processes reach 1.1 s. A live scan with 8 accounts in separate processes took 2.9 s, against about 16 s for one account. If you need more than one process can parse, run a scan per process, each with its interleaved slice chunks[i::n] of a discovery's content_chunks.

Concurrent scans of the same kingdom are not merged into one: each call scans on its own clients.

Keeping only what you need

A ScanResult keeps every item and owner record of the kingdom, and Python's full garbage collections walk all of them while the result lives: for a dense kingdom that is a pause of a tenth of a second or more each time. Pass on_chunk to take each chunk as it is read and keep only what you need; the scan then keeps nothing itself, and items and objects come back empty:

wanted_players = {1234, 5678}
castles: dict[tuple[int, int], int] = {}

def keep_castles(chunk, items, objects):
    for item in items:
        if item.owner_id in wanted_players:
            castles[(item.x, item.y)] = item.owner_id

result = client.map.scan_kingdom(Kingdom.GREEN, on_chunk=keep_castles)

on_chunk is called for every chunk that answered, empty ones too; failed chunks are still in failed_chunks, and content_chunks is filled as usual. An exception raised by on_chunk ends the scan.

Cancelling a scan

Pass a threading.Event as cancel to scan_kingdom() or scan_chunks(). Once it is set, the scan stops before its next chunk and returns what it has, with the chunks it did not scan in failed_chunks, as a timeout does:

import threading

stop = threading.Event()
result = client.map.scan_kingdom(Kingdom.GREEN, cancel=stop)
# on another thread: stop.set()

The chunk in flight is not abandoned: it ends with its reply or its request_timeout, so its reply cannot reach the next map request. The same event can cancel spy missions too (see Spy).

The session leaves its castle

A scan moves the session off the castle it had joined and drops client.state.get_mines(); get_joined_area() still names the castle, as the game keeps it. client.army methods join their castle again themselves; anything else castle-scoped needs client.castle.select() first.

Other map calls

• client.map.find_next(area_type) finds the nearest object of one area type, with the map rows around it.
• client.map.find_next_enemy_castle(x, y, index=0) finds an enemy castle near a position, as the game's "search enemy" button does. The button asks with index 0 to 9 in turn, so each press finds another castle; pass min_level and max_level to narrow it. Outposts count: on a live account, the first three indexes found three players' outposts. The request names no kingdom; pass kingdom for the one the reply's rows are read in.
• client.map.find_next_tower() finds the next Berimond tower, the button's search in Berimond.

Each find returns None when nothing matches, and found() on the reply gives the row it found. - client.map.scan_map_area(x1, y1, x2, y2) reads one rectangle of the map. - client.castle.join_area(x, y) joins an outpost, capital, metropolis or faction camp by its position.

Runnable example

examples/kingdom_scan.py

"""Scan a kingdom for castles, then re-scan only the chunks that had content.

A full ``client.map.scan_kingdom()`` walks the map by BFS from your own castle
and takes minutes. ``result.content_chunks`` records which chunks actually held
items, so a follow-up ``client.map.scan_chunks()`` refreshes the same region for
roughly a third fewer requests. Run the full scan occasionally, the targeted one
often.

    export GGE_USERNAME=your_user
    export GGE_PASSWORD=your_pass
    python examples/kingdom_scan.py
"""

import logging
import os
import sys

from empire_core import EmpireClient, EmpireError, Kingdom, MapItemType, ScanResult

logging.basicConfig(level=logging.INFO, format="%(levelname)s %(name)s: %(message)s")
logging.getLogger("websocket").setLevel(logging.WARNING)

KINGDOM = Kingdom.GREEN
ITEM_TYPES = [MapItemType.CASTLE]


def summarize(label: str, result: ScanResult) -> None:
    print(
        f"{label}: {len(result.items)} items, {len(result.objects)} objects, "
        f"{len(result.content_chunks)} chunks with content, "
        f"{len(result.failed_chunks)} failed chunks"
    )
    if result.failed_chunks:
        # Failed chunks are reported, not silently dropped: the scan is partial.
        print(f"  failed: {result.failed_chunks[:10]}{'...' if len(result.failed_chunks) > 10 else ''}")


def main() -> int:
    username = os.getenv("GGE_USERNAME")
    password = os.getenv("GGE_PASSWORD")
    if not username or not password:
        print("Set GGE_USERNAME and GGE_PASSWORD first.", file=sys.stderr)
        return 2

    client = EmpireClient(username=username, password=password)
    try:
        client.login()

        # Requests go out back to back; pass chunk_delay to wait between them.
        discovery = client.map.scan_kingdom(KINGDOM, item_types=ITEM_TYPES)
        summarize("discovery scan", discovery)

        if not discovery.content_chunks:
            print("Nothing found; skipping the targeted re-scan.")
            return 0

        fresh = client.map.scan_chunks(KINGDOM, list(discovery.content_chunks), item_types=ITEM_TYPES)
        summarize("targeted re-scan", fresh)

        for item in fresh.items[:10]:
            print(f"  {item}")

    except EmpireError as e:
        print(f"{type(e).__name__}: {e}", file=sys.stderr)
        return 1
    finally:
        client.close()

    return 0


if __name__ == "__main__":
    sys.exit(main())

API: MapService, ScanResult, scan_kingdom_with, MapAreaItem