945 lines
34 KiB
Python
Executable File
945 lines
34 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""lamatest - unbeaufsichtigter Test-Harness fuer "Ein Koenigreich fuer ein Lama".
|
|
|
|
Startet groove.exe (Wine) unsichtbar auf Xvfb :5 mit Null-Audio-Sink, laedt den
|
|
TestLevel (Welt 9, Kapitel 1) ueber Spielstand-Slot 3 und macht Screenshots.
|
|
|
|
lamatest.py start Xvfb + Sink + Spiel, wartet aufs Fenster
|
|
lamatest.py shot out.png Screenshot (1920x1080) von :5
|
|
lamatest.py key <key> [hold] Taste halten (xdotool-Name, Default 0.12 s)
|
|
lamatest.py keys <k1> <k2> ... mehrere Tasten nacheinander (k[:hold])
|
|
lamatest.py seq <outdir> "TOKENS" Tastenfolge wie run --seq im laufenden Spiel (shot:/fp:/rj: ...)
|
|
lamatest.py tolevel Demo -> Hauptmenue -> Slot 3 -> TestLevel 1
|
|
lamatest.py tour <outdir> [n] Rundum-Fotos: drehen/laufen/Kamera
|
|
lamatest.py status laeuft was? Fenster? Prozesse?
|
|
lamatest.py stop Spiel beenden, Sink entladen, Xvfb killen
|
|
lamatest.py run <wad> <outdir> [n] WAD als T1L9M001.wad installieren, starten,
|
|
tolevel, tour, Crash-Erkennung, Log sichern
|
|
lamatest.py restore-wad Original-T1L9M001.wad (.orig) zurueck
|
|
run-Optionen: --seq "Up:1.5 x shot:name sleep=1 ..." statt Tour eine Tastenfolge;
|
|
shot:NAME = Foto NAME.jpg, fp:NAME = Foto in Lama-Sicht (Return gehalten),
|
|
dialog = Dialog wegklicken, keys wie bei 'keys' (Taste[:Haltezeit]),
|
|
Up+x:0.8 = Tasten gleichzeitig, rj:T1:T2 = Anlauf-Sprung (Up halten,
|
|
nach T1 s X, dann noch T2 s Up),
|
|
run:T:x@1.3,shot@5,Left@3-3.4 = Up T s halten, X/Foto/Taste nach Zeitplan
|
|
(run/Left:T:... haelt eine andere Taste)
|
|
--tour zusaetzlich zur --seq auch die Standard-Tour
|
|
--keep getestete WAD installiert lassen (sonst wird die vorher
|
|
installierte T1L9M001.wad wiederhergestellt - nicht .orig!)
|
|
|
|
Zustand/Logs: $LAMATEST_DIR (Default /tmp/lamatest-$USER).
|
|
Doku: ../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("FEHLER:", 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
|
|
|
|
|
|
# ---------------------------------------------------------------- Prozesse
|
|
def groove_procs():
|
|
"""Alle laufenden groove.exe-Prozesse: [(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):
|
|
# Sicherheitsnetz: xdotool immer nur gegen :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-Fehlerdialog / Absturzfenster auf :5? Liefert Fenstertitel oder None."""
|
|
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
|
|
|
|
|
|
# ---------------------------------------------------------------- Spielstaende
|
|
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("Spielstand %s wurde veraendert -> wiederhergestellt" % f)
|
|
shutil.copy2(src, dst)
|
|
for f in ksv_files():
|
|
if not os.path.exists(os.path.join(SAVEBAK, f)):
|
|
log("neuer Spielstand %s vom Test -> entfernt" % 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) fehlt" % p)
|
|
b = open(p, "rb").read()
|
|
# Kartencursor (+4/+5) bestimmt, was auf der Auswahlkarte vorgewaehlt ist
|
|
if len(b) != 256 or b[4] != 9 or b[5] != 1:
|
|
log("WARNUNG: Slot 3: Kartencursor nicht auf Welt 9/Kapitel 1 (Cursor %d/%d, "
|
|
"aktuell %d/%d)" % (b[4], b[5], b[0x3f], b[0x40]))
|
|
|
|
|
|
# ---------------------------------------------------------------- Kommandos
|
|
def cmd_start(timeout=60):
|
|
os.makedirs(STATE, exist_ok=True)
|
|
f = foreign_groove()
|
|
if f:
|
|
die("Elias' eigene Spielinstanz laeuft (pid %s) - Test abgebrochen" % f)
|
|
if game_alive():
|
|
log("Spiel laeuft schon auf %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 ist nicht aktiv (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("fremder X-Server auf :5 laeuft bereits - wird mitbenutzt")
|
|
else:
|
|
die("/tmp/.X5-lock existiert, aber :5 antwortet nicht")
|
|
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 startet nicht (%s)" % XVFBLOG)
|
|
# Null-Sink (nie die App stummschalten!)
|
|
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("WARNUNG: Null-Sink nicht ladbar:", r.stderr.strip())
|
|
# Spiel
|
|
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("Fenster da nach %.1f s" % (time.time() - t0))
|
|
return
|
|
if p.poll() is not None and not game_alive():
|
|
die("wine beendet sich sofort (Log: %s)" % WINELOG)
|
|
time.sleep(0.5)
|
|
die("kein Groove-Fenster nach %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):
|
|
"""Kleiner Screenshot (960x540 JPEG) fuer Serien - Platte ist knapp.
|
|
Mit LAMATEST_SHOT=1280 (Umgebung) 1280x720 / Qualitaet 85 (Vorschaubilder)."""
|
|
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("kein Groove-Fenster")
|
|
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)
|
|
|
|
|
|
# Bildvergleich ueber kleine Graustufen-Thumbnails (nur ImageMagick noetig)
|
|
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):
|
|
"""Warten bis sich das Bild `still` Sekunden kaum aendert."""
|
|
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
|
|
|
|
|
|
|
|
# ---------------------------------------------------------------- Bilderkennung
|
|
def rgb_thumb():
|
|
"""48x27 RGB-Thumbnail des aktuellen Bildes (Liste von (r,g,b)-Zeilen)."""
|
|
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):
|
|
"""Grobe Bildklasse: black / letterbox / map / bright (Menue oder Spiel)."""
|
|
t = t or rgb_thumb()
|
|
if t is None:
|
|
return "none"
|
|
allm = _rows_mean(t, range(27))
|
|
if allm < 6:
|
|
return "black"
|
|
# nur oberste/unterste Thumbnail-Zeile pruefen: Untertitel liegen im unteren Balken
|
|
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" # Kamera-Flug / Dialog-Zwischensequenz
|
|
# Auswahlkarte Welt 9: fast einfarbig 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():
|
|
"""Goldene Energieleiste/Sonnenkrone oben links (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 auf dem Bildschirm (tesseract, deu), klein geschrieben.
|
|
threshold (Prozent): nur sehr helle Schrift behalten (z. B. markierter Menuepunkt)."""
|
|
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 oder Grund-String, wenn das Spiel abgestuerzt ist."""
|
|
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" # Textfeld "Gespeichertes Spiel N / Level Kapitel"
|
|
|
|
|
|
def slot_text():
|
|
# zwei Schwellen kombinieren - der Hintergrund flimmert, OCR schwankt
|
|
return ocr(SLOT_CROP, threshold=70, psm=6) + " | " + ocr(SLOT_CROP, threshold=82, psm=6)
|
|
|
|
|
|
def fuzzy_in(word, text, ratio=0.75):
|
|
"""word (evtl. mehrere Woerter) unscharf in text enthalten?"""
|
|
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):
|
|
"""Hauptmenue -> Spiel starten -> Slot mit TestLevel -> Kapitel 1 -> Gameplay."""
|
|
T = log_times if log_times is not None else {}
|
|
t0 = time.time()
|
|
# 1) Hauptmenue (Logo ~2 s, Menue nach ~5-6 s). Kommt stattdessen die Demo,
|
|
# bringt Return zurueck ins Menue.
|
|
tx = time.time()
|
|
while True:
|
|
c = crashed()
|
|
if c:
|
|
raise Crash(c)
|
|
# Hauptmenue: Menuepunkte rechts unten; bei 75 % Schwelle bleibt nur der
|
|
# markierte (helle) Punkt lesbar -> prueft zugleich die Cursorposition.
|
|
txt = ocr("700x300+1150+650", threshold=75, psm=6, scale="100%")
|
|
if fuzzy_in("spiel starten", txt):
|
|
log("Hauptmenue erkannt")
|
|
time.sleep(0.5)
|
|
key("Return")
|
|
break
|
|
if fuzzy_in("gespeichertes", slot_text()):
|
|
break # schon im Slot-Bildschirm
|
|
if time.time() - tx > 40:
|
|
log("kein Menue erkannt -> Return (Demo/Logo ueberspringen)")
|
|
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) Spielstandauswahl: ganz nach links (kein Wrap), dann den Slot mit TestLevel suchen
|
|
wait_for(_has("gespeichertes"), 20, "Spielstandauswahl")
|
|
time.sleep(0.4)
|
|
# Dreier-Slider ohne Wrap: Slot 3 ist ganz rechts, weiteres Right bleibt dort.
|
|
# Also einfach 2x Right (egal wo man startet) und einmal per OCR pruefen.
|
|
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) # 0.3 s halten: bei niedriger Bildrate gehen kurze Druecke verloren
|
|
time.sleep(0.3)
|
|
for i in range(3): # unter Last kann ein Druck verloren gehen -> noch einmal Right
|
|
time.sleep(0.6)
|
|
if _is_test():
|
|
log("Slot 3: TestLevel gefunden (Versuch %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) Auswahlkarte
|
|
wait_for(lambda: classify() == "map", 30, "Auswahlkarte")
|
|
time.sleep(1.0)
|
|
# Vorauswahl auf der Karte = Kartencursor im KSV (+4/+5), geprueft in check_slot3()
|
|
key("Return")
|
|
t_load = time.time()
|
|
T["map"] = round(t_load - t0, 1)
|
|
# 4) Laden -> (Kamera-Flug im 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("Zustand: %s (%.1f s nach Level-Start)" % (st, time.time() - t_load))
|
|
if st == "letterbox" and skip_flyby and not skipped:
|
|
time.sleep(0.5)
|
|
key("Return") # Kamera-Flug ueberspringen
|
|
skipped = True
|
|
T["flyby"] = round(time.time() - t_load, 1)
|
|
elif st == "letterbox" and skipped and time.time() - t_load > 15:
|
|
key("x") # Dialog weiterklicken
|
|
# Gameplay = helles Bild, kein Letterbox, und HUD sichtbar (oder nach
|
|
# Kamera-Flug / spaetestens 25 s - falls das HUD schon ausgeblendet ist)
|
|
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 nach %.1f s (Laden %.1f s), Zustaende %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):
|
|
"""Dialog-/Zwischensequenz (Letterbox) wegklicken."""
|
|
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"):
|
|
"""Fotoserie: n Bilder Kamera-Rundumblick (A), dann n Bilder Laufen+Drehen.
|
|
Liefert Liste der Dateien. Bilder als 960x540-JPEG (Platz sparen)."""
|
|
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")
|
|
# Kamera einmal rundum: 0.5 s A-Taste ~ 90 Grad
|
|
hold = 2.0 / max(n, 1)
|
|
for i in range(n):
|
|
key("a", hold)
|
|
time.sleep(0.7)
|
|
snap("cam%d" % i)
|
|
# Laufen: vorwaerts, zwischendurch drehen; Dialoge wegklicken
|
|
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)
|
|
# Lama-Sicht (Return gehalten) einmal pro Richtung
|
|
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("ABSTURZ: %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 gesichert: %s" % orig)
|
|
if os.path.abspath(MODWAD) != wad:
|
|
shutil.copyfile(wad, MODWAD)
|
|
# Overlay zeigt per Symlink auf MODWAD -> neu bauen nur wenn Link fehlt/falsch
|
|
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 fehlt -> 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("WARNUNG: weitere Schreibweise im Overlay: %s" % others)
|
|
log("WAD installiert: %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("kein %s" % orig)
|
|
shutil.copy2(orig, MODWAD)
|
|
log("Original-T1L9M001.wad wiederhergestellt")
|
|
|
|
|
|
def run_seq(outdir, seq):
|
|
"""Tastenfolge abarbeiten; 'shot:NAME' fotografiert. Liefert Bilddateien."""
|
|
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:"):
|
|
# Lama-Sicht: Return halten, waehrenddessen fotografieren
|
|
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:"):
|
|
# Anlauf-Sprung: Up halten, nach T1 s X tippen, Up noch T2 s halten
|
|
_, 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/"):
|
|
# Durchlaufen mit Zeitplan: run:T:x@2.1,x@3.4,shot@4.0,Left@5.0-5.3
|
|
# Up (bzw. bei run/Left:... die angegebene Taste) wird T s gehalten; zu den Zeiten (s ab Beginn) X tippen / Foto (seq_NN_runK) /
|
|
# Taste von-bis halten. Fotos dauern ~0.2 s, der Zeitplan laeuft weiter (Echtzeit).
|
|
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:"):
|
|
# Up halten und N-mal alle T s springen (Fotos dazwischen: shot nach dem 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:"):
|
|
# Doppelsprung mit Anlauf: Up halten, X, nach 0.3 s nochmal X, Up noch 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]:
|
|
# mehrere Tasten gleichzeitig: 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' eigene Spielinstanz laeuft - abgebrochen")
|
|
# vorher installierte WAD merken -> danach zurueck (nicht .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))
|
|
# eingefroren? (Kamera dreht sich -> aufeinanderfolgende Bilder muessen sich aendern)
|
|
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):
|
|
# Log kuerzen: max. 2 MB behalten (Anfang + Ende)
|
|
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)
|
|
# bekannte harmlose Meldungen (laufen auch bei funktionierendem Spiel)
|
|
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("fremde Instanz laeuft (%s) - kein wineserver -k, nur eigene Prozesse" % 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
|
|
# Sicherheitsnetz: Sink per Name entladen, falls Modulnummer verloren
|
|
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("gestoppt")
|
|
|
|
|
|
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": # Tastenfolge wie run --seq, aber im laufenden Spiel: 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: Helligkeit/Diff
|
|
print(mean(thumb(a[0] if a else None)))
|
|
else:
|
|
print(__doc__)
|
|
sys.exit(2)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main(sys.argv)
|