122 lines
4.9 KiB
Python
122 lines
4.9 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
|