holobrine.bsky.social
An implementation of indivisible stochastic processes where the process is keyboard mashing
10 kB
282 lines
1#!/usr/bin/env python3
2"""
3ISP Generative Sampler
4======================
5Loads a transition matrix Gamma saved by ispKeyboard.py and generates
6a stochastic sequence characteristic of the one that produced the model.
7
8Usage
9-----
10 python3 isp_generate.py <model.csv> [options]
11
12 -n, --length N number of keys to generate (default: 100)
13 -s, --start KEY starting key (default: sample from stationary dist)
14 -t, --tempo MS delay between keys in milliseconds (default: 120)
15 -o, --output FILE write generated string to file instead of stdout
16 --no-display suppress the TUI, just print the raw string
17
18Formal basis
19------------
20At each step, the current configuration j acts as a division event:
21 p(t0) = e_j (delta spike on current key)
22 p(t) = Gamma * e_j = column j (predicted distribution over next key)
23 j' ~ p(t) (sample next key from that distribution)
24
25The generated sequence has the same bigram statistics as the original,
26and is a fresh stochastic realization of the learned indivisible law.
27"""
28
29import sys, os, time, csv, argparse, random
30import numpy as np
31
32# ── ANSI ─────────────────────────────────────────────────────────────────────
33RS = "\033[0m"
34BLD = "\033[1m"
35DIM = "\033[2m"
36CYN = "\033[96m"
37GRN = "\033[92m"
38YLW = "\033[93m"
39RED = "\033[91m"
40
41def col(text, *codes):
42 return "".join(codes) + str(text) + RS
43
44# ── Load model ────────────────────────────────────────────────────────────────
45
46def load_model(path: str):
47 """
48 Load Gamma and config space from a CSV saved by ispKeyboard.py.
49
50 Returns
51 -------
52 configs : list of str — ordered key labels
53 gamma : np.ndarray — N x N column-stochastic transition matrix
54 """
55 with open(path, newline='') as f:
56 rows = list(csv.reader(f))
57
58 if not rows:
59 raise ValueError("Empty CSV file.")
60
61 header = rows[0]
62 if header[0] != 'key':
63 raise ValueError("Expected first column header to be 'key', got %r." % header[0])
64
65 configs = header[1:]
66 N = len(configs)
67
68 if len(rows) - 1 != N:
69 raise ValueError("Expected %d data rows, got %d." % (N, len(rows) - 1))
70
71 gamma = np.zeros((N, N))
72 for i, row in enumerate(rows[1:]):
73 if row[0] != configs[i]:
74 raise ValueError(
75 "Row %d label mismatch: expected %r, got %r." % (i, configs[i], row[0]))
76 for j in range(N):
77 gamma[i, j] = float(row[j + 1])
78
79 # Sanity check
80 col_sums = gamma.sum(axis=0)
81 if not np.allclose(col_sums, 1.0, atol=1e-4):
82 raise ValueError("Columns do not sum to 1: %s" % col_sums)
83
84 return configs, gamma
85
86
87# ── Stationary distribution ───────────────────────────────────────────────────
88
89def stationary(gamma: np.ndarray) -> np.ndarray:
90 """
91 Compute the stationary distribution of Gamma by finding the eigenvector
92 for eigenvalue 1 (left eigenvector of the column-stochastic matrix).
93 Falls back to the uniform distribution if computation fails.
94 """
95 N = gamma.shape[0]
96 try:
97 vals, vecs = np.linalg.eig(gamma)
98 # Find eigenvector closest to eigenvalue 1
99 idx = np.argmin(np.abs(vals - 1.0))
100 v = np.real(vecs[:, idx])
101 v = np.abs(v)
102 s = v.sum()
103 if s > 1e-10:
104 return v / s
105 except Exception:
106 pass
107 return np.ones(N) / N
108
109
110# ── Sampler ───────────────────────────────────────────────────────────────────
111
112def sample_sequence(configs, gamma, length: int, start_idx: int) -> "list[int]":
113 """
114 Generate a sequence of config indices by repeatedly:
115 p(t) = column j of Gamma
116 j' ~ p(t)
117 """
118 N = len(configs)
119 seq = [start_idx]
120 j = start_idx
121 rng = np.random.default_rng()
122 for _ in range(length - 1):
123 p = gamma[:, j]
124 p = p / p.sum() # renormalize to guard against floating-point drift
125 j = int(rng.choice(N, p=p))
126 seq.append(j)
127 return seq
128
129
130# ── Display ───────────────────────────────────────────────────────────────────
131
132def fk(k: str) -> str:
133 return {' ': '␣', '\t': '⇥', '\r': '↵'}.get(k, k)
134
135W = 72
136
137def run_display(configs, gamma, seq: "list[int]", tempo_ms: int):
138 """
139 Animate the generation in a TUI: show the active column of Gamma
140 and the growing output string, one key at a time.
141 """
142 import tty, termios, fcntl
143
144 N = len(configs)
145 delay = tempo_ms / 1000.0
146 output = []
147
148 def build(current_idx: int, next_idx: int) -> str:
149 ln = []
150
151 ln.append(col(" ISP Generative Sampler", BLD, CYN))
152 ln.append(col("━" * W, CYN))
153 ln.append('')
154
155 # Current division event
156 ln.append(col(" Current division event", BLD))
157 ln.append(" Key j : " + col("[%s]" % fk(configs[current_idx]), CYN)
158 + col(" p(t0) = delta spike", DIM))
159 ln.append('')
160
161 # Active column of Gamma = p(t)
162 ln.append(col(" p(t) = Gamma * e_j = column j [predicted next key]", BLD))
163 p = gamma[:, current_idx]
164 bw = 30
165 order = np.argsort(p)[::-1]
166 for i in order:
167 v = p[i]
168 f = int(v * bw)
169 bar = col('█' * f, GRN) + col('░' * (bw - f), DIM)
170 picked = col(' <- sampled', YLW) if i == next_idx else ''
171 ln.append(" %6s %s %.3f%s" % (
172 col("[%s]" % fk(configs[i]), YLW), bar, v, picked))
173 ln.append('')
174
175 # Generated string so far
176 ln.append(col(" Generated output", BLD))
177 raw = ''.join(configs[i] for i in output)
178 # Wrap at W-4 chars
179 chunk = W - 4
180 lines = [raw[i:i+chunk] for i in range(0, max(len(raw), 1), chunk)]
181 for line in lines[-4:]: # show last 4 rows
182 ln.append(" " + col(line, CYN))
183 ln.append(col(" (%d keys generated)" % len(output), DIM))
184 ln.append('')
185
186 ln.append(col("─" * W, DIM))
187 ln.append(col(" Ctrl-C to stop early", DIM))
188 ln.append(col("─" * W, DIM))
189
190 return "\r\n".join(ln)
191
192 # Enter alternate screen
193 sys.stdout.write("\033[?1049h\033[2J\033[1;1H\033[?25l")
194 sys.stdout.flush()
195
196 try:
197 for step in range(len(seq) - 1):
198 current_idx = seq[step]
199 next_idx = seq[step + 1]
200 output.append(current_idx)
201
202 sys.stdout.write("\033[2J\033[1;1H" + build(current_idx, next_idx))
203 sys.stdout.flush()
204 time.sleep(delay)
205
206 # Final key
207 output.append(seq[-1])
208 sys.stdout.write("\033[2J\033[1;1H" + build(seq[-1], seq[-1]))
209 sys.stdout.flush()
210 time.sleep(1.0)
211
212 except KeyboardInterrupt:
213 pass
214 finally:
215 sys.stdout.write("\033[?25h\033[?1049l")
216 sys.stdout.flush()
217
218 return ''.join(configs[i] for i in output)
219
220
221# ── Main ──────────────────────────────────────────────────────────────────────
222
223def main():
224 parser = argparse.ArgumentParser(
225 description="Generate a stochastic sequence from an ISP model CSV.")
226 parser.add_argument('model',
227 help="Path to the CSV model file saved by ispKeyboard.py")
228 parser.add_argument('-n', '--length', type=int, default=100,
229 metavar='N', help="Number of keys to generate (default: 100)")
230 parser.add_argument('-s', '--start', type=str, default=None,
231 metavar='KEY', help="Starting key (default: sample from stationary dist)")
232 parser.add_argument('-t', '--tempo', type=int, default=120,
233 metavar='MS', help="Delay between keys in milliseconds (default: 120)")
234 parser.add_argument('-o', '--output', type=str, default=None,
235 metavar='FILE', help="Write generated string to file")
236 parser.add_argument('--no-display', action='store_true',
237 help="Skip the TUI, print raw string to stdout")
238 args = parser.parse_args()
239
240 # Load
241 try:
242 configs, gamma = load_model(args.model)
243 except (FileNotFoundError, ValueError) as e:
244 print("Error loading model: %s" % e, file=sys.stderr)
245 sys.exit(1)
246
247 N = len(configs)
248 print("Loaded model: %d keys, %d x %d matrix" % (N, N, N), file=sys.stderr)
249
250 # Determine start key
251 if args.start is not None:
252 if args.start not in configs:
253 print("Error: start key %r not in model config space %s"
254 % (args.start, configs), file=sys.stderr)
255 sys.exit(1)
256 start_idx = configs.index(args.start)
257 else:
258 stat = stationary(gamma)
259 start_idx = int(np.random.default_rng().choice(N, p=stat))
260 print("Sampled start key: %r (from stationary dist)" % configs[start_idx],
261 file=sys.stderr)
262
263 # Generate sequence
264 seq = sample_sequence(configs, gamma, args.length, start_idx)
265
266 # Output
267 if args.no_display or not sys.stdout.isatty():
268 result = ''.join(configs[i] for i in seq)
269 else:
270 result = run_display(configs, gamma, seq, args.tempo)
271
272 if args.output:
273 with open(args.output, 'w') as f:
274 f.write(result)
275 print("Written to %s" % args.output, file=sys.stderr)
276 else:
277 # Print to stdout (after TUI exits)
278 print(result)
279
280
281if __name__ == "__main__":
282 main()