#!/usr/bin/env python3 """ Keyboard-Driven Indivisible Stochastic Process ================================================ Based on Jacob Barandes (arXiv:2507.21192, 2302.10778). Formal mapping -------------- C : configuration space — the set of known keys (grows on first press) eⱼ : delta spike on key j — the state at a division event (we observed j) Γ(t←t₀) : N×N column-stochastic transition matrix — the nomological law updated from bigram history after every division event p(t) : Γ · eⱼ = column j of Γ — the predicted distribution over the NEXT keypress, given key j was just pressed; this is where all memory of the process lives (encoded in Γ, not in p(t₀)) Every keypress is a division event: 1. Observe key j → p(t₀) = eⱼ (delta spike, t₀ resets) 2. Update Γ from the new (prev→j) bigram observation 3. p(t) = column j of updated Γ → predicted distribution over next key Indivisibility: Γ is a primitive law between consecutive division events. It is NOT required to factor as a product of sub-step matrices. Controls: any printable key → division event (immediate) BACKSPACE → undo last keypress ESC / Ctrl-C → quit """ import sys, os, tty, termios, fcntl, re import numpy as np from typing import Optional # ── ANSI ───────────────────────────────────────────────────────────────────── RS = "\033[0m" BLD = "\033[1m" DIM = "\033[2m" CYN = "\033[96m" GRN = "\033[92m" YLW = "\033[93m" RED = "\033[91m" MAG = "\033[95m" def col(text, *codes): return "".join(codes) + str(text) + RS _ANSI = re.compile(r'\033\[[0-9;]*m') def vlen(s): return len(_ANSI.sub('', s)) # ── ISP model ───────────────────────────────────────────────────────────────── class KeyboardISP: """ Indivisible stochastic process over a keyboard configuration space. State ----- configs : ordered list of known keys — the configuration space C gamma : N×N column-stochastic matrix — Γ(t←t₀), the nomological law p : N-vector — current predicted distribution over next keypress = column of Γ for the last observed key = Γ · e_last (law of total probability, §2.2 eq.20) last_key : the key observed at the most recent division event (t₀) Every keypress is a division event: - p(t₀) collapses to a delta spike on the observed key - Γ is updated from the new bigram observation - p(t) = Γ · e_j (column j of updated Γ) """ PRIOR = 0.5 # Laplace / Krichevsky-Trofimov smoothing def __init__(self): self.configs = [] # C: ordered list of keys self.index = {} # key -> matrix index self.counts = np.zeros((0, 0)) # bigram counts[src_idx, dst_idx] self.gamma = None # Γ: current transition matrix self.p = None # predicted distribution over next key self.last_key = None # key at last division event (t₀) self.prev_key = None # key at division event before that self.history = [] # full sequence of observed keys self.n_events = 0 # total division events self.last_save_path = None # most recent save path # ── configuration space ─────────────────────────────────────────────── def _expand(self, key): """Add a new key to C, growing Γ by one row and column.""" n = len(self.configs) self.configs.append(key) self.index[key] = n new = np.zeros((n + 1, n + 1)) if n > 0: new[:n, :n] = self.counts self.counts = new def _recompute_gamma(self): """Recompute Γ from bigram counts with Laplace smoothing.""" N = len(self.configs) if N == 0: self.gamma = np.zeros((0, 0)) return # counts[src, dst]: number of times dst followed src # Γ[dst, src] = p(dst | src) → column src sums to 1 num = self.counts.T + self.PRIOR # (N, N), rows=dst cols=src self.gamma = num / num.sum(axis=0, keepdims=True) # ── division event ──────────────────────────────────────────────────── def press(self, key) -> dict: """ Process a keypress as a division event. Returns a summary dict for the renderer. """ self.n_events += 1 new_key = (key not in self.index) # Expand C if needed if new_key: self._expand(key) # Update bigram counts: (prev → key) if self.last_key is not None: src = self.index[self.last_key] dst = self.index[key] self.counts[src, dst] += 1 # Recompute Γ from updated counts prev_gamma = self.gamma self._recompute_gamma() # p(t) = Γ · eⱼ = column j of Γ (law of total probability) j = self.index[key] self.p = self.gamma[:, j].copy() self.history.append(key) self.prev_key = self.last_key self.last_key = key return { "key": key, "new_key": new_key, "n_events": self.n_events, "j": j, "p": self.p.copy(), "gamma": self.gamma.copy(), "configs": list(self.configs), "history": list(self.history), } def save(self, path: str): """ Save Gamma and the config space to a CSV file. Format: - First row: header = "key" followed by config labels (the column keys) - Each subsequent row: row-key label, then Gamma[i, j] values - Config order matches the matrix indices """ import csv with open(path, 'w', newline='') as f: writer = csv.writer(f) writer.writerow(['key'] + [k for k in self.configs]) for i, ci in enumerate(self.configs): writer.writerow([ci] + ['%.6f' % self.gamma[i, j] for j in range(len(self.configs))]) def undo(self): """Remove the last keypress and recompute state.""" if not self.history: return self.history.pop() self.n_events -= 1 # Rebuild from scratch self.configs = [] self.index = {} self.counts = np.zeros((0, 0)) self.gamma = None self.p = None self.last_key = None self.prev_key = None self.ck_violation = None saved = list(self.history) self.history = [] for k in saved: self.press(k) # ── Screen renderer ─────────────────────────────────────────────────────────── def fk(k): """Human-readable key label.""" specials = {' ': 'SPC', '\t': 'TAB', '\r': 'RET'} if k in specials: return specials[k] if not k.isprintable(): return '0x%02x' % ord(k) return k W = 72 # fixed content width — never reaches terminal edge def build_screen(isp: KeyboardISP) -> str: SEP = col('─' * W, DIM) TOP = col('━' * W, CYN) ln = [] # ── header ──────────────────────────────────────────────────────────── ln.append(col(" Keyboard Indivisible Stochastic Process", BLD, CYN)) ln.append(TOP) ln.append('') # ── configuration space ─────────────────────────────────────────────── cfgs = isp.configs ln.append(col(" Configuration space C (%d keys)" % len(cfgs), BLD)) if cfgs: ln.append(" " + " ".join( col("[%s]" % fk(k), YLW) + col("=%d" % i, DIM) for i, k in enumerate(cfgs))) else: ln.append(col(" (none yet — press any key to begin)", DIM)) ln.append('') # ── last division event ─────────────────────────────────────────────── ln.append(col(" Last division event", BLD)) if isp.last_key is not None: ln.append(" Observed key : " + col("[%s]" % fk(isp.last_key), CYN) + col(" (p(t₀) = delta spike on this key)", DIM)) if isp.prev_key is not None: ln.append(" Previous key : " + col("[%s]" % fk(isp.prev_key), YLW)) else: ln.append(col(" (no keypresses yet)", DIM)) ln.append(" Division events so far : %d" % isp.n_events) ln.append('') # ── transition matrix Γ ─────────────────────────────────────────────── ln.append(col(" Transition matrix Gamma [Gamma_ij = p(i next | j just pressed)]", BLD)) if isp.gamma is not None and cfgs: N = len(cfgs) cw = 7 # Column headers (source keys) hdr = " " + "".join(("%*s" % (cw, fk(k))) for k in cfgs) ln.append(col(hdr, YLW)) for i, ci in enumerate(cfgs): # Highlight the column of the last pressed key row = col(" %4s " % fk(ci), YLW) for j in range(N): v = isp.gamma[i, j] # Highlight selected column (last key pressed) if isp.last_key is not None and j == isp.index[isp.last_key]: clr = GRN if v > 0.3 else (YLW if v > 0.1 else DIM) row += col("%*.3f" % (cw, v), clr, BLD) else: clr = GRN if v > 0.5 else (YLW if v > 0.2 else DIM) row += col("%*.3f" % (cw, v), clr) ln.append(row) if isp.last_key is not None: ln.append(col(" (highlighted column = Gamma·e_j for last key)", DIM)) else: ln.append(col(" (awaiting first keypress)", DIM)) ln.append('') # ── predicted distribution p(t) ─────────────────────────────────────── ln.append(col(" p(t) = Gamma · e_j = predicted distribution over NEXT key", BLD)) if isp.p is not None and cfgs: bw = 30 # Sort by probability descending for readability order = np.argsort(isp.p)[::-1] for i in order: v = isp.p[i] f = int(v * bw) bar = col('█' * f, GRN) + col('░' * (bw - f), DIM) # Mark if this is the last observed key marker = col(' <- last pressed', DIM) if cfgs[i] == isp.last_key else '' ln.append(" %6s %s %.3f%s" % ( col("[%s]" % fk(cfgs[i]), YLW), bar, v, marker)) else: ln.append(col(" (awaiting first keypress)", DIM)) ln.append('') # ── history ─────────────────────────────────────────────────────────── ln.append(col(" Key history", BLD)) if isp.history: # Show last 40 keys recent = isp.history[-40:] ln.append(" " + " ".join(col("[%s]" % fk(k), CYN) for k in recent)) if len(isp.history) > 40: ln.append(col(" ... (%d total)" % len(isp.history), DIM)) else: ln.append(col(" (none yet)", DIM)) ln.append('') # ── footer ──────────────────────────────────────────────────────────── ln.append(SEP) if isp.last_save_path: ln.append(col(" Last saved: %s" % isp.last_save_path, GRN)) ln.append(col(" Every keypress is a division event | BKSP=undo | Ctrl-S=save | ESC=quit", DIM)) ln.append(SEP) return "\n".join(ln) class Renderer: def draw(self, isp: KeyboardISP): # Prefix every line with \r so cursor snaps to col 1 regardless of # prior position. \033[2J clears the whole screen; \033[1;1H moves # explicitly to row 1, col 1 (more reliable than \033[H in raw mode). lines = build_screen(isp).split('\n') out = "\033[2J\033[1;1H" + "\r\n".join(lines) sys.stdout.write(out) sys.stdout.flush() # ── Keyboard input ──────────────────────────────────────────────────────────── def read_key(fd): ch = os.read(fd, 1) if ch == b'\x1b': fl = fcntl.fcntl(fd, fcntl.F_GETFL) fcntl.fcntl(fd, fcntl.F_SETFL, fl | os.O_NONBLOCK) try: rest = os.read(fd, 8) except BlockingIOError: rest = b'' finally: fcntl.fcntl(fd, fcntl.F_SETFL, fl) return '\x1b' + rest.decode('utf-8', errors='replace') return ch.decode('utf-8', errors='replace') # ── Main ────────────────────────────────────────────────────────────────────── def main(): if not sys.stdin.isatty(): print("Requires an interactive terminal.", file=sys.stderr) sys.exit(1) isp = KeyboardISP() renderer = Renderer() fd = sys.stdin.fileno() old_tty = termios.tcgetattr(fd) # Enter alt screen, hide cursor, clear, go to 1;1 — all one atomic write sys.stdout.write("\033[?1049h\033[2J\033[1;1H\033[?25l") sys.stdout.flush() tty.setraw(fd) try: renderer.draw(isp) while True: key = read_key(fd) if key in ('\x1b', '\x03', '\x04'): # ESC / Ctrl-C / Ctrl-D break elif key in ('\x7f', '\x08'): # Backspace isp.undo() elif key == '\x13': # Ctrl-S -> save if isp.gamma is not None: import time path = 'isp_model_%s.csv' % time.strftime('%Y%m%d_%H%M%S') isp.save(path) isp.last_save_path = path elif key.isprintable() or key == ' ': isp.press(key) renderer.draw(isp) except KeyboardInterrupt: pass finally: termios.tcsetattr(fd, termios.TCSADRAIN, old_tty) sys.stdout.write("\033[?25h\033[?1049l") # restore cursor then screen sys.stdout.flush() if __name__ == "__main__": main()