mtg/tools/batch_run.py

#!/usr/bin/env python3
"""Run Forge simulations across many variants in parallel.

Forge is CPU-only (Java rules engine, no CUDA) and one instance uses ~1.5 cores,
so N workers on an M-core box wants N ~= M/2.

    python3 tools/batch_run.py --deck decks/reaper-king-no-more --games 30 --workers 6
    python3 tools/batch_run.py --deck decks/reaper-king-no-more --games 30 --workers 6 --only aggro-hate,go-wide
    python3 tools/batch_run.py --deck decks/reaper-king-no-more --resume   # skip variants already simulated

Each worker gets its OWN Forge profile dir, because Forge writes deck files into
~/.forge and concurrent runs would otherwise clobber each other.

Designed to be interrupted: progress is written per-variant as it completes, so
killing the job loses at most one variant's work.
"""
import sys, os, re, json, time, shutil, subprocess, argparse
from concurrent.futures import ThreadPoolExecutor, as_completed
from datetime import datetime, timezone

sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from deckctx import DeckContext, ROOT as TOOLS_ROOT

FORGE = os.path.expanduser("~/forge")
HEAP = os.environ.get("FORGE_HEAP", "2g")   # per-worker JVM heap

# Forge's -c is a PER-GAME wall clock; on expiry it abandons the game as a Draw.
# The 120s default abandoned 7 of our first 39 games (18%) -- and a draw means
# nobody wins, so those games were unwinnable. They cluster in long grindy games,
# exactly what an aristocrats deck is built to win, so the default biases our
# measured win rate DOWNWARD. 600s effectively removes the artifact.
CLOCK = 600

# Our seat's AI profile. "Default" = stock Forge (the A/B control arm); "Combo" =
# forge-overrides/ai/Combo.ai, tuned to actually operate a sacrifice deck.
# Opponents ALWAYS stay Default -- tuning them too would just move the baseline.
AI_PROFILE = os.environ.get("FORGE_AI_PROFILE", "Default")
# Install forge-overrides/cardsfolder/* into each worker's custom-cards dir.
# These un-blacklist cards Forge refuses to play (AI:RemoveDeck:All). Off by
# default so stock behaviour is the control arm.
USE_OVERLAY = os.environ.get("FORGE_OVERLAY", "0") == "1"
PRECONS = os.path.join(FORGE, "res", "quest", "commanderprecons")

# Rotating opponent pods, so results aren't specific to one matchup.
# Each pod is 3 decks; a variant faces every pod in turn.
PODS = [
    ["Faceless Menace [C19] [2019]", "Merciless Rage [C19] [2019]",
     "Arcane Maelstrom [C20] [2020]"],
    ["Mystic Intellect [C19] [2019]", "Primal Genesis [C19] [2019] ",
     "Enhanced Evolution [C20] [2020]"],
    ["Bedecked Brokers [NCC] [2022]", "Cabaretti Cacophony [NCC] [2022]",
     "Maestros Massacre [NCC] [2022]"],
    ["Lorehold Legacies [C21] [2021]", "Obscura Operation [NCC] [2022]",
     "Divine Convocation [MOC] [2023]"],
]


def variants(deck_root, only=None, resume=False):
    out = []
    for name in sorted(os.listdir(os.path.join(deck_root, "decks"))):
        d = os.path.join(deck_root, "decks", name)
        if not os.path.isdir(d) or not os.path.exists(os.path.join(d, "decklist.txt")):
            continue
        if only and name not in only:
            continue
        if resume and os.path.isdir(os.path.join(d, "sim")) and \
           any(f.endswith(".log") for f in os.listdir(os.path.join(d, "sim"))):
            continue
        out.append(name)
    return out


