#!/usr/bin/env python3 """ ISP Generative Sampler ====================== Loads a transition matrix Gamma saved by ispKeyboard.py and generates a stochastic sequence characteristic of the one that produced the model. Usage ----- python3 isp_generate.py [options] -n, --length N number of keys to generate (default: 100) -s, --start KEY starting key (default: sample from stationary dist) -t, --tempo MS delay between keys in milliseconds (default: 120) -o, --output FILE write generated string to file instead of stdout --no-display suppress the TUI, just print the raw string Formal basis ------------ At each step, the current configuration j acts as a division event: p(t0) = e_j (delta spike on current key) p(t) = Gamma * e_j = column j (predicted distribution over next key) j' ~ p(t) (sample next key from that distribution) The generated sequence has the same bigram statistics as the original, and is a fresh stochastic realization of the learned indivisible law. """ import sys, os, time, csv, argparse, random import numpy as np # ── ANSI ───────────────────────────────────────────────────────────────────── RS = "\033[0m" BLD = "\033[1m" DIM = "\033[2m" CYN = "\033[96m" GRN = "\033[92m" YLW = "\033[93m" RED = "\033[91m" def col(text, *codes): return "".join(codes) + str(text) + RS # ── Load model ──────────────────────────────────────────────────────────────── def load_model(path: str): """ Load Gamma and config space from a CSV saved by ispKeyboard.py. Returns ------- configs : list of str — ordered key labels gamma : np.ndarray — N x N column-stochastic transition matrix """ with open(path, newline='') as f: rows = list(csv.reader(f)) if not rows: raise ValueError("Empty CSV file.") header = rows[0] if header[0] != 'key': raise ValueError("Expected first column header to be 'key', got %r." % header[0]) configs = header[1:] N = len(configs) if len(rows) - 1 != N: raise ValueError("Expected %d data rows, got %d." % (N, len(rows) - 1)) gamma = np.zeros((N, N)) for i, row in enumerate(rows[1:]): if row[0] != configs[i]: raise ValueError( "Row %d label mismatch: expected %r, got %r." % (i, configs[i], row[0])) for j in range(N): gamma[i, j] = float(row[j + 1]) # Sanity check col_sums = gamma.sum(axis=0) if not np.allclose(col_sums, 1.0, atol=1e-4): raise ValueError("Columns do not sum to 1: %s" % col_sums) return configs, gamma # ── Stationary distribution ─────────────────────────────────────────────────── def stationary(gamma: np.ndarray) -> np.ndarray: """ Compute the stationary distribution of Gamma by finding the eigenvector for eigenvalue 1 (left eigenvector of the column-stochastic matrix). Falls back to the uniform distribution if computation fails. """ N = gamma.shape[0] try: vals, vecs = np.linalg.eig(gamma) # Find eigenvector closest to eigenvalue 1 idx = np.argmin(np.abs(vals - 1.0)) v = np.real(vecs[:, idx]) v = np.abs(v) s = v.sum() if s > 1e-10: return v / s except Exception: pass return np.ones(N) / N # ── Sampler ─────────────────────────────────────────────────────────────────── def sample_sequence(configs, gamma, length: int, start_idx: int) -> "list[int]": """ Generate a sequence of config indices by repeatedly: p(t) = column j of Gamma j' ~ p(t) """ N = len(configs) seq = [start_idx] j = start_idx rng = np.random.default_rng() for _ in range(length - 1): p = gamma[:, j] p = p / p.sum() # renormalize to guard against floating-point drift j = int(rng.choice(N, p=p)) seq.append(j) return seq # ── Display ─────────────────────────────────────────────────────────────────── def fk(k: str) -> str: return {' ': '␣', '\t': '⇥', '\r': '↵'}.get(k, k) W = 72 def run_display(configs, gamma, seq: "list[int]", tempo_ms: int): """ Animate the generation in a TUI: show the active column of Gamma and the growing output string, one key at a time. """ import tty, termios, fcntl N = len(configs) delay = tempo_ms / 1000.0 output = [] def build(current_idx: int, next_idx: int) -> str: ln = [] ln.append(col(" ISP Generative Sampler", BLD, CYN)) ln.append(col("━" * W, CYN)) ln.append('') # Current division event ln.append(col(" Current division event", BLD)) ln.append(" Key j : " + col("[%s]" % fk(configs[current_idx]), CYN) + col(" p(t0) = delta spike", DIM)) ln.append('') # Active column of Gamma = p(t) ln.append(col(" p(t) = Gamma * e_j = column j [predicted next key]", BLD)) p = gamma[:, current_idx] bw = 30 order = np.argsort(p)[::-1] for i in order: v = p[i] f = int(v * bw) bar = col('█' * f, GRN) + col('░' * (bw - f), DIM) picked = col(' <- sampled', YLW) if i == next_idx else '' ln.append(" %6s %s %.3f%s" % ( col("[%s]" % fk(configs[i]), YLW), bar, v, picked)) ln.append('') # Generated string so far ln.append(col(" Generated output", BLD)) raw = ''.join(configs[i] for i in output) # Wrap at W-4 chars chunk = W - 4 lines = [raw[i:i+chunk] for i in range(0, max(len(raw), 1), chunk)] for line in lines[-4:]: # show last 4 rows ln.append(" " + col(line, CYN)) ln.append(col(" (%d keys generated)" % len(output), DIM)) ln.append('') ln.append(col("─" * W, DIM)) ln.append(col(" Ctrl-C to stop early", DIM)) ln.append(col("─" * W, DIM)) return "\r\n".join(ln) # Enter alternate screen sys.stdout.write("\033[?1049h\033[2J\033[1;1H\033[?25l") sys.stdout.flush() try: for step in range(len(seq) - 1): current_idx = seq[step] next_idx = seq[step + 1] output.append(current_idx) sys.stdout.write("\033[2J\033[1;1H" + build(current_idx, next_idx)) sys.stdout.flush() time.sleep(delay) # Final key output.append(seq[-1]) sys.stdout.write("\033[2J\033[1;1H" + build(seq[-1], seq[-1])) sys.stdout.flush() time.sleep(1.0) except KeyboardInterrupt: pass finally: sys.stdout.write("\033[?25h\033[?1049l") sys.stdout.flush() return ''.join(configs[i] for i in output) # ── Main ────────────────────────────────────────────────────────────────────── def main(): parser = argparse.ArgumentParser( description="Generate a stochastic sequence from an ISP model CSV.") parser.add_argument('model', help="Path to the CSV model file saved by ispKeyboard.py") parser.add_argument('-n', '--length', type=int, default=100, metavar='N', help="Number of keys to generate (default: 100)") parser.add_argument('-s', '--start', type=str, default=None, metavar='KEY', help="Starting key (default: sample from stationary dist)") parser.add_argument('-t', '--tempo', type=int, default=120, metavar='MS', help="Delay between keys in milliseconds (default: 120)") parser.add_argument('-o', '--output', type=str, default=None, metavar='FILE', help="Write generated string to file") parser.add_argument('--no-display', action='store_true', help="Skip the TUI, print raw string to stdout") args = parser.parse_args() # Load try: configs, gamma = load_model(args.model) except (FileNotFoundError, ValueError) as e: print("Error loading model: %s" % e, file=sys.stderr) sys.exit(1) N = len(configs) print("Loaded model: %d keys, %d x %d matrix" % (N, N, N), file=sys.stderr) # Determine start key if args.start is not None: if args.start not in configs: print("Error: start key %r not in model config space %s" % (args.start, configs), file=sys.stderr) sys.exit(1) start_idx = configs.index(args.start) else: stat = stationary(gamma) start_idx = int(np.random.default_rng().choice(N, p=stat)) print("Sampled start key: %r (from stationary dist)" % configs[start_idx], file=sys.stderr) # Generate sequence seq = sample_sequence(configs, gamma, args.length, start_idx) # Output if args.no_display or not sys.stdout.isatty(): result = ''.join(configs[i] for i in seq) else: result = run_display(configs, gamma, seq, args.tempo) if args.output: with open(args.output, 'w') as f: f.write(result) print("Written to %s" % args.output, file=sys.stderr) else: # Print to stdout (after TUI exits) print(result) if __name__ == "__main__": main()