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.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172#!/usr/bin/env python3"""Bounded 80x24 decoded-grid proof for the live essentials chrome."""import codecsimport errnoimport fcntlimport jsonimport osimport selectorsimport structimport subprocessimport sysimport termiosimport timeimport unicodedata
class Grid: """Incremental decoder for the VT sequences emitted by this application.
SGR and cursor visibility do not affect text. Unknown CSI fails closed; UTF-8 and CSI may arrive across arbitrary PTY read boundaries. """ def __init__(self): self.cells = [[" " for _ in range(80)] for _ in range(24)] self.row = self.col = 0 self.state = "text" self.params = "" self.frames = 0 self.completed = None self.utf8 = codecs.getincrementaldecoder("utf-8")("strict")
def feed(self, data): for char in self.utf8.decode(data): if self.state == "esc": if char != "[": raise RuntimeError("unsupported escape %r" % char) self.state = "csi" self.params = "" continue if self.state == "csi": if "@" <= char <= "~": self.csi(char) self.state = "text" else: self.params += char continue if char == "\x1b": self.state = "esc" elif char == "\r": self.col = 0 elif char == "\n": self.row = min(self.row + 1, 23) elif char == "\x07": pass elif ord(char) < 32: raise RuntimeError("unsupported control %r" % char) else: width = 0 if unicodedata.combining(char) else (2 if unicodedata.east_asian_width(char) in "WF" else 1) if width == 0: if 0 < self.col <= 80: self.cells[self.row][self.col - 1] += char continue if self.col >= 80: self.col = 0 self.row = min(self.row + 1, 23) if 0 <= self.row < 24 and self.col + width <= 80: self.cells[self.row][self.col] = char if width == 2: self.cells[self.row][self.col + 1] = "" self.col += width if self.row == 23 and self.col == 80: self.frames += 1 self.completed = self.rows()
def csi(self, command): p = self.params if command in "Hf": values = [int(x or "1") for x in p.split(";")] self.row = (values[0] if values else 1) - 1 self.col = (values[1] if len(values) > 1 else 1) - 1 elif command == "J" and p in ("2", "3"): if p == "2": self.cells = [[" " for _ in range(80)] for _ in range(24)] elif command == "K" and p == "2": self.cells[self.row] = [" " for _ in range(80)] elif command == "m": pass elif command in "hl" and p in ("?1049", "?25", "?7", "?2026", "?1004"): pass else: raise RuntimeError("unsupported CSI %r %r" % (p, command))
def rows(self): return ["".join(row) for row in self.cells]
def main(): binary, data, content, label = sys.argv[1:] master, slave = os.openpty() fcntl.ioctl(slave, termios.TIOCSWINSZ, struct.pack("HHHH", 24, 80, 0, 0)) size = struct.unpack("HHHH", fcntl.ioctl(slave, termios.TIOCGWINSZ, b"\0" * 8)) assert size[:2] == (24, 80) child = subprocess.Popen( [binary, "--data-dir", data, "--content-root", content, "--seed", "48", "--name", "HUD essentials", "--ancestry", "1", "--calling", "1"], stdin=slave, stdout=slave, stderr=slave, close_fds=True, start_new_session=True) os.close(slave) os.set_blocking(master, False) selector = selectors.DefaultSelector() selector.register(master, selectors.EVENT_READ) registered = True grid = Grid() snapshots = [] seen_frame = 0 total = 0 deadline = time.monotonic() + 240 try: while child.poll() is None or registered: remaining = deadline - time.monotonic() if remaining <= 0: raise RuntimeError("PTY deadline; grid=%r" % grid.rows()) if registered: for _, _ in selector.select(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 total += len(chunk) if total > 4 * 1024 * 1024: raise RuntimeError("PTY capture limit exceeded") grid.feed(chunk) else: try: child.wait(timeout=min(0.25, remaining)) except subprocess.TimeoutExpired: pass if child.poll() is None and registered and grid.frames > seen_frame: seen_frame = grid.frames rows = grid.completed if len(snapshots) == 0 and rows[17].startswith(label + " quit") and rows[20].startswith("Recent feedback") and "TACTICAL" not in rows[19]: snapshots.append(rows) os.write(master, b"m") elif len(snapshots) == 1 and rows[17].startswith(label + " close") and "[TACTICAL Z " in rows[19]: snapshots.append(rows) # The advertised quit binding closes map before quitting Play. os.write(master, b"\x1b" if label == "Esc" else label.encode("ascii")) elif len(snapshots) == 2 and rows[17].startswith(label + " quit") and "TACTICAL" not in rows[19]: os.write(master, b"\x1b" if label == "Esc" else label.encode("ascii")) status = child.wait(timeout=5) if status != 0 or len(snapshots) != 2: raise RuntimeError("exit=%r snapshots=%r grid=%r" % (status, len(snapshots), grid.rows())) print(json.dumps({"size": [80, 24], "exit": status, "play": snapshots[0], "map": snapshots[1]})) finally: selector.close() os.close(master) if child.poll() is None: child.kill() child.wait(timeout=5)
if __name__ == "__main__": main()