def worker_profile(wid, deck_root):
    """Give each worker its own Forge install + data dir.

    Forge reads its data path from forge.profile.properties *in the program
    directory* -- not from $HOME -- so isolating workers means one Forge dir
    each. The jars and res/ are symlinked (465MB) rather than copied, so a
    worker dir costs kilobytes.
    """
    home = os.path.expanduser("~")
    prog = os.path.join(home, f".forge-run/w{wid}")
    data = os.path.join(home, f".forge-run/data{wid}")
    os.makedirs(os.path.join(data, "decks", "commander"), exist_ok=True)
    os.makedirs(prog, exist_ok=True)

    for entry in os.listdir(FORGE):
        link = os.path.join(prog, entry)
        if entry == "forge.profile.properties" or os.path.lexists(link):
            continue
        os.symlink(os.path.join(FORGE, entry), link)

    with open(os.path.join(prog, "forge.profile.properties"), "w") as f:
        f.write(f"userDir={data}\ncacheDir={os.path.join(data, 'cache')}\n")

    # Card-script overlay: Forge reads <userDir>/custom/cards/*.txt after the
    # stock zip, so a same-named file replaces the stock rules for that card.
    custom = os.path.join(data, "custom", "cards")
    os.makedirs(custom, exist_ok=True)
    for f in os.listdir(custom):
        os.remove(os.path.join(custom, f))          # never leave a stale overlay
    if USE_OVERLAY:
        src = os.path.join(deck_root, "forge-overrides", "cardsfolder")
        for root_dir, _, files in os.walk(src):
            for fn in files:
                if fn.endswith(".txt"):
                    shutil.copy(os.path.join(root_dir, fn), custom)
    return prog, data


def run_variant(ctx, name, games, seed, wid, pod_idx):
    deck_root = ctx.dir
    prog, data = worker_profile(wid, deck_root)
    dst = os.path.join(data, "decks", "commander")
    stamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
    vdir = os.path.join(deck_root, "decks", name)
    simdir = os.path.join(vdir, "sim")
    os.makedirs(simdir, exist_ok=True)
    out = os.path.join(simdir, stamp)
    dck_name = ctx.dck_filename()

    # build this variant's .dck into the worker's own profile
    r = subprocess.run([sys.executable, os.path.join(TOOLS_ROOT, "tools", "to_dck.py"),
                        "--deck", deck_root,
                        os.path.join(vdir, "decklist.txt"),
                        "-o", os.path.join(dst, dck_name)],
                       capture_output=True, text=True)
    if r.returncode != 0:
        return name, None, f"to_dck failed: {r.stderr[:200]}"
    # capture which cards were proxied so reports never attribute a stand-in's
    # performance to the real card
    subs_used = dict(re.findall(r"^\s+(.+?)\s+->\s+(.+)$", r.stderr, re.M))

    pod = PODS[pod_idx % len(PODS)]
    for p in pod:
        src = os.path.join(PRECONS, p + ".dck")
        if os.path.exists(src):
            shutil.copy(src, dst)

    # Call java directly so heap size is controllable; forge.sh hardcodes -Xmx4096m,
    # which would exhaust RAM well before cores on a many-worker box.
    jar = next(f for f in os.listdir(FORGE)
               if f.startswith("forge-gui-desktop") and f.endswith(".jar"))
    cmd = ["java", f"-Xmx{HEAP}", "-Dio.netty.tryReflectionSetAccessible=true",
           "-Dfile.encoding=UTF-8", "-jar", jar,
           "sim", "-d", dck_name] + \
          [p + ".dck" for p in pod] + \
          ["-f", "Commander", "-n", str(games), "-s", str(seed),
           "-c", str(CLOCK),
           # profiles map to seats positionally; ours is always first
           "-a", AI_PROFILE, "Default", "Default", "Default"]
    t0 = time.time()
    with open(out + ".log", "w") as log:
        proc = subprocess.run(cmd, cwd=prog, stdout=subprocess.PIPE,
                              stderr=subprocess.STDOUT, text=True)
        for line in proc.stdout.splitlines():
            if not re.match(r"^(Phase:|Warning:|WARNING)", line) and \
               "not assigned to any set" not in line and "Upcoming set" not in line:
                log.write(line + "\n")

    subprocess.run([sys.executable, os.path.join(TOOLS_ROOT, "tools", "forge_report.py"),
                    out + ".log", out + ".json", out + ".txt", str(games), str(seed),
                    os.path.join(vdir, "decklist.txt"), ctx.forge_metadata_name()],
                   capture_output=True)

    # Stamp the run config so tuned and stock arms are never silently compared.
    try:
        d = json.load(open(out + ".json"))
        d["meta"]["config"] = {"ai_profile": AI_PROFILE, "opponents": "Default",
                               "card_overlay": USE_OVERLAY, "clock_seconds": CLOCK,
                               "pod": pod, "substitutions": subs_used}
        json.dump(d, open(out + ".json", "w"), indent=1)
    except (OSError, ValueError, KeyError):
        pass
    with open(os.path.join(vdir, "card_stats.txt"), "w") as cs:
        subprocess.run([sys.executable, os.path.join(TOOLS_ROOT, "tools", "card_stats.py"),
                        "--deck", deck_root, out + ".log"],
                       stdout=cs, stderr=subprocess.DEVNULL)
    return name, time.time() - t0, None


