BT411/scratchpad/sim3.py

"""3-mech live combat simulation -- verify the designation fix end to end.

Three pods on one relay, each driving at the enemy with the trigger held, so they
actually converge, shoot, kill and respawn.  What this proves or disproves:

  * #57  every own-death should now produce a PLAYER_DEAD record and a completed
         respawn ([respawn] ... RESET ... latch cleared).  Before the fix, any
         target designation set deathPending and the death was swallowed.
  * the SWALLOWED warning must NOT appear -- if it does, a latch is still sticking.
  * #45  kills/deaths still expected WRONG cross-node (not fixed yet) -- record
         what each node thinks, as the baseline for that work.
  * stability: no crash across repeated death/respawn cycles.

Kills only the PIDs it spawns.
"""
import os
import re
import shutil
import subprocess
import sys
import time

REPO = r"C:\git\bt411"
CONTENT = os.path.join(REPO, "content")
EXE = os.path.join(REPO, "build", "Release", "btl4.exe")
EGG = "_sim3.egg"
RELAY_PORT = 2800
OUT = os.path.join(REPO, "scratchpad", "sim3")
os.makedirs(OUT, exist_ok=True)
for f in os.listdir(OUT):
    try: os.remove(os.path.join(OUT, f))
    except Exception: pass

RUN_SECONDS = int(sys.argv[1]) if len(sys.argv) > 1 else 260

procs = []
def spawn(cmd, env, tag):
    p = subprocess.Popen(cmd, cwd=CONTENT, env=env,
                         stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
    procs.append((tag, p))
    print("  %-8s pid %d" % (tag, p.pid))
    return p

base = dict(os.environ)
base.pop("BT_NO_PILOTLIST", None)

# relay
renv = dict(base)
relay_log = os.path.join(OUT, "relay.log")
p = subprocess.Popen(
    ["python", os.path.join(REPO, "tools", "btconsole.py"),
     "--relay", str(RELAY_PORT), EGG, "--bind", "127.0.0.1"],
    cwd=CONTENT, env=renv,
    stdout=open(relay_log, "w"), stderr=subprocess.STDOUT)
procs.append(("relay", p))
print("  relay    pid %d" % p.pid)
time.sleep(3)

# three combatants; #1 is the observer (gauges + screenshots)
for i in (1, 2, 3):
    e = dict(base)
    e.update({
        "BT_RELAY": "127.0.0.1:%d" % RELAY_PORT,
        "BT_SELF": "10.99.0.%d:1502" % i,
        "BT_START_INSIDE": "1",
        "BT_GOTO": "enemy",          # close on the nearest enemy
        "BT_AUTOFIRE": "1",          # hold the trigger
        "BT_MATCHLOG": "1",
        "BT_SCORE_LOG": "1",
        "BT_AUDIO_LOG": "1",
        "BT_HEAT_LOG": "1",    # coolant/heat state (Gitea #55 probe)   # exposes AreWatchersDelayed (Gitea #59 probe)
        "BT_LOG": os.path.join(OUT, "pod%d.log" % i),
    })
    if i == 2:
        e["BT_MP_FORCE_DMG"] = "1"   # fast kills: hit the victim once a second
    if i == 1:
        e["BT_DEV_GAUGES"] = "1"
        e["BT_SHOT"] = os.path.join(REPO, "scratchpad_sim3.png")
    spawn([EXE, "-egg", EGG, "-net", "1%d01" % i], e, "pod%d" % i)
    time.sleep(1.5)

print("\nrunning %ds ..." % RUN_SECONDS)
t0 = time.time()
shots = 0
try:
    while time.time() - t0 < RUN_SECONDS:
        time.sleep(20)
        src = os.path.join(REPO, "scratchpad_sim3.png")
        if os.path.exists(src):
            try:
                shutil.copy(src, os.path.join(OUT, "shot_%03d.png" % shots)); shots += 1
            except Exception:
                pass
        el = int(time.time() - t0)
        d = 0
        for i in (1, 2, 3):
            lp = os.path.join(OUT, "pod%d.log" % i)
            if os.path.exists(lp):
                d += len(re.findall(r"death cycle START", open(lp, errors="replace").read()))
        print("  t=%3ds  death-cycles so far: %d" % (el, d))
finally:
    for tag, p in procs:
        try: p.terminate()
        except Exception: pass
    time.sleep(2)

print("\n================ RESULTS ================")
for i in (1, 2, 3):
    lp = os.path.join(OUT, "pod%d.log" % i)
    if not os.path.exists(lp):
        print("pod%d: NO LOG" % i); continue
    t = open(lp, errors="replace").read()
    print("\n--- pod%d" % i)
    print("   death cycle START : %d" % len(re.findall(r"death cycle START", t)))
    print("   respawn RESET     : %d" % len(re.findall(r"RESET at drop zone", t)))
    print("   SWALLOWED (BAD)   : %d" % len(re.findall(r"SWALLOWED", t)))
    print("   emitter FIRED     : %d" % len(re.findall(r"\[emitter\] FIRED", t)))
    dl = re.findall(r"delayed=(\d+)", t)
    print("   watchers delayed=1: %d of %d samples (Gitea #59 -- must stay 0)"
          % (sum(1 for x in dl if x != "0"), len(dl)))
    print("   DESTROYED         : %d" % len(re.findall(r"\*\*\* .* DESTROYED", t)))
    for pat in (r"^\[respawn\].*$", r"^\[layout\].*$"):
        for l in re.findall(pat, t, re.M)[:4]:
            print("     " + l[:120])
    if re.search(r"exception|access violation|assert", t, re.I):
        print("   !! fault text present")

# matchlogs written this run
ml = os.path.join(CONTENT, "matchlogs")
recent = sorted((os.path.getmtime(os.path.join(ml, f)), f)
                for f in os.listdir(ml) if os.path.getmtime(os.path.join(ml, f)) > t0)
print("\n--- matchlogs from this run: %d" % len(recent))
for _m, f in recent:
    t = open(os.path.join(ml, f), errors="replace").read()
    print("   %-46s DEATH(M)=%d PLAYER_DEAD=%d RESPAWN=%d SCORE(kill)=%d" % (
        f[:46],
        len(re.findall(r"^DEATH.*inst=M", t, re.M)),
        len(re.findall(r"^PLAYER_DEAD", t, re.M)),
        len(re.findall(r"^RESPAWN", t, re.M)),
        len(re.findall(r"^SCORE.*type=2", t, re.M))))
print("\nshots: %d in %s" % (shots, OUT))