Something went wrong. Try again.
HEXWORK - A Lantern in the Deep: 3D ASCII terminal roguelike RPG in Rust.
Entirely designed and built by Muse Spark 1.3 to evaluate if I like the model.
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191#!/usr/bin/env python3"""Bounded stdlib PTY proof for live boot presentation preferences."""
import errnoimport fcntlimport jsonimport osimport pathlibimport reimport selectorsimport structimport subprocessimport sysimport termiosimport time
DEADLINE_SECONDS = 240.0CAPTURE_LIMIT = 4 * 1024 * 1024ANSI = re.compile(rb"\x1b(?:\[[0-?]*[ -/]*[@-~]|\][^\x07]*(?:\x07|\x1b\\)|[@-_])")PLAY_CUE = b"Esc quit"
def stripped(data): return ANSI.sub(b"", data).replace(b"\r", b"")
def spawn(binary, data_dir, content, extra): master, slave = os.openpty() fcntl.ioctl(slave, termios.TIOCSWINSZ, struct.pack("HHHH", 24, 100, 0, 0)) child = subprocess.Popen( [binary, "--data-dir", str(data_dir), "--content-root", str(content)] + list(extra), stdin=slave, stdout=slave, stderr=slave, close_fds=True, start_new_session=True, ) os.close(slave) os.set_blocking(master, False) return child, master
def terminate(child): if child.poll() is None: child.kill() try: child.wait(timeout=5) except subprocess.TimeoutExpired: child.kill() child.wait(timeout=5)
def drive(binary, data_dir, content, extra, cue, on_cue, keys, expected_exit=0): child, master = spawn(binary, data_dir, content, extra) selector = selectors.DefaultSelector() selector.register(master, selectors.EVENT_READ) registered = True captured = bytearray() acted = False deadline = time.monotonic() + DEADLINE_SECONDS try: while child.poll() is None or registered: remaining = deadline - time.monotonic() if remaining <= 0: raise RuntimeError("PTY child or output drain exceeded 240 second deadline") if registered: for _, _ in selector.select(timeout=min(0.25, remaining)): try: chunk = os.read(master, 65536) except OSError as error: if error.errno == errno.EIO: chunk = b"" elif error.errno in (errno.EAGAIN, errno.EWOULDBLOCK): continue else: raise if not chunk: selector.unregister(master) registered = False break if len(captured) + len(chunk) > CAPTURE_LIMIT: raise RuntimeError("PTY capture exceeded 4 MiB") captured.extend(chunk) else: try: child.wait(timeout=min(0.25, remaining)) except subprocess.TimeoutExpired: pass if ( child.poll() is None and registered and not acted and cue is not None and stripped(bytes(captured)).find(cue) >= 0 ): on_cue(bytes(captured)) if keys: os.write(master, keys) acted = True status = child.wait(timeout=5) if cue is not None and not acted: raise RuntimeError("child exited before render cue; tail=%r" % stripped(bytes(captured))[-2000:]) if status != expected_exit: raise RuntimeError("child exit %d expected %d; tail=%r" % (status, expected_exit, stripped(bytes(captured))[-2000:])) return bytes(captured), status finally: selector.close() os.close(master) terminate(child)
def only_run_id(data_dir): values = sorted(path.name for path in (data_dir / "runs").iterdir() if path.is_dir()) if len(values) != 1: raise RuntimeError("expected exactly one run, got %r" % values) return values[0]
def explicit_args(): return ["--ancestry", "1", "--calling", "1", "--gamma", "1.5", "--fov", "60", "--palette", "protan"]
def assert_complete_preferences(path): value = json.loads(path.read_text(encoding="utf-8")) expected = {"schema_version": 1, "gamma": 1.5, "fov_deg": 60.0, "palette": "protan"} if value != expected: raise RuntimeError("unexpected preferences: %r" % value)
def main(): if len(sys.argv) != 5: raise RuntimeError("usage: presentation_scenario.py BINARY DATA CONTENT MODE") binary = sys.argv[1] data_dir = pathlib.Path(sys.argv[2]) content = pathlib.Path(sys.argv[3]) mode = sys.argv[4] data_dir.mkdir(parents=True, exist_ok=True) path = data_dir / "presentation.json"
if mode == "contended": captured, status = drive(binary, data_dir, content, explicit_args(), None, lambda _: None, b"", 6) summary = {"exit": status, "contention": b"lock" in stripped(captured).lower()} elif mode == "publish": before_quit = [] def published(_): assert_complete_preferences(path) before_quit.append(True) drive(binary, data_dir, content, explicit_args(), PLAY_CUE, published, b"\x1b") summary = {"published_before_quit": bool(before_quit)} elif mode == "persist-reload": before_quit = [] def published(_): assert_complete_preferences(path) before_quit.append(True) first, _ = drive(binary, data_dir, content, explicit_args(), PLAY_CUE, published, b"\x1b") run_id = only_run_id(data_dir) run_save_path = data_dir / "runs" / run_id / "run.save" capture_root = data_dir.parent first_save_capture = capture_root / "presentation-first-actual.run.save" second_save_capture = capture_root / "presentation-second-actual.run.save" first_save_capture.write_bytes(run_save_path.read_bytes()) before_bytes = path.read_bytes() before_mtime = path.stat().st_mtime_ns second, _ = drive(binary, data_dir, content, ["--load", run_id], PLAY_CUE, lambda _: None, b"\x1b") second_save_capture.write_bytes(run_save_path.read_bytes()) plain_second = stripped(second) summary = { "published_before_quit": bool(before_quit), "reload_completed": b"HEXWORK session over" in plain_second, "bytes_unchanged": path.read_bytes() == before_bytes, "mtime_unchanged": path.stat().st_mtime_ns == before_mtime, "run_id_unchanged": only_run_id(data_dir) == run_id, "reload_status_has_values": b"source=file gamma=1.5 fov=60 palette=protan commit=unchanged" in plain_second, "first_rendered": PLAY_CUE in stripped(first), } elif mode == "defaults": rendered = [] drive(binary, data_dir, content, ["--ancestry", "1", "--calling", "1"], PLAY_CUE, lambda _: rendered.append(True), b"\x1b") summary = {"rendered": bool(rendered), "presentation_absent": not path.exists()} else: raise RuntimeError("unknown scenario mode: " + mode) sys.stdout.write(json.dumps(summary, separators=(",", ":")) + "\n")
if __name__ == "__main__": try: main() except Exception as error: sys.stderr.write(("presentation PTY scenario failed: %s\n" % error)[-8192:]) raise