#!/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 # say TEXT type TEXT, press Return, give the game a second # to answer, then clear # 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. # # Time is the script's, as in tests/hostPort.c's virtual time: every step # happens at its own time since the interpreter started (the moment DOSBox # logs the switch to its graphics mode), a key taking KEY_HOLD + KEY_GAP, a # cleared window CLEAR_GAP and a screenshot no time at all. What a step # costs here (the capture, each xdotool call) delays at most the next one # and never the rest of the script. A recording's stop is the exception: # the schedule starts again once the video is written. # Env: DOSBOX (default JoeyLib's toolchains/opt/dosbox/dosbox), JOEYLIB. import datetime 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") # JoeyLib's 386SX-16 configuration, but a fast 386: Sierra's interpreter # paces itself by the clock, so speed only shortens the time it spends # drawing and loading, which the host harness spends none of (at 2200 # cycles/ms Black Cauldron's title picture alone put it 1.9 s behind). CPU_CYCLES = "50000" 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 SAY_SETTLE = 1.0 # The interpreter's start: DOSBox's log line for a graphics mode, and the # timestamp at the head of each log line. START_MARK = "graphics mode" START_POLLS = 100 START_POLL_SECONDS = 0.1 LOG_TIME_FORMAT = "%Y-%m-%d %H:%M:%S.%f" LOG_TIME_LEN = 23 # 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] args += ["--set", "cpu_cycles=" + CPU_CYCLES, "--set", "cpu_cycles_protected=" + CPU_CYCLES] for s in (extraSet or []): args += ["--set", s] args += ["-c", "MOUNT C \"%s\"" % hostDir, "-c", "C:", "-c", command] self.log = os.path.join(self.work, "dosbox.log") self.proc = subprocess.Popen(args, env=self.env, stdout=open(self.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 = self.interpreterStart() self.due = 0.0 def interpreterStart(self): # When the program switched to a graphics mode, from DOSBox's log # (the host harness's frame 0); the window's appearance if it never # says so. for _ in range(START_POLLS): with open(self.log, errors="replace") as f: for line in f: if START_MARK in line: try: return datetime.datetime.strptime(line[:LOG_TIME_LEN], LOG_TIME_FORMAT).timestamp() except ValueError: break time.sleep(START_POLL_SECONDS) print("no graphics mode in DOSBox's log: timing from the window", file=sys.stderr) return time.time() def sleepUntil(self, due): delay = self.t0 + due - time.time() if delay > 0: time.sleep(delay) 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.sleepUntil(self.due) self.xdo("keydown", "--window", self.win, name) self.sleepUntil(self.due + hold) self.xdo("keyup", "--window", self.win, name) self.due += hold + KEY_GAP def typeText(self, text): for ch in text: if ch.isupper(): self.sleepUntil(self.due) 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): self.sleepUntil(self.due) 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.sleepUntil(self.due) 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 self.t0 = time.time() - self.due 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 # The Return is pressed within the gap, as on the host. start = self.due self.key("Return") self.due = start + 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": self.due += float(arg) self.sleepUntil(self.due) 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 == "say": self.typeText(arg) self.key("Return") self.due += SAY_SETTLE self.clearWindows() 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.sleepUntil(self.due) self.xdo("keydown", "--window", self.win, arg) elif op == "up": self.sleepUntil(self.due) 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()