721 lines
30 KiB
Python
Executable File
721 lines
30 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""apptest - run the game through the LlamaMod app, invisibly and muted, for testing mods.
|
|
|
|
Every session NAME has its own LlamaMod data folder, X display (Xvfb), private PulseAudio
|
|
daemon (null sink only) and Wine prefix, so several agents can test in parallel:
|
|
|
|
<root>/<NAME>/data LLAMAMOD_DATA (config.json, original/ (hard links), game/, cd-active/, wine/)
|
|
<root>/<NAME>/harness state.json, pulse/ (socket, rt, home), logs, xvfb.log
|
|
|
|
root = $APPTEST_ROOT or /home/elias/.claude/jobs/90c6eed7/tmp
|
|
|
|
Commands (NAME first; --display only for 'start'/'run', remembered afterwards):
|
|
apptest.py setup NAME install from the ISO (shared hard-linked template)
|
|
apptest.py apply NAME [ids] [--opt id.k=v]... apply mods (none = original) via the app CLI
|
|
apptest.py unlock NAME SLOT WORLD CHAPTER unlock up to WORLD/CHAPTER, map cursor there
|
|
apptest.py start NAME --display N Xvfb :N + private pulse + game (app --launch)
|
|
apptest.py tolevel NAME [--world W] main menu -> slot 3 -> map -> level -> gameplay
|
|
apptest.py keys NAME Up:1.5 x sleep=1 a+x:0.5 shot:name ... key sequence (hold s, default 0.2)
|
|
apptest.py shot NAME out.png [--small] screenshot (1920x1080 PNG, --small = 960x540 JPEG)
|
|
apptest.py ocr NAME [WxH+X+Y] German OCR of the screen (or a crop)
|
|
apptest.py status NAME game/window/crash/audio state (exit 3 = crashed)
|
|
apptest.py stop NAME kill game (wineserver -k), Xvfb and pulse of NAME
|
|
apptest.py run NAME --display N --mods a,b [--opt id.k=v]... --level W C [--seq "..."] --out DIR [--keep]
|
|
all in one; exit 0 ok, 3 crash, 4 timeout, 2 usage
|
|
|
|
Options of every command: --root DIR, --jar FILE (default: newest build/libs/*-all.jar),
|
|
--iso FILE (default ~/Downloads/GROOVE_DK.iso). Python: `from apptest import Session`.
|
|
Environment: APPTEST_CP=<extra classpath> + APPTEST_JAVA_OPTS="-Dllamamod.extraMods=<class>" test
|
|
mods that are not registered yet (e.g. APPTEST_CP=build/classes/java/test).
|
|
|
|
Safety: never uses display :0, never touches ~/.wine-lama, kills only PIDs it started itself
|
|
(checked against /proc/PID/cmdline) and the wineserver of its own prefix.
|
|
"""
|
|
import fcntl
|
|
import json
|
|
import os
|
|
import shutil
|
|
import signal
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
|
|
HOME = os.path.expanduser("~")
|
|
PROJECT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
|
DEFAULT_ROOT = os.environ.get("APPTEST_ROOT", "/home/elias/.claude/jobs/90c6eed7/tmp")
|
|
DEFAULT_ISO = os.path.join(HOME, "Downloads", "GROOVE_DK.iso")
|
|
TEMPLATE = "_llamamod-template"
|
|
SCREEN = (1920, 1080)
|
|
FULL = "%dx%d+0+0" % SCREEN
|
|
SLOT_CROP = "1600x280+160+740" # save-slot text "Gespeichertes Spiel N / <level> Kapitel n"
|
|
MENU_CROP = "700x300+1150+650" # main menu items (bottom right)
|
|
|
|
|
|
class Crash(Exception):
|
|
pass
|
|
|
|
|
|
class Timeout(Exception):
|
|
pass
|
|
|
|
|
|
def log(*a):
|
|
print("[apptest]", *a, flush=True)
|
|
|
|
|
|
def find_java():
|
|
for j in (os.environ.get("JAVA"), os.path.join(HOME, ".jdks", "openjdk-25", "bin", "java")):
|
|
if j and os.access(j, os.X_OK):
|
|
return j
|
|
return shutil.which("java") or "java"
|
|
|
|
|
|
def find_jar():
|
|
if os.environ.get("APPTEST_JAR"):
|
|
return os.environ["APPTEST_JAR"]
|
|
d = os.path.join(PROJECT, "build", "libs")
|
|
jars = [os.path.join(d, f) for f in os.listdir(d) if f.endswith("-all.jar")] if os.path.isdir(d) else []
|
|
if not jars:
|
|
raise SystemExit("no fat jar - run ./gradlew fatJar in %s" % PROJECT)
|
|
return max(jars, key=os.path.getmtime)
|
|
|
|
|
|
def fuzzy_in(word, text, ratio=0.75):
|
|
import difflib
|
|
if word in text:
|
|
return True
|
|
toks = text.split()
|
|
n = len(word.split())
|
|
for i in range(len(toks) - n + 1):
|
|
if difflib.SequenceMatcher(None, word, " ".join(toks[i:i + n])).ratio() >= ratio:
|
|
return True
|
|
return False
|
|
|
|
|
|
def _cmdline(pid):
|
|
try:
|
|
return open("/proc/%d/cmdline" % pid, "rb").read().split(b"\0")
|
|
except OSError:
|
|
return []
|
|
|
|
|
|
def _environ(pid):
|
|
try:
|
|
return open("/proc/%d/environ" % pid, "rb").read().split(b"\0")
|
|
except OSError:
|
|
return []
|
|
|
|
|
|
def _kill_own(pid, expect):
|
|
"""Kill pid only if its executable name contains `expect` (guards against reused PIDs)."""
|
|
if not pid:
|
|
return
|
|
cmd = _cmdline(pid)
|
|
if not cmd or expect.encode() not in os.path.basename(cmd[0]):
|
|
return
|
|
try:
|
|
os.kill(pid, signal.SIGTERM)
|
|
for _ in range(30):
|
|
time.sleep(0.1)
|
|
if not _cmdline(pid):
|
|
return
|
|
os.kill(pid, signal.SIGKILL)
|
|
except ProcessLookupError:
|
|
pass
|
|
|
|
|
|
class Session:
|
|
def __init__(self, name, display=None, root=DEFAULT_ROOT, jar=None, iso=DEFAULT_ISO):
|
|
if not name or "/" in name or name.startswith("_"):
|
|
raise SystemExit("bad session name %r" % name)
|
|
self.name = name
|
|
self.root = root
|
|
self.dir = os.path.join(root, name)
|
|
self.data = os.path.join(self.dir, "data")
|
|
self.hdir = os.path.join(self.dir, "harness")
|
|
self.pulse_dir = os.path.join(self.hdir, "pulse")
|
|
self.state_file = os.path.join(self.hdir, "state.json")
|
|
self.jar = jar
|
|
self.iso = iso
|
|
os.makedirs(self.hdir, exist_ok=True)
|
|
self.state = {}
|
|
if os.path.exists(self.state_file):
|
|
self.state = json.load(open(self.state_file))
|
|
if display is not None:
|
|
d = ":%s" % str(display).lstrip(":")
|
|
if self.state.get("display") not in (None, d) and self.state.get("xvfb_pid") \
|
|
and _cmdline(self.state["xvfb_pid"]):
|
|
raise SystemExit("session %s already runs on %s" % (name, self.state["display"]))
|
|
self.state["display"] = d
|
|
self._save()
|
|
self.display = self.state.get("display")
|
|
if self.display in (":0", "0"):
|
|
raise SystemExit("refusing display :0")
|
|
|
|
def _save(self):
|
|
json.dump(self.state, open(self.state_file, "w"), indent=1)
|
|
|
|
# ------------------------------------------------------------ environment
|
|
@property
|
|
def prefix(self):
|
|
return os.path.join(self.data, "wine")
|
|
|
|
@property
|
|
def pulse_server(self):
|
|
return "unix:" + os.path.join(self.pulse_dir, "native")
|
|
|
|
def env(self):
|
|
e = dict(os.environ)
|
|
e["LLAMAMOD_DATA"] = self.data
|
|
e.pop("DBUS_SESSION_BUS_ADDRESS", None)
|
|
if self.display:
|
|
e["DISPLAY"] = self.display
|
|
e["PULSE_SERVER"] = self.pulse_server
|
|
e.pop("PULSE_SINK", None)
|
|
return e
|
|
|
|
def app(self, *args, check=True, timeout=1800):
|
|
"""Runs the LlamaMod CLI (fat jar) for this data folder."""
|
|
jar = self.jar or find_jar()
|
|
cmd = [find_java(), "-Djava.awt.headless=true"]
|
|
if self.display:
|
|
cmd.append("-Dllamamod.gameDisplay=" + self.display)
|
|
cmd += os.environ.get("APPTEST_JAVA_OPTS", "").split()
|
|
if os.environ.get("APPTEST_CP"):
|
|
cmd += ["-cp", jar + os.pathsep + os.environ["APPTEST_CP"], "com.eliasmueller.llamamod.Main"]
|
|
else:
|
|
cmd += ["-jar", jar]
|
|
cmd += list(args)
|
|
r = subprocess.run(cmd, env=self.env(), stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
|
|
text=True, timeout=timeout, start_new_session=True)
|
|
if check and r.returncode != 0:
|
|
raise SystemExit("llamamod %s failed (%d):\n%s" % (" ".join(args), r.returncode, r.stdout))
|
|
return r.stdout
|
|
|
|
# ------------------------------------------------------------ setup
|
|
def setup(self):
|
|
"""Installs the game into this session's data folder (once)."""
|
|
cfgp = os.path.join(self.data, "config.json")
|
|
if os.path.exists(cfgp) and json.load(open(cfgp)).get("layout", 0) >= 1:
|
|
self._config_defaults()
|
|
return
|
|
tdata = os.path.join(self.root, TEMPLATE, "data")
|
|
os.makedirs(os.path.dirname(tdata), exist_ok=True)
|
|
with open(os.path.join(self.root, TEMPLATE + ".lock"), "w") as lk:
|
|
fcntl.flock(lk, fcntl.LOCK_EX) # parallel agents: install the template only once
|
|
tcfg = os.path.join(tdata, "config.json")
|
|
if not (os.path.exists(tcfg) and json.load(open(tcfg)).get("layout", 0) >= 1):
|
|
if not os.path.exists(self.iso):
|
|
raise SystemExit("ISO not found: %s" % self.iso)
|
|
log("installing template from %s (once) ..." % self.iso)
|
|
r = subprocess.run([find_java(), "-Djava.awt.headless=true", "-jar", self.jar or find_jar(),
|
|
"--install", self.iso], env=dict(os.environ, LLAMAMOD_DATA=tdata),
|
|
stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True)
|
|
if r.returncode:
|
|
raise SystemExit("template install failed:\n" + r.stdout)
|
|
os.makedirs(self.data, exist_ok=True)
|
|
if not os.path.isdir(os.path.join(self.data, "original")):
|
|
# hard links: the originals are never written (mods write to cache/, game/ is a copy)
|
|
subprocess.run(["cp", "-al", os.path.join(tdata, "original"), os.path.join(self.data, "original")],
|
|
check=True)
|
|
log("setting up %s ..." % self.data)
|
|
self.app("--install", self.iso)
|
|
self._config_defaults()
|
|
|
|
def _config_defaults(self):
|
|
cfgp = os.path.join(self.data, "config.json")
|
|
c = json.load(open(cfgp))
|
|
c.update(screenW=SCREEN[0], screenH=SCREEN[1], virtualDesktop=True)
|
|
json.dump(c, open(cfgp, "w"), indent=2)
|
|
|
|
def apply(self, mods=(), opts=()):
|
|
"""mods: ids; opts: 'id.key=value' strings. Returns the CLI output."""
|
|
args = ["--apply"]
|
|
if mods:
|
|
args.append(",".join(mods))
|
|
for o in opts:
|
|
args += ["--opt", o]
|
|
out = self.app(*args)
|
|
log(out.strip().splitlines()[-1] if out.strip() else "applied")
|
|
return out
|
|
|
|
def unlock(self, slot, world, chapter):
|
|
return self.app("--unlock", str(slot), str(world), str(chapter), "--cursor")
|
|
|
|
# ------------------------------------------------------------ processes
|
|
def game_pids(self, what=b"groove.exe"):
|
|
want = ("WINEPREFIX=" + self.prefix).encode()
|
|
out = []
|
|
for p in os.listdir("/proc"):
|
|
if not p.isdigit():
|
|
continue
|
|
cmd = b" ".join(_cmdline(int(p))).lower()
|
|
if what in cmd and want in _environ(int(p)):
|
|
out.append(int(p))
|
|
return out
|
|
|
|
def _start_xvfb(self):
|
|
if self.state.get("xvfb_pid") and _cmdline(self.state["xvfb_pid"]):
|
|
return
|
|
n = self.display.lstrip(":")
|
|
if os.path.exists("/tmp/.X%s-lock" % n) or os.path.exists("/tmp/.X11-unix/X%s" % n):
|
|
raise SystemExit("display %s is already in use by someone else - pick another number" % self.display)
|
|
xl = open(os.path.join(self.hdir, "xvfb.log"), "w")
|
|
p = subprocess.Popen(["Xvfb", self.display, "-screen", "0", "%dx%dx24" % SCREEN, "-nolisten", "tcp",
|
|
"+extension", "GLX"], stdout=xl, stderr=xl, start_new_session=True)
|
|
self.state["xvfb_pid"] = p.pid
|
|
self._save()
|
|
for _ in range(100):
|
|
if subprocess.run(["xdpyinfo"], env=self.env(), stdout=subprocess.DEVNULL,
|
|
stderr=subprocess.DEVNULL).returncode == 0:
|
|
return
|
|
time.sleep(0.1)
|
|
raise SystemExit("Xvfb did not start (see harness/xvfb.log)")
|
|
|
|
def _start_pulse(self):
|
|
if self.state.get("pulse_pid") and _cmdline(self.state["pulse_pid"]):
|
|
return
|
|
d = self.pulse_dir
|
|
for sub in ("rt", "home"):
|
|
os.makedirs(os.path.join(d, sub), exist_ok=True)
|
|
os.chmod(os.path.join(d, "rt"), 0o700)
|
|
open(os.path.join(d, "home", ".pulse-cookie"), "a").close()
|
|
sock = os.path.join(d, "native")
|
|
if os.path.exists(sock):
|
|
os.remove(sock)
|
|
cfg = os.path.join(d, "default.pa")
|
|
open(cfg, "w").write("load-module module-null-sink sink_name=mute\nset-default-sink mute\n"
|
|
"load-module module-native-protocol-unix socket=%s auth-anonymous=1\n" % sock)
|
|
env = {k: v for k, v in os.environ.items() if k != "DBUS_SESSION_BUS_ADDRESS"}
|
|
env.update(DBUS_SESSION_BUS_ADDRESS="unix:path=/nonexistent", XDG_RUNTIME_DIR=os.path.join(d, "rt"),
|
|
HOME=os.path.join(d, "home"))
|
|
pl = open(os.path.join(self.hdir, "pulse.log"), "w")
|
|
p = subprocess.Popen(["pulseaudio", "-n", "-F", cfg, "--daemonize=no", "--exit-idle-time=-1",
|
|
"--use-pid-file=no"], env=env, stdout=pl, stderr=pl, start_new_session=True)
|
|
self.state["pulse_pid"] = p.pid
|
|
self._save()
|
|
for _ in range(100):
|
|
if os.path.exists(sock) and self.pactl("info").returncode == 0:
|
|
return
|
|
time.sleep(0.1)
|
|
raise SystemExit("private pulseaudio did not start (see harness/pulse.log)")
|
|
|
|
def pactl(self, *args):
|
|
return subprocess.run(["pactl", "-s", self.pulse_server] + list(args), stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE, text=True,
|
|
env={k: v for k, v in os.environ.items() if not k.startswith("PULSE")})
|
|
|
|
def audio_clients(self):
|
|
"""application.name of the sink inputs on the private server (the game: "groove.exe")."""
|
|
r = self.pactl("list", "sink-inputs")
|
|
return [l.split("=", 1)[1].strip().strip('"') for l in r.stdout.splitlines()
|
|
if l.strip().startswith("application.name")]
|
|
|
|
def start(self, timeout=240):
|
|
if not self.display:
|
|
raise SystemExit("start needs --display N")
|
|
if self.game_pids():
|
|
log("game already running")
|
|
return
|
|
self.setup()
|
|
self._start_xvfb()
|
|
self._start_pulse()
|
|
log("launching on %s (first start creates the Wine prefix, ~1 min) ..." % self.display)
|
|
out = self.app("--launch")
|
|
log(out.strip().splitlines()[-1] if out.strip() else "launched")
|
|
t0 = time.time()
|
|
while time.time() - t0 < timeout:
|
|
if self.find_window() and self.game_pids():
|
|
log("window after %.0f s" % (time.time() - t0))
|
|
break
|
|
time.sleep(1)
|
|
else:
|
|
raise Timeout("no game window after %d s (see data/logs/game.log)" % timeout)
|
|
for _ in range(30):
|
|
c = self.crashed()
|
|
if c:
|
|
raise Crash(c)
|
|
if "groove.exe" in " ".join(self.audio_clients()).lower():
|
|
log("audio: groove.exe plays into the private null sink")
|
|
break
|
|
time.sleep(1)
|
|
else:
|
|
log("WARNING: no groove.exe sink input on the private pulse server yet")
|
|
|
|
def stop(self):
|
|
if self.game_pids(b"") or os.path.isdir(self.prefix):
|
|
env = dict(self.env(), WINEPREFIX=self.prefix)
|
|
subprocess.run(["wineserver", "-k"], env=env, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
|
|
time.sleep(1)
|
|
for pid in self.game_pids(b""):
|
|
try:
|
|
os.kill(pid, signal.SIGKILL)
|
|
except ProcessLookupError:
|
|
pass
|
|
_kill_own(self.state.pop("pulse_pid", None), "pulseaudio")
|
|
_kill_own(self.state.pop("xvfb_pid", None), "Xvfb")
|
|
self._save()
|
|
log("stopped")
|
|
|
|
# ------------------------------------------------------------ X11
|
|
def _x(self, *args):
|
|
assert self.display and self.display != ":0"
|
|
return subprocess.run(["xdotool"] + list(args), env=self.env(), stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE, text=True)
|
|
|
|
def find_window(self):
|
|
for pat in ("Wine desktop", "^LlamaMod", "^Groove$"):
|
|
ids = self._x("search", "--name", pat).stdout.split()
|
|
if ids:
|
|
return ids[-1]
|
|
return None
|
|
|
|
def _focus(self):
|
|
wid = self.find_window()
|
|
if not wid:
|
|
raise Crash("game window gone")
|
|
self._x("windowfocus", "--sync", wid)
|
|
|
|
def key(self, k, hold=0.2):
|
|
"""Press keys (xdotool names, 'a+b' together) for `hold` seconds; DirectInput misses <0.1 s."""
|
|
ks = k.split("+")
|
|
self._focus()
|
|
for x in ks:
|
|
self._x("keydown", x)
|
|
time.sleep(float(hold))
|
|
for x in reversed(ks):
|
|
self._x("keyup", x)
|
|
|
|
def keys(self, seq, gap=0.3, outdir=None):
|
|
"""Tokens: KEY[:hold], a+b[:hold], sleep=S, shot:NAME (needs outdir), dialog."""
|
|
shots = []
|
|
for item in seq.split() if isinstance(seq, str) else seq:
|
|
c = self.crashed()
|
|
if c:
|
|
raise Crash(c)
|
|
if item.startswith("sleep="):
|
|
time.sleep(float(item[6:]))
|
|
elif item.startswith("shot:"):
|
|
f = os.path.join(outdir or self.hdir, "seq_%02d_%s.jpg" % (len(shots), item[5:]))
|
|
shots.append(self.shot(f, small=True))
|
|
elif item == "dialog":
|
|
n = 0
|
|
while self.classify() == "letterbox" and n < 12:
|
|
self.key("x")
|
|
time.sleep(1.0)
|
|
n += 1
|
|
else:
|
|
k, _, h = item.partition(":")
|
|
self.key(k, float(h) if h else 0.2)
|
|
time.sleep(gap)
|
|
return shots
|
|
|
|
def shot(self, path, small=False):
|
|
os.makedirs(os.path.dirname(os.path.abspath(path)), exist_ok=True)
|
|
extra = ["-resize", "50%", "-quality", "80"] if small else []
|
|
r = subprocess.run(["import", "-display", self.display, "-window", "root", "-crop", FULL, "+repage"]
|
|
+ extra + [path], env=self.env(), stderr=subprocess.PIPE, text=True)
|
|
if r.returncode:
|
|
raise SystemExit("import: " + r.stderr)
|
|
return path
|
|
|
|
def _grab(self, crop, fmt, resize=None):
|
|
cmd = ["import", "-display", self.display, "-window", "root", "-crop", crop, "+repage"]
|
|
if resize:
|
|
cmd += ["-resize", resize]
|
|
return subprocess.run(cmd + ["-depth", "8", fmt + ":-"], stdout=subprocess.PIPE, env=self.env()).stdout
|
|
|
|
def ocr(self, crop=FULL, threshold=None, psm=11, scale="50%"):
|
|
"""Lower-case German OCR text; threshold (%) keeps only bright text (selected menu item)."""
|
|
raw = self._grab(crop, "ppm")
|
|
filt = ["-threshold", "%d%%" % threshold] if threshold else []
|
|
img = subprocess.run(["convert", "ppm:-", "-colorspace", "Gray"] + filt + ["-negate", "-resize", scale,
|
|
"pgm:-"],
|
|
input=raw, stdout=subprocess.PIPE).stdout
|
|
r = subprocess.run(["tesseract", "-", "-", "-l", "deu", "--psm", str(psm)], input=img,
|
|
stdout=subprocess.PIPE, stderr=subprocess.DEVNULL)
|
|
return " ".join(r.stdout.decode("utf-8", "replace").lower().split())
|
|
|
|
def slot_text(self):
|
|
return self.ocr(SLOT_CROP, threshold=70, psm=6) + " | " + self.ocr(SLOT_CROP, threshold=82, psm=6)
|
|
|
|
def thumb(self):
|
|
return self._grab(FULL, "gray", "48x27!")
|
|
|
|
@staticmethod
|
|
def diff(a, b):
|
|
if not a or not b or len(a) != len(b):
|
|
return 255.0
|
|
return sum(abs(x - y) for x, y in zip(a, b)) / len(a)
|
|
|
|
def wait_stable(self, still=1.5, timeout=30, thresh=1.0):
|
|
t0 = time.time()
|
|
last, since = self.thumb(), time.time()
|
|
while time.time() - t0 < timeout:
|
|
time.sleep(0.4)
|
|
cur = self.thumb()
|
|
if self.diff(cur, last) > thresh or sum(cur) / max(1, len(cur)) < 3:
|
|
since = time.time()
|
|
elif time.time() - since >= still:
|
|
return cur
|
|
last = cur
|
|
return last
|
|
|
|
def classify(self):
|
|
"""black / letterbox (fly-by, cut scene) / map (world 9 beige map) / bright."""
|
|
b = self._grab(FULL, "rgb", "48x27!")
|
|
if len(b) != 48 * 27 * 3:
|
|
return "none"
|
|
px = [tuple(b[i:i + 3]) for i in range(0, len(b), 3)]
|
|
rows = [px[y * 48:(y + 1) * 48] for y in range(27)]
|
|
|
|
def mean(ys):
|
|
v = [sum(c) / 3 for y in ys for c in rows[y]]
|
|
return sum(v) / len(v)
|
|
if mean(range(27)) < 6:
|
|
return "black"
|
|
if mean((0,)) < 4 and mean((26,)) < 4 and mean(range(6, 18)) > 15:
|
|
return "letterbox"
|
|
beige = sum(1 for (r, g, bb) in px if 180 < r < 230 and 150 < g < 200 and 85 < bb < 140 and r > g > bb)
|
|
return "map" if beige > 0.75 * len(px) else "bright"
|
|
|
|
def hud_visible(self):
|
|
b = self._grab("900x110+100+10", "rgb", "25%")
|
|
return sum(1 for i in range(0, len(b) - 2, 3) if b[i] > 200 and b[i + 1] > 150 and b[i + 2] < 90) > 150
|
|
|
|
# ------------------------------------------------------------ crash detection
|
|
def crashed(self):
|
|
"""None, or why the game is considered crashed."""
|
|
if not self.game_pids():
|
|
return "groove.exe process gone"
|
|
gl = os.path.join(self.data, "logs", "game.log")
|
|
try:
|
|
with open(gl, "rb") as f:
|
|
f.seek(max(0, os.path.getsize(gl) - 65536))
|
|
tail = f.read().decode("latin-1")
|
|
for marker in ("Unhandled exception", "Unhandled page fault", "wine: Unhandled"):
|
|
if marker in tail:
|
|
return "game.log: " + tail[tail.index(marker):].splitlines()[0]
|
|
except OSError:
|
|
pass
|
|
if self.game_pids(b"winedbg"):
|
|
return "winedbg running (unhandled exception)"
|
|
if not self.find_window():
|
|
return "game window gone"
|
|
return None
|
|
|
|
def wait_for(self, pred, timeout, what, poll=0.7):
|
|
t0 = time.time()
|
|
while time.time() - t0 < timeout:
|
|
c = self.crashed()
|
|
if c:
|
|
raise Crash(c)
|
|
v = pred()
|
|
if v:
|
|
return v
|
|
time.sleep(poll)
|
|
raise Timeout("timeout (%d s) waiting for %s" % (timeout, what))
|
|
|
|
# ------------------------------------------------------------ navigation
|
|
def tolevel(self, world=None, skip_flyby=True):
|
|
"""Main menu -> "Spiel starten" -> slot 3 (rightmost) -> map (cursor from --unlock ... --cursor)
|
|
-> Enter -> skip the fly-by -> gameplay. Returns timings."""
|
|
T = {}
|
|
t0 = tx = time.time()
|
|
while True:
|
|
c = self.crashed()
|
|
if c:
|
|
raise Crash(c)
|
|
if fuzzy_in("spiel starten", self.ocr(MENU_CROP, threshold=75, psm=6, scale="100%")):
|
|
log("main menu")
|
|
time.sleep(0.5)
|
|
self.key("Return")
|
|
break
|
|
if fuzzy_in("gespeichertes", self.slot_text()):
|
|
break
|
|
if time.time() - tx > 40:
|
|
log("no menu yet -> Return (skips logo/demo)")
|
|
self.key("Return")
|
|
tx = time.time()
|
|
if time.time() - t0 > 120:
|
|
raise Timeout("main menu not reached")
|
|
time.sleep(1)
|
|
T["menu"] = round(time.time() - t0, 1)
|
|
self.wait_for(lambda: fuzzy_in("gespeichertes", self.slot_text()), 20, "save slot screen")
|
|
time.sleep(0.4)
|
|
for _ in range(2): # 3-slot slider without wrap: slot 3 is the rightmost
|
|
self.key("Right", 0.3)
|
|
time.sleep(0.3)
|
|
for i in range(3):
|
|
time.sleep(0.6)
|
|
t = self.slot_text()
|
|
if fuzzy_in("kapitel", t) and not fuzzy_in("leer", t):
|
|
log("slot: %s" % t.split(" | ")[0])
|
|
break
|
|
self.key("Right", 0.3)
|
|
else:
|
|
log("WARNING: slot 3 text not recognised (%s)" % t)
|
|
self.key("Return")
|
|
T["slot"] = round(time.time() - t0, 1)
|
|
if world == 9:
|
|
self.wait_for(lambda: self.classify() == "map", 30, "world 9 map")
|
|
time.sleep(1.0)
|
|
else:
|
|
self.wait_for(lambda: self.classify() not in ("black", "none"), 30, "map")
|
|
self.wait_stable(still=1.5, timeout=20)
|
|
self.key("Return")
|
|
t_load = time.time()
|
|
T["map"] = round(t_load - t0, 1)
|
|
skipped, good, states = False, 0, []
|
|
while time.time() - t_load < 150:
|
|
c = self.crashed()
|
|
if c:
|
|
raise Crash(c)
|
|
st = self.classify()
|
|
if not states or states[-1] != st:
|
|
states.append(st)
|
|
log("state: %s (%.1f s after level start)" % (st, time.time() - t_load))
|
|
if st == "letterbox" and skip_flyby and not skipped:
|
|
time.sleep(0.5)
|
|
self.key("Return")
|
|
skipped = True
|
|
elif st == "letterbox" and skipped and time.time() - t_load > 15:
|
|
self.key("x")
|
|
el = time.time() - t_load
|
|
if st == "bright" and el > 2 and (self.hud_visible() or "letterbox" in states or el > 25):
|
|
good += 1
|
|
if good >= 3:
|
|
break
|
|
else:
|
|
good = 0
|
|
time.sleep(0.6)
|
|
else:
|
|
raise Timeout("gameplay not reached (states %s)" % states)
|
|
T["gameplay"] = round(time.time() - t0, 1)
|
|
T["states"] = states
|
|
log("gameplay after %.1f s, states %s" % (T["gameplay"], states))
|
|
return T
|
|
|
|
def status(self):
|
|
s = {"display": self.display, "data": self.data, "game_pids": self.game_pids(),
|
|
"window": self.find_window() if self.display and self.state.get("xvfb_pid") else None,
|
|
"xvfb": bool(self.state.get("xvfb_pid") and _cmdline(self.state["xvfb_pid"])),
|
|
"pulse": bool(self.state.get("pulse_pid") and _cmdline(self.state["pulse_pid"]))}
|
|
s["audio_clients"] = self.audio_clients() if s["pulse"] else []
|
|
s["crashed"] = self.crashed() if s["game_pids"] and s["xvfb"] else None
|
|
return s
|
|
|
|
|
|
def run(sess, mods, opts, level, seq, out, keep=False):
|
|
os.makedirs(out, exist_ok=True)
|
|
report = {"mods": mods, "opts": opts, "level": level}
|
|
code = 0
|
|
try:
|
|
sess.setup()
|
|
report["apply"] = sess.apply(mods, opts).strip().splitlines()[-3:]
|
|
if level:
|
|
sess.unlock(3, level[0], level[1])
|
|
sess.start()
|
|
if level:
|
|
report["tolevel"] = sess.tolevel(world=level[0])
|
|
sess.shot(os.path.join(out, "00_start.png"))
|
|
if seq:
|
|
report["shots"] = sess.keys(seq, outdir=out)
|
|
c = sess.crashed()
|
|
if c:
|
|
raise Crash(c)
|
|
sess.shot(os.path.join(out, "99_end.jpg"), small=True)
|
|
report["audio_clients"] = sess.audio_clients()
|
|
except Crash as e:
|
|
report["error"] = "crash: %s" % e
|
|
code = 3
|
|
except Timeout as e:
|
|
report["error"] = str(e)
|
|
code = 4
|
|
finally:
|
|
if code and sess.state.get("xvfb_pid"):
|
|
try:
|
|
sess.shot(os.path.join(out, "zz_failure.jpg"), small=True)
|
|
except SystemExit:
|
|
pass
|
|
gl = os.path.join(sess.data, "logs", "game.log")
|
|
if os.path.exists(gl):
|
|
shutil.copy(gl, os.path.join(out, "game.log"))
|
|
al = os.path.join(sess.data, "logs", "apply.log")
|
|
if os.path.exists(al):
|
|
shutil.copy(al, os.path.join(out, "apply.log"))
|
|
json.dump(report, open(os.path.join(out, "report.json"), "w"), indent=1)
|
|
if not keep:
|
|
sess.stop()
|
|
log("result: %s" % (report.get("error") or "OK"))
|
|
return code
|
|
|
|
|
|
def main(argv):
|
|
if len(argv) < 2 or argv[0] in ("-h", "--help"):
|
|
print(__doc__)
|
|
return 2
|
|
cmd, name, rest = argv[0], argv[1], argv[2:]
|
|
kw, pos, opts = {}, [], []
|
|
i = 0
|
|
while i < len(rest):
|
|
a = rest[i]
|
|
if a in ("--display", "--root", "--jar", "--iso", "--mods", "--world", "--seq", "--out") and i + 1 < len(rest):
|
|
kw[a[2:]] = rest[i + 1]
|
|
i += 2
|
|
elif a == "--opt" and i + 1 < len(rest):
|
|
opts.append(rest[i + 1])
|
|
i += 2
|
|
elif a == "--level" and i + 2 < len(rest):
|
|
kw["level"] = (int(rest[i + 1]), int(rest[i + 2]))
|
|
i += 3
|
|
else:
|
|
pos.append(a)
|
|
i += 1
|
|
s = Session(name, display=kw.get("display"), root=kw.get("root", DEFAULT_ROOT), jar=kw.get("jar"),
|
|
iso=kw.get("iso", DEFAULT_ISO))
|
|
try:
|
|
if cmd == "setup":
|
|
s.setup()
|
|
elif cmd == "apply":
|
|
ids = [x for p in pos for x in p.split(",") if x] or [x for x in kw.get("mods", "").split(",") if x]
|
|
print(s.apply(ids, opts))
|
|
elif cmd == "unlock":
|
|
print(s.unlock(*pos[:3]))
|
|
elif cmd == "start":
|
|
s.start()
|
|
elif cmd == "tolevel":
|
|
print(json.dumps(s.tolevel(world=int(kw["world"]) if "world" in kw else None)))
|
|
elif cmd == "keys":
|
|
print(s.keys(pos, outdir=kw.get("out")))
|
|
elif cmd == "shot":
|
|
print(s.shot(pos[0], small="--small" in pos))
|
|
elif cmd == "ocr":
|
|
print(s.ocr(pos[0] if pos else FULL))
|
|
elif cmd == "status":
|
|
st = s.status()
|
|
print(json.dumps(st, indent=1))
|
|
return 3 if st["crashed"] else 0
|
|
elif cmd == "stop":
|
|
s.stop()
|
|
elif cmd == "run":
|
|
if not s.display or "out" not in kw:
|
|
print("run needs --display N and --out DIR")
|
|
return 2
|
|
mods = [x for x in kw.get("mods", "").split(",") if x]
|
|
return run(s, mods, opts, kw.get("level"), kw.get("seq"), kw["out"], keep="--keep" in pos)
|
|
else:
|
|
print(__doc__)
|
|
return 2
|
|
except Crash as e:
|
|
log("CRASH: %s" % e)
|
|
return 3
|
|
except Timeout as e:
|
|
log("TIMEOUT: %s" % e)
|
|
return 4
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main(sys.argv[1:]))
|