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.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195#!/usr/bin/env python3"""Bounded stdlib PTY proof for a recorded blocked move and continued play."""
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\\)|[@-_])")MENU_CUE = b"Cancel: [Esc]"PLAY_CUE = b"Recent feedback"FOLLOWUP_DELAY = 0.4
def stripped(data): return ANSI.sub(b"", data).replace(b"\r", b"")
def spawn(binary, data_dir, content, seed): master, slave = os.openpty() fcntl.ioctl(slave, termios.TIOCSWINSZ, struct.pack("HHHH", 24, 80, 0, 0)) child = subprocess.Popen( [ binary, "--data-dir", str(data_dir), "--content-root", str(content), "--seed", str(seed), ], 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 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 drive(binary, data_dir, content, seed, route): child, master = spawn(binary, data_dir, content, seed) selector = selectors.DefaultSelector() selector.register(master, selectors.EVENT_READ) registered = True captured = bytearray() stage = 0 search_from = 0 followup_at = None deadline = time.monotonic() + DEADLINE_SECONDS try: while child.poll() is None or registered: remaining = deadline - time.monotonic() if remaining <= 0.0: raise RuntimeError("PTY child or EOF 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
plain = stripped(bytes(captured)) if child.poll() is not None or not registered: continue if stage == 0: found = plain.find(MENU_CUE, search_from) if found >= 0: os.write(master, b"1") search_from = found + len(MENU_CUE) stage = 1 elif stage == 1: found = plain.find(MENU_CUE, search_from) if found >= 0: os.write(master, b"1") search_from = found + len(MENU_CUE) stage = 2 elif stage == 2: found = plain.find(PLAY_CUE, search_from) if found >= 0: # A blocked sample is an ordinary recorded action; the legal # follow-up keys in the same batch continue play in one batch. keys = ( route["prefix_keys"] + route["blocked_key"] + route["legal_keys"] ).encode("ascii") os.write(master, keys) search_from = found + len(PLAY_CUE) followup_at = time.monotonic() stage = 3 elif stage == 3: if time.monotonic() - followup_at >= FOLLOWUP_DELAY: os.write(master, b"\x1b") stage = 4
status = child.wait(timeout=5) if stage != 4: raise RuntimeError( "child exited before recorded blocked move/followup; stage=%d tail=%r" % (stage, stripped(bytes(captured))[-4000:]) ) if status != 0: raise RuntimeError( "child exit %d; tail=%r" % (status, stripped(bytes(captured))[-4000:]) ) return bytes(captured), status finally: selector.close() os.close(master) terminate(child)
def main(): if len(sys.argv) != 6: raise RuntimeError("usage: feedback_scenario.py BINARY DATA CONTENT ROUTE_JSON SEED") binary = sys.argv[1] data_dir = pathlib.Path(sys.argv[2]) content = pathlib.Path(sys.argv[3]) route = json.loads(pathlib.Path(sys.argv[4]).read_text(encoding="utf-8")) seed = int(sys.argv[5]) data_dir.mkdir(parents=True, exist_ok=True) if not isinstance(route.get("prefix_keys"), str) or any(key not in "wasd" for key in route["prefix_keys"]): raise RuntimeError("route prefix is not default WASD: %r" % route.get("prefix_keys")) blocked_key = route.get("blocked_key") if not isinstance(blocked_key, str) or len(blocked_key) != 1 or blocked_key not in "wasd": raise RuntimeError("blocked key is not one default WASD key: %s=%r" % ("blocked_key", blocked_key)) legal_keys = route.get("legal_keys") if not isinstance(legal_keys, str) or not legal_keys or any(key not in "wasd" for key in legal_keys): raise RuntimeError("legal followup is not default WASD: %s=%r" % ("legal_keys", legal_keys)) if route["prefix_len"] != len(route["prefix_keys"]) or route["prefix_len"] >= 64: raise RuntimeError("route prefix is not short and explicit: %r" % route)
captured, status = drive(binary, data_dir, content, seed, route) plain = stripped(captured) summary = { "exit": status, "session_over": b"HEXWORK session over" in plain, "run_id": only_run_id(data_dir), "prefix_len": route["prefix_len"], } sys.stdout.write(json.dumps(summary, separators=(",", ":")) + "\n")
if __name__ == "__main__": try: main() except Exception as error: sys.stderr.write(("feedback PTY scenario failed: %s\n" % error)[-8192:]) raise