"""T16. Fragmenting-cipher solver: joint segmentation and substitution. Model: each ciphertext word is a concatenation of SEGMENTS (glyph strings of 1-4 units) drawn from a lexicon; each segment encodes one plaintext letter (many segments per letter), and plaintext spaces are not written. Search alternates (a) key step: with the segmentation fixed, fit segment -> letter by simulated annealing (homophonic.solve on the segment sequence), and (b) segmentation step: with the key fixed, re-segment every word by beam search to maximise the plaintext 4-gram score of the decoded letters, context carried across words. Initial segmentation: the slot-grammar fillers. Lexicon: every glyph string of 1-4 units that occurs as a slot filler or as a word prefix/suffix at least `min_count` times. Controls (plant()): Latin or vowel-less Latin enciphered by an affix-table cipher of the Naibbe kind (each letter -> prefix+suffix, k prefixes and k suffixes per letter, drawn independently), which produces one ciphertext word per plaintext letter; and the same with the prefix/suffix of consecutive letters run together into longer words (word = 2-3 letters). Success = decoded-letter accuracy on held-out words. """ import os, sys _here = os.path.dirname(os.path.abspath(__file__)) # data/analysis, or a subfolder of it sys.path.insert(0, _here if os.path.exists(os.path.join(_here, "paths.py")) else os.path.dirname(_here)) from paths import * # noqa: repository locations import collections, math, random, sys import numpy as np from corpus import tokenize_glyphs import homophonic as H from units import slots class Lexicon: def __init__(self, words, max_len=8, min_count=15, all_substrings=True): """segments: every substring of 1..max_len glyph units that occurs at least min_count times (all_substrings), plus slot fillers, word prefixes/suffixes and frequent whole words""" cnt = collections.Counter(); wc = collections.Counter(words) for w, c in wc.items(): u = tokenize_glyphs(w) for s in slots(w): cnt["".join(tokenize_glyphs(s))] += c if all_substrings: for i in range(len(u)): for L in range(1, min(max_len, len(u) - i) + 1): cnt["".join(u[i:i + L])] += c else: for L in range(1, min(4, max_len) + 1): if len(u) >= L: cnt["".join(u[:L])] += c; cnt["".join(u[-L:])] += c if c >= min_count and len(u) <= max_len: cnt[w] += c # whole words as segments (word = one unit) self.segs = {s for s, c in cnt.items() if c >= min_count and len(tokenize_glyphs(s)) <= max_len} for g in set(g for w in words for g in tokenize_glyphs(w)): self.segs.add(g) # every single glyph is a segment self.by_glyphs = {} for s in self.segs: self.by_glyphs[tuple(tokenize_glyphs(s))] = s def candidates(self, u, i): """segments starting at position i of glyph list u""" out = [] for L in range(1, 9): if i + L <= len(u): t = tuple(u[i:i + L]) if t in self.by_glyphs: out.append((L, self.by_glyphs[t])) return out def initial_segmentation(words, lex, mode="coarse"): """coarse: the whole word if it is a lexicon segment, else greedy longest-match from the left; slots: the slot-grammar fillers""" out = [] for w in words: u = tokenize_glyphs(w) if mode == "slots": segs = ["".join(tokenize_glyphs(s)) for s in slots(w)] out.append(segs if all(s in lex.segs for s in segs) else list(u)); continue segs = []; i = 0 while i < len(u): cands = lex.candidates(u, i) L, s = max(cands) if cands else (1, u[i]); segs.append(s); i += L out.append(segs) return out def resegment(words, lex, lm, key, beam=6, unk_penalty=-6.0): """beam search per word for the segmentation maximising the decoded 4-gram score; key: seg -> letter index""" L = lm.L; ctx = [0, 0, 0]; out = [] def lp(c3, letter): a, b, c = c3; return lm.logP4[((a * L + b) * L + c) * L + letter] for w in words: u = tokenize_glyphs(w) # beam of (score, pos, ctx, segs) B = [(0.0, 0, tuple(ctx), [])] best_done = None while B: nb = [] for sc, pos, c3, segs in B: if pos == len(u): if best_done is None or sc > best_done[0]: best_done = (sc, c3, segs) continue for Ln, s in lex.candidates(u, pos): if s in key: letter = key[s]; nsc = sc + lp(c3, letter); nc = (c3[1], c3[2], letter) else: nsc = sc + unk_penalty; nc = c3 nb.append((nsc, pos + Ln, nc, segs + [s])) nb.sort(key=lambda x: -x[0]); B = nb[:beam] if best_done is None: segs = [g for g in u]; else: _, ctx3, segs = best_done; ctx = list(ctx3) out.append(segs) return out def fit_key(segmented, lm, restarts=2, iters=200000, seed=0, min_count=5, lam=2.0): """lam=2: the frequency penalty that lets the solver recover planted affix-table keys (T16a); the weaker 0.3 used in chapter 23 collapses onto a few letters when the units are many""" units = [s for segs in segmented for s in segs] ciph = H.Cipher(units, min_count=min_count, keep_spaces=False) s, m, per, kl = H.solve(ciph, lm, lam=lam, restarts=restarts, iters=iters, seed=seed, verbose=False) key = {u: int(m[i]) for u, i in ciph.idx.items()} return key, per, ciph def decode(segmented, key, lm): return "".join(lm.alpha[key[s]] if s in key else "?" for segs in segmented for s in segs) def solve(words_train, words_test, lm, rounds=4, verbose=True, seed=0, iters=200000, init="coarse"): lex = Lexicon(words_train) seg_tr = initial_segmentation(words_train, lex, init); seg_te = initial_segmentation(words_test, lex, init) hist = [] for r in range(rounds): key, per, ciph = fit_key(seg_tr, lm, seed=seed + r, iters=iters) txt_te = decode(seg_te, key, lm); s_te = np.array([lm.idx[c] for c in txt_te if c in lm.idx]); sc_te = lm.score_seq(s_te) n_segs = len({s for segs in seg_tr for s in segs}) hist.append(dict(round=r, fit=per, heldout=sc_te, n_segments=n_segs, letters_used=len(set(key.values())))) if verbose: print(f" round {r}: segments {n_segs} fit {per:.3f} held-out {sc_te:.3f} letters {len(set(key.values()))} sample {txt_te[:60]}", flush=True) if r < rounds - 1: seg_tr = resegment(words_train, lex, lm, key); seg_te = resegment(words_test, lex, lm, key) return key, seg_tr, seg_te, hist # ---------------- planted controls def plant(words_plain, voy_words, k=4, letters_per_word=1, seed=0, min_count=5): """affix-table cipher: letter -> prefix+suffix (k each, independent draws); words of `letters_per_word` letters""" rng = random.Random(seed) from generators import SlotModel M = SlotModel(voy_words, seed); pre, suf = M.affixes() pre_l, pre_w = list(pre), list(pre.values()); suf_l, suf_w = list(suf), list(suf.values()) text = "".join(words_plain); letters = sorted(set(text)) table = {l: ([rng.choices(pre_l, pre_w)[0] for _ in range(k)], [rng.choices(suf_l, suf_w)[0] for _ in range(k)]) for l in letters} truth = {} # segment -> letter (may be ambiguous if the same affix serves two letters; recorded as first) for l, (P, S) in table.items(): for a in P + S: truth.setdefault(a, l) out = []; plain = []; buf = []; pbuf = [] for ch in text: P, S = table[ch]; buf.append(rng.choice(P) + rng.choice(S)); pbuf.append(ch) if len(buf) == letters_per_word: out.append("".join(buf)); plain.append("".join(pbuf)); buf = []; pbuf = [] return out, truth, plain def word_accuracy(segmented, key, lm, plain_words): """share of ciphertext words whose decoded letters equal their plaintext letters""" ok = 0 for segs, p in zip(segmented, plain_words): dec = "".join(lm.alpha[key[s]] if s in key else "?" for s in segs) ok += dec == p return ok / len(plain_words) def dict_cover(text, vocab, min_len=4, max_len=14): """share of characters of a space-less string covered by dictionary words of >= min_len letters, under the segmentation that maximises coverage (dynamic programme). A scorer not used in the search.""" n = len(text); best = [0] * (n + 1) for i in range(1, n + 1): best[i] = best[i - 1] for L in range(min_len, min(max_len, i) + 1): if text[i - L:i] in vocab and best[i - L] + L > best[i]: best[i] = best[i - L] + L return best[n] / n if n else 0.0