holobrine.bsky.social
An implementation of indivisible stochastic processes where the process is keyboard mashing
15 kB
382 lines
1#!/usr/bin/env python3
2"""
3Keyboard-Driven Indivisible Stochastic Process
4================================================
5Based on Jacob Barandes (arXiv:2507.21192, 2302.10778).
6
7Formal mapping
8--------------
9 C : configuration space — the set of known keys (grows on first press)
10 eⱼ : delta spike on key j — the state at a division event (we observed j)
11 Γ(t←t₀) : N×N column-stochastic transition matrix — the nomological law
12 updated from bigram history after every division event
13 p(t) : Γ · eⱼ = column j of Γ — the predicted distribution over the
14 NEXT keypress, given key j was just pressed; this is where all
15 memory of the process lives (encoded in Γ, not in p(t₀))
16
17Every keypress is a division event:
18 1. Observe key j → p(t₀) = eⱼ (delta spike, t₀ resets)
19 2. Update Γ from the new (prev→j) bigram observation
20 3. p(t) = column j of updated Γ → predicted distribution over next key
21
22Indivisibility: Γ is a primitive law between consecutive division events.
23It is NOT required to factor as a product of sub-step matrices.
24
25Controls: any printable key → division event (immediate)
26 BACKSPACE → undo last keypress
27 ESC / Ctrl-C → quit
28"""
29
30import sys, os, tty, termios, fcntl, re
31import numpy as np
32from typing import Optional
33
34# ── ANSI ─────────────────────────────────────────────────────────────────────
35RS = "\033[0m"
36BLD = "\033[1m"
37DIM = "\033[2m"
38CYN = "\033[96m"
39GRN = "\033[92m"
40YLW = "\033[93m"
41RED = "\033[91m"
42MAG = "\033[95m"
43
44def col(text, *codes):
45 return "".join(codes) + str(text) + RS
46
47_ANSI = re.compile(r'\033\[[0-9;]*m')
48def vlen(s):
49 return len(_ANSI.sub('', s))
50
51
52# ── ISP model ─────────────────────────────────────────────────────────────────
53
54class KeyboardISP:
55 """
56 Indivisible stochastic process over a keyboard configuration space.
57
58 State
59 -----
60 configs : ordered list of known keys — the configuration space C
61 gamma : N×N column-stochastic matrix — Γ(t←t₀), the nomological law
62 p : N-vector — current predicted distribution over next keypress
63 = column of Γ for the last observed key
64 = Γ · e_last (law of total probability, §2.2 eq.20)
65 last_key : the key observed at the most recent division event (t₀)
66
67 Every keypress is a division event:
68 - p(t₀) collapses to a delta spike on the observed key
69 - Γ is updated from the new bigram observation
70 - p(t) = Γ · e_j (column j of updated Γ)
71 """
72 PRIOR = 0.5 # Laplace / Krichevsky-Trofimov smoothing
73
74 def __init__(self):
75 self.configs = [] # C: ordered list of keys
76 self.index = {} # key -> matrix index
77 self.counts = np.zeros((0, 0)) # bigram counts[src_idx, dst_idx]
78 self.gamma = None # Γ: current transition matrix
79 self.p = None # predicted distribution over next key
80 self.last_key = None # key at last division event (t₀)
81 self.prev_key = None # key at division event before that
82 self.history = [] # full sequence of observed keys
83 self.n_events = 0 # total division events
84 self.last_save_path = None # most recent save path
85
86
87
88 # ── configuration space ───────────────────────────────────────────────
89
90 def _expand(self, key):
91 """Add a new key to C, growing Γ by one row and column."""
92 n = len(self.configs)
93 self.configs.append(key)
94 self.index[key] = n
95 new = np.zeros((n + 1, n + 1))
96 if n > 0:
97 new[:n, :n] = self.counts
98 self.counts = new
99
100 def _recompute_gamma(self):
101 """Recompute Γ from bigram counts with Laplace smoothing."""
102 N = len(self.configs)
103 if N == 0:
104 self.gamma = np.zeros((0, 0))
105 return
106 # counts[src, dst]: number of times dst followed src
107 # Γ[dst, src] = p(dst | src) → column src sums to 1
108 num = self.counts.T + self.PRIOR # (N, N), rows=dst cols=src
109 self.gamma = num / num.sum(axis=0, keepdims=True)
110
111 # ── division event ────────────────────────────────────────────────────
112
113 def press(self, key) -> dict:
114 """
115 Process a keypress as a division event.
116
117 Returns a summary dict for the renderer.
118 """
119 self.n_events += 1
120 new_key = (key not in self.index)
121
122 # Expand C if needed
123 if new_key:
124 self._expand(key)
125
126 # Update bigram counts: (prev → key)
127 if self.last_key is not None:
128 src = self.index[self.last_key]
129 dst = self.index[key]
130 self.counts[src, dst] += 1
131
132 # Recompute Γ from updated counts
133 prev_gamma = self.gamma
134 self._recompute_gamma()
135
136 # p(t) = Γ · eⱼ = column j of Γ (law of total probability)
137 j = self.index[key]
138 self.p = self.gamma[:, j].copy()
139
140 self.history.append(key)
141 self.prev_key = self.last_key
142 self.last_key = key
143
144 return {
145 "key": key,
146 "new_key": new_key,
147 "n_events": self.n_events,
148 "j": j,
149 "p": self.p.copy(),
150 "gamma": self.gamma.copy(),
151 "configs": list(self.configs),
152 "history": list(self.history),
153 }
154
155 def save(self, path: str):
156 """
157 Save Gamma and the config space to a CSV file.
158
159 Format:
160 - First row: header = "key" followed by config labels (the column keys)
161 - Each subsequent row: row-key label, then Gamma[i, j] values
162 - Config order matches the matrix indices
163 """
164 import csv
165 with open(path, 'w', newline='') as f:
166 writer = csv.writer(f)
167 writer.writerow(['key'] + [k for k in self.configs])
168 for i, ci in enumerate(self.configs):
169 writer.writerow([ci] + ['%.6f' % self.gamma[i, j]
170 for j in range(len(self.configs))])
171
172 def undo(self):
173 """Remove the last keypress and recompute state."""
174 if not self.history:
175 return
176 self.history.pop()
177 self.n_events -= 1
178
179 # Rebuild from scratch
180 self.configs = []
181 self.index = {}
182 self.counts = np.zeros((0, 0))
183 self.gamma = None
184 self.p = None
185 self.last_key = None
186 self.prev_key = None
187 self.ck_violation = None
188
189 saved = list(self.history)
190 self.history = []
191 for k in saved:
192 self.press(k)
193
194
195# ── Screen renderer ───────────────────────────────────────────────────────────
196
197def fk(k):
198 """Human-readable key label."""
199 specials = {' ': 'SPC', '\t': 'TAB', '\r': 'RET'}
200 if k in specials:
201 return specials[k]
202 if not k.isprintable():
203 return '0x%02x' % ord(k)
204 return k
205
206W = 72 # fixed content width — never reaches terminal edge
207
208def build_screen(isp: KeyboardISP) -> str:
209 SEP = col('─' * W, DIM)
210 TOP = col('━' * W, CYN)
211 ln = []
212
213 # ── header ────────────────────────────────────────────────────────────
214 ln.append(col(" Keyboard Indivisible Stochastic Process", BLD, CYN))
215 ln.append(TOP)
216 ln.append('')
217
218 # ── configuration space ───────────────────────────────────────────────
219 cfgs = isp.configs
220 ln.append(col(" Configuration space C (%d keys)" % len(cfgs), BLD))
221 if cfgs:
222 ln.append(" " + " ".join(
223 col("[%s]" % fk(k), YLW) + col("=%d" % i, DIM)
224 for i, k in enumerate(cfgs)))
225 else:
226 ln.append(col(" (none yet — press any key to begin)", DIM))
227 ln.append('')
228
229 # ── last division event ───────────────────────────────────────────────
230 ln.append(col(" Last division event", BLD))
231 if isp.last_key is not None:
232 ln.append(" Observed key : " + col("[%s]" % fk(isp.last_key), CYN)
233 + col(" (p(t₀) = delta spike on this key)", DIM))
234 if isp.prev_key is not None:
235 ln.append(" Previous key : " + col("[%s]" % fk(isp.prev_key), YLW))
236 else:
237 ln.append(col(" (no keypresses yet)", DIM))
238 ln.append(" Division events so far : %d" % isp.n_events)
239 ln.append('')
240
241 # ── transition matrix Γ ───────────────────────────────────────────────
242 ln.append(col(" Transition matrix Gamma [Gamma_ij = p(i next | j just pressed)]", BLD))
243 if isp.gamma is not None and cfgs:
244 N = len(cfgs)
245 cw = 7
246 # Column headers (source keys)
247 hdr = " " + "".join(("%*s" % (cw, fk(k))) for k in cfgs)
248 ln.append(col(hdr, YLW))
249 for i, ci in enumerate(cfgs):
250 # Highlight the column of the last pressed key
251 row = col(" %4s " % fk(ci), YLW)
252 for j in range(N):
253 v = isp.gamma[i, j]
254 # Highlight selected column (last key pressed)
255 if isp.last_key is not None and j == isp.index[isp.last_key]:
256 clr = GRN if v > 0.3 else (YLW if v > 0.1 else DIM)
257 row += col("%*.3f" % (cw, v), clr, BLD)
258 else:
259 clr = GRN if v > 0.5 else (YLW if v > 0.2 else DIM)
260 row += col("%*.3f" % (cw, v), clr)
261 ln.append(row)
262 if isp.last_key is not None:
263 ln.append(col(" (highlighted column = Gamma·e_j for last key)", DIM))
264 else:
265 ln.append(col(" (awaiting first keypress)", DIM))
266 ln.append('')
267
268 # ── predicted distribution p(t) ───────────────────────────────────────
269 ln.append(col(" p(t) = Gamma · e_j = predicted distribution over NEXT key", BLD))
270 if isp.p is not None and cfgs:
271 bw = 30
272 # Sort by probability descending for readability
273 order = np.argsort(isp.p)[::-1]
274 for i in order:
275 v = isp.p[i]
276 f = int(v * bw)
277 bar = col('█' * f, GRN) + col('░' * (bw - f), DIM)
278 # Mark if this is the last observed key
279 marker = col(' <- last pressed', DIM) if cfgs[i] == isp.last_key else ''
280 ln.append(" %6s %s %.3f%s" % (
281 col("[%s]" % fk(cfgs[i]), YLW), bar, v, marker))
282 else:
283 ln.append(col(" (awaiting first keypress)", DIM))
284 ln.append('')
285
286 # ── history ───────────────────────────────────────────────────────────
287 ln.append(col(" Key history", BLD))
288 if isp.history:
289 # Show last 40 keys
290 recent = isp.history[-40:]
291 ln.append(" " + " ".join(col("[%s]" % fk(k), CYN) for k in recent))
292 if len(isp.history) > 40:
293 ln.append(col(" ... (%d total)" % len(isp.history), DIM))
294 else:
295 ln.append(col(" (none yet)", DIM))
296 ln.append('')
297
298 # ── footer ────────────────────────────────────────────────────────────
299 ln.append(SEP)
300 if isp.last_save_path:
301 ln.append(col(" Last saved: %s" % isp.last_save_path, GRN))
302 ln.append(col(" Every keypress is a division event | BKSP=undo | Ctrl-S=save | ESC=quit", DIM))
303 ln.append(SEP)
304
305 return "\n".join(ln)
306
307
308class Renderer:
309 def draw(self, isp: KeyboardISP):
310 # Prefix every line with \r so cursor snaps to col 1 regardless of
311 # prior position. \033[2J clears the whole screen; \033[1;1H moves
312 # explicitly to row 1, col 1 (more reliable than \033[H in raw mode).
313 lines = build_screen(isp).split('\n')
314 out = "\033[2J\033[1;1H" + "\r\n".join(lines)
315 sys.stdout.write(out)
316 sys.stdout.flush()
317
318
319# ── Keyboard input ────────────────────────────────────────────────────────────
320
321def read_key(fd):
322 ch = os.read(fd, 1)
323 if ch == b'\x1b':
324 fl = fcntl.fcntl(fd, fcntl.F_GETFL)
325 fcntl.fcntl(fd, fcntl.F_SETFL, fl | os.O_NONBLOCK)
326 try:
327 rest = os.read(fd, 8)
328 except BlockingIOError:
329 rest = b''
330 finally:
331 fcntl.fcntl(fd, fcntl.F_SETFL, fl)
332 return '\x1b' + rest.decode('utf-8', errors='replace')
333 return ch.decode('utf-8', errors='replace')
334
335
336# ── Main ──────────────────────────────────────────────────────────────────────
337
338def main():
339 if not sys.stdin.isatty():
340 print("Requires an interactive terminal.", file=sys.stderr)
341 sys.exit(1)
342
343 isp = KeyboardISP()
344 renderer = Renderer()
345 fd = sys.stdin.fileno()
346 old_tty = termios.tcgetattr(fd)
347
348 # Enter alt screen, hide cursor, clear, go to 1;1 — all one atomic write
349 sys.stdout.write("\033[?1049h\033[2J\033[1;1H\033[?25l")
350 sys.stdout.flush()
351 tty.setraw(fd)
352
353 try:
354 renderer.draw(isp)
355 while True:
356 key = read_key(fd)
357
358 if key in ('\x1b', '\x03', '\x04'): # ESC / Ctrl-C / Ctrl-D
359 break
360 elif key in ('\x7f', '\x08'): # Backspace
361 isp.undo()
362 elif key == '\x13': # Ctrl-S -> save
363 if isp.gamma is not None:
364 import time
365 path = 'isp_model_%s.csv' % time.strftime('%Y%m%d_%H%M%S')
366 isp.save(path)
367 isp.last_save_path = path
368 elif key.isprintable() or key == ' ':
369 isp.press(key)
370
371 renderer.draw(isp)
372
373 except KeyboardInterrupt:
374 pass
375 finally:
376 termios.tcsetattr(fd, termios.TCSADRAIN, old_tty)
377 sys.stdout.write("\033[?25h\033[?1049l") # restore cursor then screen
378 sys.stdout.flush()
379
380
381if __name__ == "__main__":
382 main()