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.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248#!/usr/bin/env python3"""Independent VT + physical controls; requires opt-in production frame metadata.
No parsing renderer state or importing rendering code. Metadata is a diagnosticside channel only; all actions enter the real executable through its PTY."""import errnoimport fcntlimport jsonimport osimport pathlibimport selectorsimport signalimport structimport subprocessimport sysimport termiosimport timefrom vt import VT, export
class Driver: def __init__(self, binary, data, content, style, color): self.vt = VT(100,24) self.total = 0 self.metadata = pathlib.Path(data).parent / ('interface-'+style+'-'+color+'.jsonl') self.metadata.unlink(missing_ok=True) self.last = None self.frames = [] env = os.environ.copy() for name in ('HEXWORK_FORCE_TC','HEXWORK_FORCE_256','HEXWORK_NO_COLOR','HEXWORK_FORCE_ASCII','NO_COLOR','HEXWORK_CONTENT_ROOT'): env.pop(name, None) env.update(TERM='xterm-256color', COLORTERM='truecolor', LANG='C.UTF-8', HEXWORK_INTERFACE_METADATA=str(self.metadata)) env[{'truecolor':'HEXWORK_FORCE_TC','indexed':'HEXWORK_FORCE_256','none':'HEXWORK_NO_COLOR'}[color]]='1' if os.environ.get('HEXWORK_PTY_ASCII') == '1': env['HEXWORK_FORCE_ASCII']='1' args=[binary,'--data-dir',data,'--content-root',content,'--seed',os.environ.get('HEXWORK_PTY_SEED','223'), '--name','Interface evidence','--ancestry','1','--calling','1'] if style in ('story','legacy','controls'): args.append('--story') if style == 'controls': env['HEXWORK_MOUSE']='1' if style == 'legacy': pathlib.Path(data).mkdir(parents=True,exist_ok=True) (pathlib.Path(data)/'keymap.toml').write_bytes(pathlib.Path(__file__).with_name('legacy-interface-keymap.toml').read_bytes()) if style == 'resume': runs=sorted(path.name for path in (pathlib.Path(data)/'runs').iterdir() if path.is_dir()) assert len(runs)==1, runs args=[binary,'--data-dir',data,'--content-root',content,'--load',runs[0]] self.master, slave = os.openpty() fcntl.ioctl(slave,termios.TIOCSWINSZ,struct.pack('HHHH',24,100,0,0)) try: self.child=subprocess.Popen(args,stdin=slave,stdout=slave,stderr=slave,env=env, close_fds=True,start_new_session=True) finally: os.close(slave) os.set_blocking(self.master,False) self.selector=selectors.DefaultSelector() self.selector.register(self.master,selectors.EVENT_READ) self.style,self.color=style,color
def pump(self, timeout=0.05): for _, _ in self.selector.select(timeout): while True: try: chunk=os.read(self.master,65536) except OSError as error: if error.errno in (errno.EAGAIN,errno.EWOULDBLOCK): break if error.errno == errno.EIO: return raise if not chunk: return self.total+=len(chunk) if self.total > 32*1024*1024: raise RuntimeError('PTY exceeds 32 MiB') self.vt.feed(chunk)
def records(self): if not self.metadata.exists(): return [] lines=self.metadata.read_bytes().split(b'\n')[:-1] return [json.loads(line) for line in lines if line]
def wait(self, predicate, label, previous=-1): deadline=time.monotonic()+180 while time.monotonic()<deadline: self.pump() for raw in self.records(): value=dict(raw) value['mode']={'FirstPerson':'first-person','ChaseOrbit':'chase','TopTactical':'tactical'}.get(raw['mode'],raw['mode']) if 'inventory' in str(raw.get('screen','')).lower(): value['mode']='inventory' if value['frame'] > previous and predicate(value): # Metadata is written after flushing the terminal frame. Drain # already available bytes before inspecting the reconstructed grid. self.pump(0) self.last=value self.frames.append(label) artifact_name=self.style+'-'+self.color+'-seed'+str(value['seed'])+('-ascii' if value.get('ascii_only') else '-unicode')+'-'+label export(self.vt,artifact_name,value) return value if self.child.poll() is not None: raise RuntimeError('child exited before '+label+': '+self.vt.text()) raise RuntimeError('deadline waiting for '+label+'; metadata adapter present='+str(self.metadata.exists())+'; screen='+self.vt.text())
def key(self, wire, predicate, label): before=self.last['frame'] os.write(self.master,wire) return self.wait(predicate,label,before)
def resize(self,width,height): self.vt.resize(width,height) fcntl.ioctl(self.master,termios.TIOCSWINSZ,struct.pack('HHHH',height,width,0,0)) os.kill(self.child.pid,signal.SIGWINCH)
def close(self): self.selector.close() os.close(self.master) if self.child.poll() is None: self.child.kill() self.child.wait(timeout=5)
def canonical(value): return tuple(value[key] for key in ('tick','diagnostic_hash','history_len'))
def main(): binary,data,content,style,color=sys.argv[1:] driver=Driver(binary,data,content,style,color) try: initial=driver.wait(lambda m:not m['look_active'] and m['mode']=='first-person','play') assert 'Esc' in driver.vt.text(), driver.vt.text() assert initial['seed']==int(os.environ.get('HEXWORK_PTY_SEED','223')) assert initial['build_identifier'].startswith('fnv1a64:') if os.environ.get('HEXWORK_PTY_ASCII') == '1': assert initial['ascii_only'] is True assert all(ord(cell[0])<128 for row in driver.vt.cells[:16] for cell in row) if style=='resume': prior=json.loads(driver.metadata.with_name('interface-story-'+color+'.jsonl').read_text().splitlines()[-1]) assert initial['run_id']==prior['run_id'] and initial['build_identifier']==prior['build_identifier'] assert initial['history_len']==prior['history_len'] and initial['player_location']==prior['player_location'] driver.key(b'\t',lambda m:m['mode']=='chase','resume-camera') tactical=driver.key(b'm',lambda m:m['mode']=='tactical','resume-map') sliced=driver.key(b'\x1b[5~',lambda m:m['selected_z']!=m['player_z'],'resume-slice') home=driver.key(b'\x1b[H',lambda m:m['selected_z']==m['player_z'],'resume-home') assert canonical(tactical)==canonical(sliced)==canonical(home) driver.key(b'\x1b',lambda m:m['mode']=='chase','resume-return') assert driver.last['history_len']==initial['history_len'] os.write(driver.master,b'\x1b') deadline=time.monotonic()+180 while driver.child.poll() is None and time.monotonic()<deadline: driver.pump() assert driver.child.wait(timeout=5)==0 print(json.dumps({'style':style,'color':color,'stages':driver.frames,'exit':0},sort_keys=True)) return colors=[cell[1:3] for row in driver.vt.cells for cell in row] if color=='truecolor': assert any(isinstance(c,tuple) for pair in colors for c in pair) elif color=='indexed': assert any(isinstance(c,int) for pair in colors for c in pair) else: assert all(c is None for pair in colors for c in pair) if style=='controls': import math assert initial['mouse_capture'] is True driver.key(b'\x1b[<0;10;10M',lambda m:True,'mouse-down') drag=driver.key(b'\x1b[<32;13;8M',lambda m:m['yaw']!=initial['yaw'],'mouse-drag') assert abs(drag['yaw']-initial['yaw']-math.radians(6))<1e-5 assert abs(drag['pitch']-math.radians(4))<1e-5 assert drag['mode']=='first-person' driver.key(b'\x1b[O',lambda m:True,'focus-lost') stopped=driver.key(b'\x1b[<32;30;30M',lambda m:True,'drag-after-focus') assert (stopped['yaw'],stopped['pitch'])==(drag['yaw'],drag['pitch']) driver.key(b'\x1b[<0;30;30m',lambda m:True,'mouse-up') driver.key(b'x',lambda m:m['look_active'],'controls-look') reset=driver.key(b'0',lambda m:m['pitch']==0,'controls-reset') turned=driver.key(b'\x1b[C'*6,lambda m:abs(m['yaw']-reset['yaw']-math.pi/2)<1e-5,'controls-east') driver.key(b'\x1b',lambda m:not m['look_active'],'controls-look-exit') assert turned['history_len']==initial['history_len'] moved=driver.key(b'w',lambda m:m['history_len']>initial['history_len'],'camera-relative-forward') old=initial['player_location']; new=moved['player_location'] assert new==[old[0]+1,old[1],old[2]], (old,new,moved) os.write(driver.master,b'\x1b') deadline=time.monotonic()+180 while driver.child.poll() is None and time.monotonic()<deadline: driver.pump() assert driver.child.wait(timeout=5)==0 print(json.dumps({'style':style,'color':color,'stages':driver.frames,'exit':0},sort_keys=True)) return if style=='legacy': tactical=driver.key(b'm',lambda m:m['mode']=='tactical','legacy-map') text=driver.vt.text() assert 'Home recenter' not in text and 'PageUp/PageDown' not in text and 'PgUp/PgDn' not in text, text moved=driver.key(b'\x1b[5~',lambda m:m['mode']=='tactical','legacy-pageup') assert moved['selected_z']==tactical['selected_z'], (tactical,moved) waited=driver.key(b'0',lambda m:m['tick']>moved['tick'],'legacy-zero-wait') assert waited['history_len']>moved['history_len'], (moved,waited) driver.key(b'\x1b',lambda m:m['mode']=='first-person','legacy-map-exit') os.write(driver.master,b'\x1b') deadline=time.monotonic()+180 while driver.child.poll() is None and time.monotonic()<deadline: driver.pump() assert driver.child.wait(timeout=5)==0 print(json.dumps({'style':style,'color':color,'stages':driver.frames,'exit':0},sort_keys=True)) return look=driver.key(b'x',lambda m:m['look_active'],'look') turned=driver.key(b'\x1b[C',lambda m:m['look_active'] and m['yaw']!=look['yaw'],'look-right') raised=driver.key(b'\x1b[A',lambda m:m['pitch']!=turned['pitch'],'look-up') reset=driver.key(b'0',lambda m:m['pitch']==0,'look-reset') driver.key(b'\x1b',lambda m:not m['look_active'],'look-exit') chase=driver.key(b'\t',lambda m:m['mode']=='chase','chase') driver.key(b'i',lambda m:m['mode']=='inventory','inventory') restored=driver.key(b'\x1b',lambda m:m['mode']=='chase','modal-return') assert restored['yaw']==chase['yaw'] and restored['elevation']==chase['elevation'] tactical=driver.key(b'm',lambda m:m['mode']=='tactical','map') assert ('PAUSE-STEP' if style=='story' else 'LIVE') in driver.vt.text() off=driver.key(b'\x1b[5~',lambda m:m['selected_z']!=m['player_z'],'slice-up') assert 'Home' in driver.vt.text(), driver.vt.text() rejected=driver.key(b'\x1b[A',lambda m:m['selected_z']==off['selected_z'],'off-slice-move') assert canonical(rejected)==canonical(off), (off,rejected) driver.key(b'\x1b[H',lambda m:m['selected_z']==m['player_z'],'recenter') # Stay tactical during resize: Story pauses here, but perspective is # intentionally live. This isolates suspension from ordinary live ticks. before=driver.last['frame']; driver.resize(40,10) deadline=time.monotonic()+30 while 'Resize to 80x24; play suspended.' not in driver.vt.text() and time.monotonic()<deadline: driver.pump() assert driver.vt.text().splitlines()[0].rstrip() == 'Resize to 80x24; play suspended.', driver.vt.text() # The warning is intentionally clipped at 40 columns. Check actual # suspension too: physical actions during this interval must be ignored. os.write(driver.master,b'wwww') until=time.monotonic()+0.3 while time.monotonic()<until: driver.pump() driver.frames.append('too-small') export(driver.vt,style+'-'+color+'-seed'+str(initial['seed'])+('-ascii' if initial['ascii_only'] else '-unicode')+'-too-small',{'width':40,'height':10,'suspended':True}) driver.resize(100,24) final=driver.wait(lambda m:m['width']==100 and m['height']==24,'restored',before) before=final['frame']; driver.resize(120,36) grown=driver.wait(lambda m:m['width']==120 and m['height']==36,'grown',before) before=grown['frame']; driver.resize(80,24) final=driver.wait(lambda m:m['width']==80 and m['height']==24,'shrunk',before) assert len(driver.vt.cells)==24 and all(len(row)==80 for row in driver.vt.cells) if style=='story': assert canonical(final)==canonical(tactical), (tactical,final) assert final['history_len']==initial['history_len'] and final['player_location']==initial['player_location'], (initial,final) driver.key(b'\x1b',lambda m:m['mode']=='chase','map-exit') # Chrome must not inherit scene paint, regardless of selected color mode. for row in driver.vt.cells: text=''.join(cell[0] for cell in row) if 'Esc' in text or 'Recent feedback' in text: assert all(cell[1:] == (None,None,False) for cell in row if cell[0]!=' '), row os.write(driver.master,b'\x1b') deadline=time.monotonic()+180 while driver.child.poll() is None and time.monotonic()<deadline: driver.pump() assert driver.child.wait(timeout=5)==0 print(json.dumps({'style':style,'color':color,'stages':driver.frames,'exit':0},sort_keys=True)) finally: driver.close()
if __name__=='__main__': main()