MineTec/llamamod
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 4 Wiki Activity
Files
526f6072f6549b1b8e32cf1ebefd82cba35055f3
llamamod/tools/lamatest.py
T

948 lines
34 KiB
Python
Executable File
Raw Blame History

#!/usr/bin/env python3
"""lamatest - unattended test harness for "Ein Koenigreich fuer ein Lama".
Starts groove.exe (Wine) invisibly on Xvfb :5 with a null audio sink, loads the
TestLevel (world 9, chapter 1) via save slot 3 and takes screenshots.
lamatest.py start Xvfb + sink + game, waits for the window
lamatest.py shot out.png screenshot (1920x1080) of :5
lamatest.py key <key> [hold] hold a key (xdotool name, default 0.12 s)
lamatest.py keys <k1> <k2> ... several keys one after another (k[:hold])
lamatest.py seq <outdir> "TOKENS" key sequence like run --seq in the running game (shot:/fp:/rj: ...)
lamatest.py tolevel demo -> main menu -> slot 3 -> TestLevel 1
lamatest.py tour <outdir> [n] all-round photos: turn/walk/camera
lamatest.py status anything running? window? processes?
lamatest.py stop quit the game, unload the sink, kill Xvfb
lamatest.py run <wad> <outdir> [n] install WAD as T1L9M001.wad, start,
tolevel, tour, crash detection, save the log
lamatest.py restore-wad restore the original T1L9M001.wad (.orig)
run options: --seq "Up:1.5 x shot:name sleep=1 ..." a key sequence instead of the tour;
shot:NAME = photo NAME.jpg, fp:NAME = photo in llama view (Return held),
dialog = click dialog away, keys as with 'keys' (key[:hold time]),
Up+x:0.8 = keys simultaneously, rj:T1:T2 = running jump (hold Up,
X after T1 s, then Up for another T2 s),
run:T:x@1.3,shot@5,Left@3-3.4 = hold Up for T s, X/photo/key on a schedule
(run/Left:T:... holds a different key)
--tour run the standard tour in addition to --seq
--keep leave the tested WAD installed (otherwise the previously
installed T1L9M001.wad is restored - not .orig!)
State/logs: $LAMATEST_DIR (default /tmp/lamatest-$USER).
Report lines (report.txt, "ergebnis: ...") and `status` output stay German: helper
scripts grep for them. OCR expectations are German game text.
Docs: ../notes/testharness.md
"""
import hashlib
import os
import shutil
import signal
import subprocess
import sys
import time
HOME = os.path.expanduser("~")
DISPLAY = ":5"
SINK = "lama_stumm"
PREFIX = os.path.join(HOME, ".wine-lama")
GAMEDIR = os.path.join(PREFIX, "drive_c", "Program Files", "Disney Interactive",
"Ein Koenigreich fuer ein Lama")
GAMEEXE = r"C:\Program Files\Disney Interactive\Ein Koenigreich fuer ein Lama\groove.exe"
MODWAD = os.path.join(PREFIX, "mods", "testlevel", "wads", "T1L9M001.wad")
OVERLAY = os.path.join(PREFIX, "cd-overlay")
LAMA_MOD = os.path.join(HOME, ".local", "bin", "lama-mod")
STATE = os.environ.get("LAMATEST_DIR", "/tmp/lamatest-%s" % os.environ.get("USER", "x"))
WINELOG = os.path.join(STATE, "wine.log")
XVFBLOG = os.path.join(STATE, "xvfb.log")
SAVEBAK = os.path.join(STATE, "ksv-backup")
ENV = dict(os.environ, DISPLAY=DISPLAY, PULSE_SINK=SINK, WINEPREFIX=PREFIX,
WINEDEBUG=os.environ.get("WINEDEBUG", "fixme-all"))
def log(*a):
print("[lamatest %s]" % time.strftime("%H:%M:%S"), *a, file=sys.stderr, flush=True)
def die(msg, code=1):
log("ERROR:", msg)
sys.exit(code)
def sh(cmd, check=False, **kw):
kw.setdefault("env", ENV)
return subprocess.run(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
text=True, check=check, **kw)
def pidfile(name):
return os.path.join(STATE, name + ".pid")
def read_pid(name):
try:
pid = int(open(pidfile(name)).read().strip())
os.kill(pid, 0)
return pid
except (OSError, ValueError):
return None
# ---------------------------------------------------------------- processes
def groove_procs():
"""All running groove.exe processes: [(pid, environ DISPLAY)]."""
out = []
for p in os.listdir("/proc"):
if not p.isdigit():
continue
try:
cmd = open("/proc/%s/cmdline" % p, "rb").read().replace(b"\0", b" ").lower()
except OSError:
continue
if b"groove.exe" not in cmd or b"lamatest" in cmd:
continue
disp = None
try:
for kv in open("/proc/%s/environ" % p, "rb").read().split(b"\0"):
if kv.startswith(b"DISPLAY="):
disp = kv[8:].decode()
except OSError:
pass
out.append((int(p), disp))
return out
def foreign_groove():
return [(p, d) for p, d in groove_procs() if d != DISPLAY]
def xdo(*args):
# safety net: xdotool only ever against :5
assert ENV["DISPLAY"] == DISPLAY
return sh(["xdotool"] + list(args))
def find_window():
r = xdo("search", "--name", "^Groove$")
ids = r.stdout.split()
return ids[-1] if ids else None
def game_alive():
return any(d == DISPLAY for _, d in groove_procs())
def error_dialog():
"""Wine error dialog / crash window on :5? Returns the window title or None
(matches English and German Wine dialog titles)."""
r = xdo("search", "--name", ".")
titles = []
for wid in r.stdout.split():
t = xdo("getwindowname", wid).stdout.strip()
if t:
titles.append(t)
for t in titles:
tl = t.lower()
if ("program error" in tl or "programmfehler" in tl or "wine" in tl and "error" in tl
or "fehler" in tl or "croc" in tl or "exception" in tl or "winedbg" in tl):
return t
return None
# ---------------------------------------------------------------- save games
def ksv_files():
return sorted(f for f in os.listdir(GAMEDIR) if f.lower().endswith(".ksv"))
def backup_saves():
os.makedirs(SAVEBAK, exist_ok=True)
for f in ksv_files():
shutil.copy2(os.path.join(GAMEDIR, f), os.path.join(SAVEBAK, f))
def restore_saves():
if not os.path.isdir(SAVEBAK):
return
for f in os.listdir(SAVEBAK):
src, dst = os.path.join(SAVEBAK, f), os.path.join(GAMEDIR, f)
if not os.path.exists(dst) or md5(src) != md5(dst):
log("save game %s was modified -> restored" % f)
shutil.copy2(src, dst)
for f in ksv_files():
if not os.path.exists(os.path.join(SAVEBAK, f)):
log("new save game %s from the test -> removed" % f)
os.remove(os.path.join(GAMEDIR, f))
def md5(p):
return hashlib.md5(open(p, "rb").read()).hexdigest()
def check_slot3():
p = os.path.join(GAMEDIR, "(null)king2.ksv")
if not os.path.exists(p):
die("slot 3 (%s) missing" % p)
b = open(p, "rb").read()
# map cursor (+4/+5) decides what is preselected on the selection map
if len(b) != 256 or b[4] != 9 or b[5] != 1:
log("WARNING: slot 3: map cursor not on world 9/chapter 1 (cursor %d/%d, "
"current %d/%d)" % (b[4], b[5], b[0x3f], b[0x40]))
# ---------------------------------------------------------------- commands
def cmd_start(timeout=60):
os.makedirs(STATE, exist_ok=True)
f = foreign_groove()
if f:
die("Elias' own game instance is running (pid %s) - test aborted" % f)
if game_alive():
log("game already running on %s" % DISPLAY)
return
if not os.path.isdir(OVERLAY) or not os.path.realpath(
os.path.join(PREFIX, "dosdevices", "l:")) == os.path.realpath(OVERLAY):
die("mod is not active (lama-mod an)")
check_slot3()
backup_saves()
# Xvfb
if not read_pid("xvfb"):
if os.path.exists("/tmp/.X11-unix/X5") or os.path.exists("/tmp/.X5-lock"):
if sh(["xdpyinfo"]).returncode == 0:
log("foreign X server already running on :5 - sharing it")
else:
die("/tmp/.X5-lock exists, but :5 does not respond")
else:
xl = open(XVFBLOG, "w")
p = subprocess.Popen(["Xvfb", DISPLAY, "-screen", "0", "1920x1200x24",
"-nolisten", "tcp", "+extension", "GLX"],
stdout=xl, stderr=xl, start_new_session=True)
open(pidfile("xvfb"), "w").write(str(p.pid))
for _ in range(50):
if sh(["xdpyinfo"]).returncode == 0:
break
time.sleep(0.1)
else:
die("Xvfb does not start (%s)" % XVFBLOG)
# null sink (never mute the app!)
if not os.path.exists(pidfile("sink")):
r = sh(["pactl", "load-module", "module-null-sink", "sink_name=" + SINK,
"sink_properties=device.description=lama_stumm"])
if r.returncode == 0:
open(pidfile("sink"), "w").write(r.stdout.strip())
else:
log("WARNING: cannot load null sink:", r.stderr.strip())
# game
wl = open(WINELOG, "w")
p = subprocess.Popen(["wine", GAMEEXE], cwd=GAMEDIR, env=ENV, stdout=wl,
stderr=subprocess.STDOUT, start_new_session=True)
open(pidfile("wine"), "w").write(str(p.pid))
t0 = time.time()
while time.time() - t0 < timeout:
if find_window():
log("window up after %.1f s" % (time.time() - t0))
return
if p.poll() is not None and not game_alive():
die("wine exits immediately (log: %s)" % WINELOG)
time.sleep(0.5)
die("no Groove window after %d s" % timeout)
def shot(path):
d = os.path.dirname(os.path.abspath(path))
os.makedirs(d, exist_ok=True)
r = sh(["import", "-display", DISPLAY, "-window", "root", "-crop",
"1920x1080+0+0", "+repage", path])
if r.returncode:
die("import: " + r.stderr)
return path
def small_shot(path):
"""Small screenshot (960x540 JPEG) for series - disk space is tight.
With LAMATEST_SHOT=1280 (environment) 1280x720 / quality 85 (preview images)."""
w = os.environ.get("LAMATEST_SHOT")
scale, q = ("%sx" % w, "85") if w else ("50%", "80")
r = sh(["import", "-display", DISPLAY, "-window", "root", "-crop",
"1920x1080+0+0", "+repage", "-resize", scale, "-quality", q, path])
if r.returncode:
die("import: " + r.stderr)
return path
def key(k, hold=0.12):
wid = find_window()
if not wid:
die("no Groove window")
xdo("windowfocus", "--sync", wid)
xdo("keydown", k)
time.sleep(float(hold))
xdo("keyup", k)
def keys(seq, gap=0.35):
for item in seq:
if item.startswith("sleep="):
time.sleep(float(item[6:]))
continue
k, _, h = item.partition(":")
key(k, float(h) if h else 0.12)
time.sleep(gap)
# image comparison via small greyscale thumbnails (only ImageMagick needed)
def thumb(path=None):
src = ["import", "-display", DISPLAY, "-window", "root", "-crop", "1920x1080+0+0",
"+repage"] if path is None else ["convert", path]
r = subprocess.run(src + ["-resize", "48x27!", "-colorspace", "Gray", "-depth", "8",
"gray:-"], stdout=subprocess.PIPE, env=ENV)
return r.stdout
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 mean(t):
return sum(t) / len(t) if t else 0
def wait_stable(min_time=0.0, timeout=30, still=1.5, thresh=1.0, dark_ok=False):
"""Wait until the image barely changes for `still` seconds."""
t0 = time.time()
time.sleep(min_time)
last, since = thumb(), time.time()
while time.time() - t0 < timeout:
time.sleep(0.4)
cur = thumb()
if diff(cur, last) > thresh or (not dark_ok and mean(cur) < 3):
since = time.time()
elif time.time() - since >= still:
return cur
last = cur
return last
# ---------------------------------------------------------------- image recognition
def rgb_thumb():
"""48x27 RGB thumbnail of the current image (list of rows of (r,g,b))."""
r = subprocess.run(["import", "-display", DISPLAY, "-window", "root", "-crop",
"1920x1080+0+0", "+repage", "-resize", "48x27!", "-depth", "8",
"rgb:-"], stdout=subprocess.PIPE, env=ENV)
b = r.stdout
if len(b) != 48 * 27 * 3:
return None
return [[tuple(b[(y * 48 + x) * 3:(y * 48 + x) * 3 + 3]) for x in range(48)]
for y in range(27)]
def _rows_mean(t, ys):
px = [sum(c) / 3 for y in ys for c in t[y]]
return sum(px) / len(px)
def classify(t=None):
"""Rough image class: black / letterbox / map / bright (menu or game)."""
t = t or rgb_thumb()
if t is None:
return "none"
allm = _rows_mean(t, range(27))
if allm < 6:
return "black"
# only check the top/bottom thumbnail row: subtitles sit in the lower bar
top, bot, mid = _rows_mean(t, (0,)), _rows_mean(t, (26,)), _rows_mean(t, range(6, 18))
if top < 4 and bot < 4 and mid > 15:
return "letterbox" # camera fly-by / dialogue cutscene
# selection map world 9: almost uniformly beige (~205,175,110)
beige = sum(1 for row in t for (r, g, b) in row
if 180 < r < 230 and 150 < g < 200 and 85 < b < 140 and r > g > b)
if beige > 0.75 * 48 * 27:
return "map"
return "bright"
def hud_visible():
"""Golden energy bar/sun crown at the top left (gameplay HUD)?"""
r = subprocess.run(["import", "-display", DISPLAY, "-window", "root", "-crop",
"900x110+100+10", "+repage", "-resize", "25%", "-depth", "8",
"rgb:-"], stdout=subprocess.PIPE, env=ENV)
b = r.stdout
gold = 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)
return gold > 150
def ocr(crop="1920x1080+0+0", threshold=None, psm=11, scale="50%"):
"""Text on screen (tesseract, deu), lower-cased.
threshold (percent): keep only very bright text (e.g. the highlighted menu item)."""
filt = ["-threshold", "%d%%" % threshold] if threshold else []
raw = subprocess.run(["import", "-display", DISPLAY, "-window", "root", "-crop",
crop, "+repage", "ppm:-"], stdout=subprocess.PIPE, env=ENV).stdout
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 crashed():
"""None, or a reason string if the game has crashed (goes into report.txt)."""
if not game_alive():
return "groove.exe-Prozess weg"
e = error_dialog()
if e:
return "Fehlerdialog: %s" % e
if not find_window():
return "Groove-Fenster weg"
return None
def wait_for(pred, timeout, what, poll=0.7):
t0 = time.time()
while time.time() - t0 < timeout:
c = crashed()
if c:
raise Crash(c)
v = pred()
if v:
return v
time.sleep(poll)
raise Timeout("Zeitueberschreitung (%ds) beim Warten auf %s" % (timeout, what))
class Crash(Exception):
pass
class Timeout(Exception):
pass
SLOT_CROP = "1600x280+160+740" # text field "Gespeichertes Spiel N / Level Kapitel" (German game text)
def slot_text():
# combine two thresholds - the background flickers, OCR fluctuates
return ocr(SLOT_CROP, threshold=70, psm=6) + " | " + ocr(SLOT_CROP, threshold=82, psm=6)
def fuzzy_in(word, text, ratio=0.75):
"""Is word (possibly several words) fuzzily contained in text?"""
import difflib
if word in text:
return True
toks = text.split()
n = len(word.split())
for i in range(len(toks) - n + 1):
cand = " ".join(toks[i:i + n])
if difflib.SequenceMatcher(None, word, cand).ratio() >= ratio:
return True
return False
def _has(*words):
def f():
t = slot_text()
return t if all(fuzzy_in(w, t) for w in words) else None
return f
def tolevel(skip_flyby=True, log_times=None):
"""Main menu -> "Spiel starten" -> slot with TestLevel -> chapter 1 -> gameplay."""
T = log_times if log_times is not None else {}
t0 = time.time()
# 1) main menu (logo ~2 s, menu after ~5-6 s). If the demo comes instead,
# Return brings you back to the menu.
tx = time.time()
while True:
c = crashed()
if c:
raise Crash(c)
# main menu: menu items at the bottom right; at a 75 % threshold only the
# highlighted (bright) item stays readable -> also checks the cursor position.
txt = ocr("700x300+1150+650", threshold=75, psm=6, scale="100%")
if fuzzy_in("spiel starten", txt):
log("main menu detected")
time.sleep(0.5)
key("Return")
break
if fuzzy_in("gespeichertes", slot_text()):
break # already on the slot screen
if time.time() - tx > 40:
log("no menu detected -> Return (skip demo/logo)")
key("Return")
tx = time.time()
if time.time() - t0 > 90:
raise Timeout("Hauptmenue nicht erreicht")
time.sleep(1)
T["menu"] = round(time.time() - t0, 1)
# 2) save game selection: all the way left (no wrap), then find the slot with TestLevel
wait_for(_has("gespeichertes"), 20, "Spielstandauswahl")
time.sleep(0.4)
# three-slot slider without wrap: slot 3 is at the far right, further Right stays there.
# So just press Right twice (wherever you start) and check once via OCR.
def _is_test():
return (fuzzy_in("testlevel", ocr(SLOT_CROP, threshold=70, psm=6))
or fuzzy_in("testlevel", ocr(SLOT_CROP, threshold=82, psm=6)))
for _ in range(2):
key("Right", 0.3) # hold 0.3 s: short presses get lost at low frame rates
time.sleep(0.3)
for i in range(3): # under load a press can get lost -> Right once more
time.sleep(0.6)
if _is_test():
log("slot 3: TestLevel found (attempt %d)" % (i + 1))
break
key("Right", 0.3)
else:
raise Timeout("kein Spielstand mit TestLevel gefunden (Slot 3 kaputt?)")
key("Return")
T["slot"] = round(time.time() - t0, 1)
# 3) selection map
wait_for(lambda: classify() == "map", 30, "Auswahlkarte")
time.sleep(1.0)
# preselection on the map = map cursor in the KSV (+4/+5), checked in check_slot3()
key("Return")
t_load = time.time()
T["map"] = round(t_load - t0, 1)
# 4) loading -> (camera fly-by in letterbox) -> gameplay
skipped = False
good = 0
states = []
while time.time() - t_load < 120:
c = crashed()
if c:
raise Crash(c)
st = 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)
key("Return") # skip the camera fly-by
skipped = True
T["flyby"] = round(time.time() - t_load, 1)
elif st == "letterbox" and skipped and time.time() - t_load > 15:
key("x") # click through the dialogue
# gameplay = bright image, no letterbox, and HUD visible (or after the
# camera fly-by / after 25 s at the latest - in case the HUD has already faded out)
el = time.time() - t_load
if st == "bright" and el > 2 and (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 nicht erreicht (Zustaende: %s)" % states)
T["gameplay"] = round(time.time() - t0, 1)
T["load"] = round(time.time() - t_load, 1)
T["states"] = states
log("gameplay after %.1f s (loading %.1f s), states %s" % (T["gameplay"], T["load"], states))
return T
def cmd_tolevel():
try:
T = tolevel()
except (Crash, Timeout) as e:
die(str(e))
print(T)
def settle_dialog(maxpress=12):
"""Click away a dialogue/cutscene (letterbox)."""
n = 0
while classify() == "letterbox" and n < maxpress:
key("x")
time.sleep(1.0)
n += 1
return n
def tour(outdir, n=8, prefix="tour"):
"""Photo series: n images of a camera pan (A), then n images walking+turning.
Returns the list of files. Images as 960x540 JPEG (saves space)."""
os.makedirs(outdir, exist_ok=True)
files = []
def snap(tag):
c = crashed()
if c:
raise Crash(c)
f = os.path.join(outdir, "%s_%02d_%s.jpg" % (prefix, len(files), tag))
small_shot(f)
files.append(f)
snap("start")
# camera once all round: 0.5 s of the A key ~ 90 degrees
hold = 2.0 / max(n, 1)
for i in range(n):
key("a", hold)
time.sleep(0.7)
snap("cam%d" % i)
# walking: forwards, turning in between; click dialogues away
for i in range(n):
key("Up", 1.2)
time.sleep(0.4)
if classify() == "letterbox":
snap("dialog%d" % i)
settle_dialog()
snap("walk%d" % i)
key("Left" if i % 3 != 2 else "Right", 0.45)
time.sleep(0.3)
# llama view (Return held) once per direction
key("Return", 1.5)
time.sleep(0.2)
snap("lamasicht")
return files
def cmd_tour(outdir, n):
try:
fs = tour(outdir, n)
except Crash as e:
die("CRASH: %s" % e, 3)
print("\n".join(fs))
# ---------------------------------------------------------------- WAD + run
def install_wad(wad):
wad = os.path.abspath(wad)
orig = MODWAD + ".orig"
if not os.path.exists(orig):
shutil.copy2(MODWAD, orig)
log("original backed up: %s" % orig)
if os.path.abspath(MODWAD) != wad:
shutil.copyfile(wad, MODWAD)
# overlay points to MODWAD via symlink -> rebuild only if the link is missing/wrong
link = os.path.join(OVERLAY, "wads", "T1L9M001.wad")
if not (os.path.islink(link) and os.path.realpath(link) == os.path.realpath(MODWAD)):
log("overlay link missing -> lama-mod an")
sh([LAMA_MOD, "an"], check=True)
others = [f for f in os.listdir(os.path.dirname(link))
if f.lower() == "t1l9m001.wad" and f != "T1L9M001.wad"]
if others:
log("WARNING: other spelling in the overlay: %s" % others)
log("WAD installed: %s (%d bytes, md5 %s)" % (wad, os.path.getsize(MODWAD), md5(MODWAD)[:12]))
def cmd_restore_wad():
orig = MODWAD + ".orig"
if not os.path.exists(orig):
die("no %s" % orig)
shutil.copy2(orig, MODWAD)
log("original T1L9M001.wad restored")
def run_seq(outdir, seq):
"""Process a key sequence; 'shot:NAME' takes a photo. Returns the image files."""
files = []
for item in seq.split():
c = crashed()
if c:
raise Crash(c)
if item.startswith("shot:"):
f = os.path.join(outdir, "seq_%02d_%s.jpg" % (len(files), item[5:]))
small_shot(f)
files.append(f)
elif item.startswith("fp:"):
# llama view: hold Return, take the photo meanwhile
wid = find_window()
xdo("windowfocus", "--sync", wid)
xdo("keydown", "Return")
time.sleep(1.2)
f = os.path.join(outdir, "seq_%02d_%s.jpg" % (len(files), item[3:]))
small_shot(f)
files.append(f)
xdo("keyup", "Return")
time.sleep(0.4)
elif item.startswith("rj:"):
# running jump: hold Up, tap X after T1 s, keep holding Up for T2 s
_, t1, t2 = item.split(":")
wid = find_window()
xdo("windowfocus", "--sync", wid)
xdo("keydown", "Up")
time.sleep(float(t1))
xdo("keydown", "x")
time.sleep(0.15)
xdo("keyup", "x")
time.sleep(float(t2))
xdo("keyup", "Up")
time.sleep(0.3)
elif item.startswith("run:") or item.startswith("run/"):
# run through with a schedule: run:T:x@2.1,x@3.4,shot@4.0,Left@5.0-5.3
# Up (or the given key for run/Left:...) is held for T s; at the given times (s from start) tap X / photo (seq_NN_runK) /
# hold a key from-to. Photos take ~0.2 s, the schedule keeps running (real time).
parts = item.split(":", 2)
mk = parts[0][4:] or "Up"
total, ev = float(parts[1]), []
for e in (parts[2].split(",") if len(parts) > 2 and parts[2] else []):
k, _, t = e.partition("@")
if "-" in t:
a, b = t.split("-")
ev.append((float(a), "down", k)); ev.append((float(b), "up", k))
else:
ev.append((float(t), k, None))
ev.sort(key=lambda e: e[0])
wid = find_window()
xdo("windowfocus", "--sync", wid)
t0 = time.time()
xdo("keydown", mk)
for t, k, arg in ev + [(total, "end", None)]:
dt = t0 + t - time.time()
if dt > 0:
time.sleep(dt)
if k == "end":
break
if k == "down":
xdo("keydown", arg)
elif k == "up":
xdo("keyup", arg)
elif k == "shot":
f = os.path.join(outdir, "seq_%02d_run%.1f.jpg" % (len(files), t))
small_shot(f)
files.append(f)
else:
xdo("keydown", k); time.sleep(0.1); xdo("keyup", k)
xdo("keyup", mk)
for t, k, arg in ev:
if k == "down":
xdo("keyup", arg)
time.sleep(0.3)
elif item.startswith("hop:"):
# hold Up and jump N times every T s (photos in between: shot after the token)
_, n, t = item.split(":")
wid = find_window()
xdo("windowfocus", "--sync", wid)
xdo("keydown", "Up")
for _ in range(int(n)):
xdo("keydown", "x"); time.sleep(0.12); xdo("keyup", "x"); time.sleep(float(t) - 0.12)
xdo("keyup", "Up")
time.sleep(0.3)
elif item.startswith("dj:"):
# double jump with run-up: hold Up, X, X again after 0.3 s, Up for another T s
t = float(item.split(":")[1])
wid = find_window()
xdo("windowfocus", "--sync", wid)
xdo("keydown", "Up")
for _ in range(2):
xdo("keydown", "x"); time.sleep(0.12); xdo("keyup", "x"); time.sleep(0.25)
time.sleep(t)
xdo("keyup", "Up")
time.sleep(0.3)
elif "+" in item.split(":")[0]:
# several keys simultaneously: Up+x:0.8
ks, _, h = item.partition(":")
wid = find_window()
xdo("windowfocus", "--sync", wid)
for k in ks.split("+"):
xdo("keydown", k)
time.sleep(float(h) if h else 0.12)
for k in reversed(ks.split("+")):
xdo("keyup", k)
time.sleep(0.3)
elif item == "dialog":
settle_dialog()
else:
keys([item])
return files
def cmd_run(wad, outdir, n=8, seq=None, do_tour=None, keep=False):
os.makedirs(outdir, exist_ok=True)
report = os.path.join(outdir, "report.txt")
rep = ["wad: %s" % os.path.abspath(wad)]
status = 0
if foreign_groove():
die("Elias' own game instance is running - aborted")
# remember the previously installed WAD -> put it back afterwards (not .orig!)
prev = os.path.join(STATE, "vorher_T1L9M001.wad")
os.makedirs(STATE, exist_ok=True)
if not keep and os.path.exists(MODWAD):
shutil.copyfile(MODWAD, prev)
install_wad(wad)
rep.append("md5: %s" % md5(MODWAD))
t0 = time.time()
try:
cmd_start()
rep.append("fenster: %.1f s" % (time.time() - t0))
T = tolevel()
rep.append("tolevel: %s" % T)
small_shot(os.path.join(outdir, "00_geladen.jpg"))
shot(os.path.join(outdir, "00_geladen.png"))
fs = []
if seq:
fs += run_seq(outdir, seq)
if do_tour or (do_tour is None and not seq):
fs += tour(outdir, n)
rep.append("bilder: %d" % (len(fs) + 2))
# frozen? (the camera turns -> consecutive images must differ)
th = [thumb(f) for f in fs]
dmax = max([diff(a, b) for a, b in zip(th, th[1:])] or [0])
rep.append("bildaenderung max: %.1f" % dmax)
if dmax < 1.0:
status = 5
rep.append("WARNUNG: Bild scheint eingefroren")
c = crashed()
if c:
raise Crash(c)
if status == 0:
rep.append("ergebnis: OK (laeuft nach %.0f s noch)" % (time.time() - t0))
else:
rep.append("ergebnis: EINGEFROREN?")
except Crash as e:
status = 3
rep.append("ergebnis: ABSTURZ - %s" % e)
try:
small_shot(os.path.join(outdir, "zz_absturz.jpg"))
except SystemExit:
pass
except Timeout as e:
status = 4
rep.append("ergebnis: HAENGT/TIMEOUT - %s" % e)
try:
small_shot(os.path.join(outdir, "zz_timeout.jpg"))
except SystemExit:
pass
finally:
try:
cmd_stop()
finally:
if os.path.exists(WINELOG):
# truncate the log: keep at most 2 MB (start + end)
data = open(WINELOG, "rb").read()
if len(data) > 2_000_000:
data = data[:1_000_000] + b"\n...[gekuerzt]...\n" + data[-1_000_000:]
open(os.path.join(outdir, "wine.log"), "wb").write(data)
# known harmless messages (also appear when the game works fine)
noise = ("com_get_class_object", "ChangeDisplaySettingsExW",
"wined3d_restore_display_modes", "gstreamer", "MIDIMAP",
"RestoreDisplayMode")
errs = [l for l in data.decode("latin-1").splitlines()
if (":err:" in l or "exception" in l.lower() or "Unhandled" in l
or "croc" in l.lower()) and not any(x in l for x in noise)]
rep.append("wine-log: %d err/exception-Zeilen" % len(errs))
rep += [" " + l for l in errs[:15]]
if not keep and os.path.exists(prev):
shutil.copyfile(prev, MODWAD)
os.remove(prev)
rep.append("WAD: vorher installierte Version wiederhergestellt (md5 %s)" % md5(MODWAD)[:12])
elif keep:
rep.append("WAD: getestete Version bleibt installiert")
rep.append("dauer: %.0f s" % (time.time() - t0))
open(report, "w").write("\n".join(rep) + "\n")
print("\n".join(rep))
return status
def cmd_stop():
f = foreign_groove()
mine = [p for p, d in groove_procs() if d == DISPLAY]
for p in mine:
try:
os.kill(p, signal.SIGTERM)
except OSError:
pass
time.sleep(1)
if f:
log("foreign instance running (%s) - no wineserver -k, only own processes" % f)
for p, d in groove_procs():
if d == DISPLAY:
os.kill(p, signal.SIGKILL)
else:
sh(["wineserver", "-k"])
sh(["wineserver", "-w"], timeout=20)
wp = read_pid("wine")
if wp:
try:
os.killpg(wp, signal.SIGKILL)
except OSError:
pass
xp = read_pid("xvfb")
if xp:
os.kill(xp, signal.SIGTERM)
for _ in range(30):
if not read_pid("xvfb"):
break
time.sleep(0.1)
try:
os.remove(pidfile("xvfb"))
except OSError:
pass
try:
mod = open(pidfile("sink")).read().strip()
if mod:
sh(["pactl", "unload-module", mod])
os.remove(pidfile("sink"))
except OSError:
pass
# safety net: unload the sink by name in case the module number got lost
r = sh(["pactl", "list", "short", "modules"])
for line in r.stdout.splitlines():
if "sink_name=" + SINK in line:
sh(["pactl", "unload-module", line.split()[0]])
restore_saves()
log("stopped")
def cmd_status():
print("eigene groove.exe (:5):", [p for p, d in groove_procs() if d == DISPLAY])
print("fremde groove.exe: ", foreign_groove())
print("Xvfb-pid:", read_pid("xvfb"), " Fenster:", find_window() if read_pid("xvfb") else None)
print("Fehlerdialog:", error_dialog() if read_pid("xvfb") else None)
print("Mod-WAD:", MODWAD, md5(MODWAD) if os.path.exists(MODWAD) else "-")
def main(argv):
if len(argv) < 2 or argv[1] in ("-h", "--help", "help"):
print(__doc__)
return
c, a = argv[1], argv[2:]
if c == "start":
cmd_start()
elif c == "stop":
cmd_stop()
elif c == "status":
cmd_status()
elif c == "shot":
print(shot(a[0] if a else "shot.png"))
elif c == "key":
key(a[0], float(a[1]) if len(a) > 1 else 0.12)
elif c == "keys":
keys(a)
elif c == "tolevel":
cmd_tolevel()
elif c == "seq": # key sequence like run --seq, but in the running game: seq <outdir> "TOKENS"
os.makedirs(a[0], exist_ok=True)
for f in run_seq(a[0], a[1]):
print(f)
elif c == "tour":
cmd_tour(a[0], int(a[1]) if len(a) > 1 else 8)
elif c == "run":
seq, do_tour, keep, pos = None, None, False, []
i = 0
while i < len(a):
if a[i] == "--seq":
seq = a[i + 1]; i += 2; continue
if a[i] == "--tour":
do_tour = True
elif a[i] == "--keep":
keep = True
else:
pos.append(a[i])
i += 1
sys.exit(cmd_run(pos[0], pos[1], int(pos[2]) if len(pos) > 2 else 8, seq, do_tour, keep))
elif c == "restore-wad":
cmd_restore_wad()
elif c == "mean": # debug: brightness/diff
print(mean(thumb(a[0] if a else None)))
else:
print(__doc__)
sys.exit(2)
if __name__ == "__main__":
main(sys.argv)
Reference in new issue View Git Blame Copy Permalink