262 lines
12 KiB
Python
262 lines
12 KiB
Python
"""Automated in-game test (Windows): does a replaced sound actually play?
|
|
|
|
Flow: apply the replacement through `Workspace.apply` (the same code path as the editor) ->
|
|
start the game under cdb with logging breakpoints on the decoder + record the speakers (WASAPI
|
|
loopback) -> check: no crash, the codec setup of the injected WEM was created and parsed OK, and the
|
|
injected audio is heard -> kill the game -> revert and check the files are byte-identical again.
|
|
|
|
Default target: SFX music EDF036D6/393BC53A (intro/title music), which the game starts ~5 s after
|
|
launch without input. That slot ignores sources shorter than ~10 s, so probes are 30 s long.
|
|
Only correlation scores of the recording are kept (the loopback may contain other apps' audio).
|
|
"""
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import re
|
|
import time
|
|
|
|
import numpy as np
|
|
|
|
from .. import audio
|
|
from ..originals import sha256
|
|
|
|
DEFAULT_TARGET = "SFX:EDF036D6/393BC53A"
|
|
RATE = 48000
|
|
|
|
|
|
def probe_signal(seconds=30.0, rate=RATE, seed=1234):
|
|
"""Pink-ish noise + vibrato tone: easy to recognise in a recording."""
|
|
rng = np.random.default_rng(seed)
|
|
n = int(seconds * rate)
|
|
white = rng.standard_normal(n)
|
|
pink = np.cumsum(white) * 0.002
|
|
pink -= np.convolve(pink, np.ones(2048) / 2048, mode="same")
|
|
t = np.arange(n) / rate
|
|
tone = 0.25 * np.sin(2 * np.pi * (440 + 20 * np.sin(2 * np.pi * 0.5 * t)) * t)
|
|
x = (pink / (np.max(np.abs(pink)) + 1e-9) * 0.35 + tone).astype(np.float32)
|
|
return x[:, None], rate
|
|
|
|
|
|
def _mono48(wem):
|
|
x, sr = audio.decode_wem(wem)
|
|
return audio.to_channels(audio.resample(x, sr, RATE), 1)[:, 0]
|
|
|
|
|
|
def _sha1_file(p):
|
|
h = hashlib.sha1()
|
|
with open(p, "rb") as f:
|
|
for c in iter(lambda: f.read(1 << 22), b""):
|
|
h.update(c)
|
|
return h.hexdigest()
|
|
|
|
|
|
def run(ws, key=DEFAULT_TARGET, wem=None, mode="modded", seconds=40.0, threshold=0.5,
|
|
logdir=None, revert_after=True, log=print):
|
|
"""mode: 'modded' (apply `wem`, test, revert), 'baseline' (nothing), 'installed' (test the
|
|
current state). Returns a report dict with report['result'] in ('PASS', 'FAIL')."""
|
|
from .debugger import Run, kill_game
|
|
from .loopback import LoopbackRecorder, detect_chunks
|
|
from ..inject import pack_bnk_path, parse_target
|
|
|
|
s = ws.catalog.resolve(key)
|
|
wad = ws.wad_path(s)
|
|
files = [wad] + ([pack_bnk_path(wad, parse_target(s.target)[0])] if s.is_pack_media else [])
|
|
logdir = logdir or os.path.join(os.getcwd(), "logs")
|
|
os.makedirs(logdir, exist_ok=True)
|
|
stamp = time.strftime("%Y%m%d_%H%M%S")
|
|
logp = os.path.join(logdir, "ingame_%s.log" % stamp)
|
|
kill_game()
|
|
report = dict(target=s.key, name=s.name, mode=mode)
|
|
before = {f: _sha1_file(f) for f in files} if mode == "modded" else {}
|
|
|
|
if mode == "modded":
|
|
ws.apply([(s, wem, {"source": "ingame-test"})], progress=log)
|
|
expect = ws.current_bytes(s)
|
|
orig = ws.original_bytes(s)
|
|
from ..wem import Wem
|
|
report["expect_hash"] = "%08x" % Wem.parse(expect).vorb["codebook_hash"]
|
|
ref_expect, ref_orig = _mono48(expect), _mono48(orig)
|
|
|
|
rec = LoopbackRecorder(seconds + 10)
|
|
rec.start()
|
|
runner = Run(ws.game.exe, logp)
|
|
try:
|
|
report.update(runner.start())
|
|
while time.time() - runner.t0 < seconds and runner.alive():
|
|
time.sleep(0.5)
|
|
if re.search(r"^ACKIT_CRASH", runner.log(), re.M):
|
|
break
|
|
report["alive_at_end"] = runner.alive()
|
|
finally:
|
|
runner.stop()
|
|
rec.stop_flag = True
|
|
rec.join(5)
|
|
if mode == "modded" and revert_after:
|
|
ws.revert([s], progress=log)
|
|
report["reverted_byte_identical"] = all(_sha1_file(f) == h for f, h in before.items())
|
|
|
|
text = runner.log()
|
|
report["crash"] = bool(re.search(r"^ACKIT_CRASH", text, re.M))
|
|
ev = []
|
|
for ln in text.splitlines():
|
|
m = re.match(r"ACKIT_SETUP hash=([0-9a-f]+)", ln)
|
|
if m:
|
|
ev.append([m.group(1), "cached"])
|
|
m = re.match(r"ACKIT_PARSE ret=([0-9a-f]+)", ln)
|
|
if m and ev and ev[-1][1] == "cached":
|
|
ev[-1][1] = m.group(1)
|
|
report["setups"] = ["%s:%s" % tuple(e) for e in ev]
|
|
mine = [r for h, r in ev if h == report["expect_hash"]]
|
|
report["setup_ok"] = bool(mine) and all(r in ("0", "cached") for r in mine)
|
|
cap = rec.audio()
|
|
report["capture_error"] = rec.error
|
|
report["ncc_expected"] = round(detect_chunks(cap, ref_expect)[0], 3)
|
|
report["ncc_original"] = round(detect_chunks(cap, ref_orig)[0], 3)
|
|
ok = (not report["crash"] and report["alive_at_end"] and report["setup_ok"]
|
|
and report["ncc_expected"] >= threshold
|
|
and (mode == "baseline" or sha256(expect) == sha256(orig) or report["ncc_expected"] > report["ncc_original"])
|
|
and report.get("reverted_byte_identical", True))
|
|
report["result"] = "PASS" if ok else "FAIL"
|
|
json.dump(report, open(logp.replace(".log", ".json"), "w"), indent=1)
|
|
return report
|
|
|
|
|
|
# --- texts --------------------------------------------------------------------------------------
|
|
def run_texts(ws, seconds=45.0, menu_files=("FRONTEND", "GFXUI"), n_menu=12, n_subs=12, logdir=None, log=print):
|
|
"""Write markers into menu texts of the game's language (front-end files, loaded at startup) and into some
|
|
subtitles, start the game, then check: no crash, the menu markers are in the game's memory (it loaded the
|
|
edited localization files), and after the revert every touched file is byte-identical. Subtitles are not
|
|
loaded on the title screen, so for them the check is that the patched packages do not break the game."""
|
|
import random
|
|
from .debugger import Run, kill_game, search_memory
|
|
tx = ws.texts
|
|
lang = ws.game.language()
|
|
menu = [e for e in tx.menu(lang) if e.file.split(".")[0].upper() in menu_files and len(e.value) >= 4]
|
|
rnd = random.Random(3)
|
|
picks = rnd.sample(menu, min(n_menu, len(menu)))
|
|
changes = {"MENU:" + e.key: "ACKIT%02d %s" % (i, e.value) for i, e in enumerate(picks)}
|
|
subs = [s for s in tx.subtitles(lang) if s.room >= 12] if lang in tx.subtitle_languages() else []
|
|
for i, s in enumerate(rnd.sample(subs, min(n_subs, len(subs)))):
|
|
changes["SUB:%s:%s:%d" % (lang, s.speech, s.line)] = ("AKS%02d " % i + s.text)[:s.room]
|
|
files = sorted({p for k in changes for p in tx.files(k)})
|
|
logdir = logdir or os.path.join(os.getcwd(), "logs")
|
|
os.makedirs(logdir, exist_ok=True)
|
|
logp = os.path.join(logdir, "ingame_texts_%s.log" % time.strftime("%Y%m%d_%H%M%S"))
|
|
report = dict(mode="texts", lang=lang, menu_texts=len(picks), subtitles=len(changes) - len(picks),
|
|
files=len(files))
|
|
kill_game()
|
|
before = {f: _sha1_file(f) for f in files}
|
|
tx.apply(changes, progress=log)
|
|
needles = {("ACKIT%02d " % i).encode("utf-16-le") for i in range(len(picks))}
|
|
runner = Run(ws.game.exe, logp)
|
|
found = set()
|
|
try:
|
|
report.update(runner.start())
|
|
while time.time() - runner.t0 < seconds and runner.alive():
|
|
time.sleep(0.5)
|
|
if re.search(r"^ACKIT_CRASH", runner.log(), re.M):
|
|
break
|
|
if runner.alive():
|
|
found = search_memory(report["pid"], needles)
|
|
report["alive_at_end"] = runner.alive()
|
|
finally:
|
|
runner.stop()
|
|
tx.revert(list(changes), progress=log)
|
|
report["reverted_byte_identical"] = all(_sha1_file(f) == h for f, h in before.items())
|
|
report["crash"] = bool(re.search(r"^ACKIT_CRASH", runner.log(), re.M))
|
|
report["menu_markers_in_memory"] = len(found)
|
|
ok = (not report["crash"] and report["alive_at_end"] and report["menu_markers_in_memory"] > 0
|
|
and report["reverted_byte_identical"])
|
|
report["result"] = "PASS" if ok else "FAIL"
|
|
json.dump(report, open(logp.replace(".log", ".json"), "w"), indent=1)
|
|
return report
|
|
|
|
|
|
# --- in-memory sounds (soundbanks inside the packages) ------------------------------------------
|
|
PROBE_RATE, PROBE_Q = 28000, 4 # a codebook setup the game's own WEMs do not use (E10)
|
|
|
|
|
|
def bank_probes(ws, sounds):
|
|
"""[(sound, probe WEM)]: the same probe format for every sound, shortened to fit its slot."""
|
|
from ..encoder import encode
|
|
x, sr = probe_signal(10.0)
|
|
out = []
|
|
for s in sounds:
|
|
lim = ws.limit(s)
|
|
secs = min(10.0, max(0.4, s.seconds or 1.0))
|
|
while secs >= 0.1:
|
|
wem, _ = encode(x[:int(secs * sr)], sr, PROBE_RATE, 1, PROBE_Q)
|
|
if len(wem) <= lim:
|
|
out.append((s, wem))
|
|
break
|
|
secs *= 0.7
|
|
return out
|
|
|
|
|
|
def run_banks(ws, keys=None, seconds=45.0, logdir=None, log=print):
|
|
"""Replace in-memory sounds (default: every sound stored in the Frontend*/Startup packages) with probes, start the game,
|
|
then check: no crash, the probe data is in the game's memory (the modified packages were loaded), and
|
|
after the revert every touched file is byte-identical. Probes decoded (codec setup log) and heard
|
|
(loopback correlation) are reported too, but no in-memory sound plays on the title screen without input,
|
|
so they are not required. Returns a report dict (report['result'] PASS/FAIL)."""
|
|
from .debugger import Run, kill_game, search_memory
|
|
from .loopback import LoopbackRecorder, detect_chunks
|
|
from ..wem import Wem
|
|
if keys:
|
|
sounds = [ws.catalog.resolve(k) for k in keys]
|
|
else: # everything stored in the packages loaded for the title screen (rain ambience plays without input)
|
|
sounds = [s for s in ws.catalog.sounds if s.archive == "BANK" and any(
|
|
l.file.lower().startswith(("frontend", "startup")) for l in ws.locations(s))]
|
|
probes = bank_probes(ws, sounds)
|
|
files = sorted({f for s, _ in probes for f in ws.files(s)})
|
|
logdir = logdir or os.path.join(os.getcwd(), "logs")
|
|
os.makedirs(logdir, exist_ok=True)
|
|
logp = os.path.join(logdir, "ingame_banks_%s.log" % time.strftime("%Y%m%d_%H%M%S"))
|
|
report = dict(mode="banks", sounds=len(sounds), probes=len(probes), files=len(files))
|
|
kill_game()
|
|
before = {f: _sha1_file(f) for f in files}
|
|
hashes = {"%08x" % Wem.parse(w).vorb["codebook_hash"] for _, w in probes}
|
|
needles = set()
|
|
for _, w in probes:
|
|
if len(w) > 400:
|
|
needles.add(w[len(w) // 2:len(w) // 2 + 32])
|
|
report.update(probe_hashes=sorted(hashes), needles=len(needles))
|
|
ws.apply([(s, w, {"source": "ingame-test"}) for s, w in probes], progress=log)
|
|
from .debugger import ADDR
|
|
# also log every setup parse (in-memory sources may not go through the stream setup cache)
|
|
extra = 'bp %#x ".printf \\"ACKIT_PARSE2 hash=%%08x\\\\n\\", poi(poi(@esp+4)+0x9c); gc"' % ADDR["parse_setup"]
|
|
runner = Run(ws.game.exe, logp, script_extra=extra)
|
|
found = set()
|
|
rec = LoopbackRecorder(seconds + 10)
|
|
rec.start()
|
|
try:
|
|
report.update(runner.start())
|
|
pid = report["pid"]
|
|
while time.time() - runner.t0 < seconds and runner.alive():
|
|
time.sleep(0.5)
|
|
if re.search(r"^ACKIT_CRASH", runner.log(), re.M):
|
|
break
|
|
if runner.alive():
|
|
found = search_memory(pid, needles)
|
|
report["alive_at_end"] = runner.alive()
|
|
finally:
|
|
runner.stop()
|
|
rec.stop_flag = True
|
|
rec.join(5)
|
|
ws.revert([s for s, _ in probes], progress=log)
|
|
report["reverted_byte_identical"] = all(_sha1_file(f) == h for f, h in before.items())
|
|
text = runner.log()
|
|
report["crash"] = bool(re.search(r"^ACKIT_CRASH", text, re.M))
|
|
ref = probe_signal(0.35)[0][:, 0]
|
|
report["capture_error"] = rec.error
|
|
report["ncc_probe"] = round(detect_chunks(rec.audio(), ref)[0], 3)
|
|
seen = re.findall(r"ACKIT_(?:SETUP|PARSE2) hash=([0-9a-f]+)", text)
|
|
report["probe_streams_decoded"] = sum(1 for h in seen if h in hashes)
|
|
report["setups_logged"] = len(seen)
|
|
report["probes_in_memory"] = len(found)
|
|
ok = (not report["crash"] and report["alive_at_end"] and report["probes_in_memory"] > 0
|
|
and report["reverted_byte_identical"])
|
|
report["result"] = "PASS" if ok else "FAIL"
|
|
json.dump(report, open(logp.replace(".log", ".json"), "w"), indent=1)
|
|
return report
|