def main():
    global HEAP, CLOCK, AI_PROFILE, USE_OVERLAY
    ctx = DeckContext.from_args()   # pops --deck/-d out of sys.argv first
    ap = argparse.ArgumentParser()
    ap.add_argument("--games", type=int, default=30)
    ap.add_argument("--workers", type=int, default=max(1, (os.cpu_count() or 4) // 3))
    ap.add_argument("--seed", type=int, default=20260907)
    ap.add_argument("--only", default=None)
    ap.add_argument("--resume", action="store_true")
    ap.add_argument("--heap", default=None,
                    help="per-worker JVM heap, e.g. 2g (default 2g)")
    ap.add_argument("--clock", type=int, default=CLOCK,
                    help="seconds per game before Forge calls it a draw (default 600)")
    ap.add_argument("--ai-profile", default=AI_PROFILE,
                    help="our seat's AI profile: Default (stock) or Combo (tuned)")
    ap.add_argument("--overlay", action="store_true",
                    help="install card-script overrides that un-blacklist AI-refused cards")
    a = ap.parse_args()

    if a.heap:
        HEAP = a.heap
    CLOCK = a.clock
    AI_PROFILE = a.ai_profile
    USE_OVERLAY = USE_OVERLAY or a.overlay
    only = set(a.only.split(",")) if a.only else None
    vs = variants(ctx.dir, only, a.resume)
    if not vs:
        sys.exit("no variants to run")

    est = len(vs) * a.games * 75 / 3600 / a.workers
    print(f"{len(vs)} variants x {a.games} games, {a.workers} workers, "
          f"{HEAP} heap each, {CLOCK}s clock")
    print(f"AI profile: {AI_PROFILE} (opponents: Default)   "
          f"card overlay: {'ON' if USE_OVERLAY else 'off'}")
    print(f"opponent pods: {len(PODS)} (rotating)")
    print(f"estimated: {est:.1f}h\n", flush=True)

    done = 0
    t0 = time.time()
    with ThreadPoolExecutor(max_workers=a.workers) as ex:
        futs = {ex.submit(run_variant, ctx, v, a.games, a.seed + i, i % a.workers, i): v
                for i, v in enumerate(vs)}
        for f in as_completed(futs):
            name, secs, err = f.result()
            done += 1
            if err:
                print(f"[{done}/{len(vs)}] {name:20} FAILED — {err}", flush=True)
            else:
                print(f"[{done}/{len(vs)}] {name:20} {secs/60:5.1f} min "
                      f"(elapsed {(time.time()-t0)/3600:.1f}h)", flush=True)

    subprocess.run([sys.executable, os.path.join(TOOLS_ROOT, "tools", "compare.py"),
                    "--deck", ctx.dir], capture_output=True)
    print(f"\ntotal {(time.time()-t0)/3600:.1f}h — see {ctx.slug}/decks/README.md")


if __name__ == "__main__":
    main()