231 lines
8.9 KiB
Python
Executable file
231 lines
8.9 KiB
Python
Executable file
#!/usr/bin/env python3
|
|
# dosDrive.py - run a DOS program in DOSBox Staging under a private Xvfb,
|
|
# type at it from an input script and take raw 320x200 screenshots
|
|
# (Ctrl+F5). scripts/oracle.sh drives Sierra's interpreter with it, using
|
|
# the same scripts tests/hostPort.c plays.
|
|
#
|
|
# dosDrive.py HOSTDIR COMMAND OUTDIR SCRIPT [--set section.key=value ...]
|
|
#
|
|
# Script lines ('#' starts a comment, except in a type line):
|
|
# wait SECONDS let SECONDS pass
|
|
# key NAME [HOLD] press xdotool key NAME for HOLD seconds (0.12)
|
|
# down NAME / up NAME hold / release a key
|
|
# type TEXT press each character's key in turn
|
|
# shot NAME save the screen as OUTDIR/NAME.png
|
|
# clear press Return while a message window is showing
|
|
# record NAME start recording every frame (70 Hz) ...
|
|
# stop ... and stop, leaving OUTDIR/NAME.avi
|
|
# A shot is Ctrl+F5, which the game also sees as a key press. Recording
|
|
# is bound to F12 by a mapper file, and a mapper file replaces every
|
|
# default binding -- the keyboard and Ctrl+F5 too -- so a script that
|
|
# records can do nothing else.
|
|
# Env: DOSBOX (default JoeyLib's toolchains/opt/dosbox/dosbox), JOEYLIB.
|
|
import glob
|
|
import os
|
|
import ctypes
|
|
import random
|
|
import shutil
|
|
import signal
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
|
|
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
|
import compareFrames
|
|
|
|
JOEYLIB = os.environ.get("JOEYLIB", os.path.join(os.path.dirname(os.path.abspath(__file__)), "../../../joeylib"))
|
|
DOSBOX = os.environ.get("DOSBOX", os.path.join(JOEYLIB, "toolchains/opt/dosbox/dosbox"))
|
|
CONF = os.path.join(JOEYLIB, "scripts/dosbox-386sx16.conf")
|
|
KEY_HOLD = 0.12
|
|
KEY_GAP = 0.08
|
|
CLEAR_GAP = 0.6
|
|
CLEAR_TRIES = 8
|
|
SHOT_POLLS = 40
|
|
VIDEO_KEY = "F12"
|
|
VIDEO_KEY_SDL_SCANCODE = 69
|
|
VIDEO_POLLS = 50
|
|
VIDEO_POLL_SECONDS = 0.3
|
|
SHOT_POLL_SECONDS = 0.1
|
|
WINDOW_POLLS = 50
|
|
# prctl option: a signal for the child when its parent dies.
|
|
PR_SET_PDEATHSIG = 1
|
|
KEYMAP = {" ": "space", ".": "period", ",": "comma", "'": "apostrophe", "-": "minus",
|
|
"?": "question", "!": "exclam", "\n": "Return"}
|
|
|
|
|
|
def dieWithParent():
|
|
# Runs in each child before it starts: Xvfb and DOSBox die with the
|
|
# driver, so a driver killed by timeout -s KILL leaves neither behind.
|
|
ctypes.CDLL(None).prctl(PR_SET_PDEATHSIG, signal.SIGKILL)
|
|
|
|
|
|
class Dos:
|
|
def __init__(self, hostDir, command, outDir, extraSet=None, recording=False):
|
|
self.out = outDir
|
|
os.makedirs(outDir, exist_ok=True)
|
|
self.work = os.path.join(outDir, ".work")
|
|
shutil.rmtree(self.work, ignore_errors=True)
|
|
os.makedirs(os.path.join(self.work, "cap"))
|
|
os.makedirs(os.path.join(self.work, "xdg"))
|
|
self.recording = None
|
|
self.disp = ":%d" % random.randint(300, 799)
|
|
self.xvfb = subprocess.Popen(["Xvfb", self.disp, "-screen", "0", "1024x768x24", "-nolisten", "tcp"],
|
|
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, preexec_fn=dieWithParent)
|
|
time.sleep(1)
|
|
self.env = dict(os.environ, DISPLAY=self.disp, XDG_CONFIG_HOME=os.path.join(self.work, "xdg"),
|
|
SDL_AUDIODRIVER="dummy")
|
|
args = [DOSBOX, "--noprimaryconf", "--nolocalconf", "--conf", CONF,
|
|
"--set", "capture_dir=" + os.path.join(self.work, "cap"),
|
|
"--set", "default_image_capture_formats=raw"]
|
|
if recording:
|
|
mapper = os.path.join(self.work, "video.map")
|
|
with open(mapper, "w") as f:
|
|
f.write('hand_video "key %d"\n' % VIDEO_KEY_SDL_SCANCODE)
|
|
args += ["--set", "mapperfile=" + mapper]
|
|
for s in (extraSet or []):
|
|
args += ["--set", s]
|
|
args += ["-c", "MOUNT C \"%s\"" % hostDir, "-c", "C:", "-c", command]
|
|
self.proc = subprocess.Popen(args, env=self.env, stdout=open(os.path.join(self.work, "dosbox.log"), "w"),
|
|
stderr=subprocess.STDOUT, preexec_fn=dieWithParent)
|
|
self.win = None
|
|
for _ in range(WINDOW_POLLS):
|
|
time.sleep(0.2)
|
|
self.win = self.findWin()
|
|
if self.win:
|
|
break
|
|
if not self.win:
|
|
self.stop()
|
|
raise RuntimeError("no DOSBox window")
|
|
self.t0 = time.time()
|
|
|
|
def findWin(self):
|
|
r = subprocess.run(["xdotool", "search", "--name", "."], env=self.env, capture_output=True, text=True)
|
|
for w in r.stdout.split():
|
|
p = subprocess.run(["xdotool", "getwindowpid", w], env=self.env, capture_output=True, text=True)
|
|
if p.stdout.strip() == str(self.proc.pid):
|
|
return w
|
|
return None
|
|
|
|
def xdo(self, *a):
|
|
subprocess.run(["xdotool"] + list(a), env=self.env, capture_output=True)
|
|
|
|
def key(self, name, hold=KEY_HOLD):
|
|
self.xdo("keydown", "--window", self.win, name)
|
|
time.sleep(hold)
|
|
self.xdo("keyup", "--window", self.win, name)
|
|
time.sleep(KEY_GAP)
|
|
|
|
def typeText(self, text):
|
|
for ch in text:
|
|
if ch.isupper():
|
|
self.xdo("keydown", "--window", self.win, "shift")
|
|
self.key(ch.lower())
|
|
self.xdo("keyup", "--window", self.win, "shift")
|
|
else:
|
|
self.key(KEYMAP.get(ch, ch))
|
|
|
|
def shot(self, name):
|
|
before = set(glob.glob(os.path.join(self.work, "cap", "*.png")))
|
|
self.xdo("key", "--window", self.win, "ctrl+F5")
|
|
for _ in range(SHOT_POLLS):
|
|
time.sleep(SHOT_POLL_SECONDS)
|
|
now = set(glob.glob(os.path.join(self.work, "cap", "*.png"))) - before
|
|
if now:
|
|
time.sleep(0.2)
|
|
shutil.move(now.pop(), os.path.join(self.out, name + ".png"))
|
|
return True
|
|
print("shot %s: no capture" % name, file=sys.stderr)
|
|
return False
|
|
|
|
def record(self, name):
|
|
self.recording = name
|
|
self.xdo("key", "--window", self.win, VIDEO_KEY)
|
|
|
|
def stopRecording(self):
|
|
# The file is finished when it stops growing.
|
|
self.xdo("key", "--window", self.win, VIDEO_KEY)
|
|
last = -1
|
|
for _ in range(VIDEO_POLLS):
|
|
time.sleep(VIDEO_POLL_SECONDS)
|
|
videos = glob.glob(os.path.join(self.work, "cap", "*.avi"))
|
|
if not videos:
|
|
continue
|
|
size = os.path.getsize(videos[0])
|
|
if size == last:
|
|
shutil.move(videos[0], os.path.join(self.out, self.recording + ".avi"))
|
|
self.recording = None
|
|
return True
|
|
last = size
|
|
print("record %s: no video" % self.recording, file=sys.stderr)
|
|
return False
|
|
|
|
def clearWindows(self):
|
|
for _ in range(CLEAR_TRIES):
|
|
path = os.path.join(self.out, "_clear.png")
|
|
self.shot("_clear")
|
|
showing = os.path.exists(path) and compareFrames.messageWindows(compareFrames.load(path))
|
|
if os.path.exists(path):
|
|
os.remove(path)
|
|
if not showing:
|
|
return
|
|
self.key("Return")
|
|
time.sleep(CLEAR_GAP)
|
|
|
|
def stop(self):
|
|
try:
|
|
self.proc.kill()
|
|
except OSError:
|
|
pass
|
|
self.proc.wait()
|
|
self.xvfb.kill()
|
|
self.xvfb.wait()
|
|
shutil.rmtree(self.work, ignore_errors=True)
|
|
|
|
def run(self, script):
|
|
for raw in script.splitlines():
|
|
line = raw.strip() if raw.strip().startswith("type ") else raw.split("#", 1)[0].strip()
|
|
if not line:
|
|
continue
|
|
op, _, arg = line.partition(" ")
|
|
if op == "wait":
|
|
time.sleep(float(arg))
|
|
elif op == "key":
|
|
parts = arg.split()
|
|
self.key(parts[0], float(parts[1]) if len(parts) > 1 else KEY_HOLD)
|
|
elif op == "type":
|
|
self.typeText(arg)
|
|
elif op == "shot":
|
|
self.shot(arg)
|
|
elif op == "clear":
|
|
self.clearWindows()
|
|
elif op == "record":
|
|
self.record(arg)
|
|
elif op == "stop":
|
|
self.stopRecording()
|
|
elif op == "down":
|
|
self.xdo("keydown", "--window", self.win, arg)
|
|
elif op == "up":
|
|
self.xdo("keyup", "--window", self.win, arg)
|
|
else:
|
|
raise ValueError("bad script op " + op)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
if len(sys.argv) < 5:
|
|
sys.exit("usage: dosDrive.py HOSTDIR COMMAND OUTDIR SCRIPT [--set section.key=value ...]")
|
|
hostDir, command, outDir, scriptFile = sys.argv[1:5]
|
|
sets = []
|
|
rest = sys.argv[5:]
|
|
while rest:
|
|
if rest[0] == "--set" and len(rest) > 1:
|
|
sets.append(rest[1])
|
|
rest = rest[2:]
|
|
else:
|
|
rest = rest[1:]
|
|
script = open(scriptFile).read()
|
|
recording = any(line.split()[:1] == ["record"] for line in script.splitlines())
|
|
dos = Dos(hostDir, command, outDir, sets, recording)
|
|
try:
|
|
dos.run(script)
|
|
finally:
|
|
dos.stop()
|