| """Job 187 (runs/CSEL/TWOLAYER_FULL_PREREG.md): the two-layer canvas (job 186's HIERARCHICAL model, SYLLABLE commit |
| unit) through the production selection (64 candidates per topic, Typhoon top-8 rank 0, NO refill), two arms: |
| A = 186's decode unchanged (twolayer_canvas.py decode --unit syllable) |
| B = A + the STUTTER FIX at decode (--stutter-fix): a cell may not take a token whose SOUND equals the sound of its |
| committed (or forced) left or right neighbour in the same วรรค, unless the two cells are consecutive syllables of |
| one word (the word is in the lexicon: attested as a word) or the WORD PAIR (left word, right word) is an attested |
| reduplication / same-sound join in the training grids, >= --allow-min times (default 2; a single occurrence can |
| be a typo). Default OFF: with the flag off this module's decode is byte-identical to twolayer_canvas.py decode. |
| New file; nothing shared is patched. twolayer_canvas.Decoder is replaced by StutterDecoder only inside this process and |
| only with --stutter-fix. |
| |
| allow DATA_DIR OUT.json [--min N] allowlist from DATA_DIR/poems.jsonl (twolayer_data.py build; 186's data) |
| decode --model RUN --prompts P --out O [--seed S] [--limit N] [--stutter-fix --allow A.json] (= 186's decode + flag) |
| merge D NT S0 NS D/cands/c{S0..}.jsonl -> D/cands.jsonl (+ seed); coverage / flag asserts |
| formgate D candidate form >= 98% (exit 1 = arm dropped before Typhoon) |
| picks D D/typhoon_top8.jsonl rank 0 -> D/pick.jsonl (id, prompt, poem, seed, all_pass) |
| attested TRAIN OUT.json [--min N] written syllables the training text repeats adjacently >= N x (gate metric) |
| gates OUT ATT CTL_CANDS CTL_FINAL CTL_PICK A=D B=D pre-judge gates (prereg) -> OUT/GATES.json; prints PASSING ... |
| pools OUT FB_DIR ARM=D ... RAW-text judge inputs per topic half: fixfinal, fixpick, arms (all 49 each) |
| score OUT FB_DIR ARM=D ... after judging runs/TLF_{CLEAN,ALIGN}0{h}/run1.json: paired stats + decision |
| test CPU tests of the pure parts (no model) |
| """ |
| import os, sys, json, collections, argparse, random |
| sys.path[:0] = [".", "scripts"] |
|
|
| POOL_PREFIX = "TLF" |
|
|
|
|
| def rows(p): return [json.loads(l) for l in open(p, encoding="utf-8") if l.strip()] |
|
|
|
|
| def wrows(p, R): |
| with open(p + ".tmp", "w", encoding="utf-8") as f: |
| for r in R: f.write(json.dumps(r, ensure_ascii=False) + "\n") |
| os.replace(p + ".tmp", p) |
|
|
|
|
| |
|
|
| def same_sound_pairs(grids): |
| """Counter {(left word, right word): n} over adjacent cells of one วรรค with the SAME sound that are not consecutive |
| syllables of one word occurrence. grids: iterable of วรรค, each a list of cells [word, k, n, phon].""" |
| c = collections.Counter() |
| for line in grids: |
| for a, b in zip(line, line[1:]): |
| if a[3] != b[3]: continue |
| if a[0] == b[0] and a[1] + 1 == b[1] and a[2] == b[2]: continue |
| c[(a[0], b[0])] += 1 |
| return c |
|
|
|
|
| def build_allow(data_dir, out, min_n=2): |
| c = same_sound_pairs(line for l in open(f"{data_dir}/poems.jsonl", encoding="utf-8") for line in json.loads(l)["grid"]) |
| keep = {f"{a}|{b}": n for (a, b), n in sorted(c.items()) if n >= int(min_n)} |
| json.dump({"min": int(min_n), "source": f"{data_dir}/poems.jsonl", "pairs": keep}, open(out, "w", encoding="utf-8"), |
| ensure_ascii=False, indent=0) |
| print(f" allowlist {out}: {len(keep)} of {len(c)} same-sound word pairs attested >= {min_n}x " |
| f"({sum(keep.values())} of {sum(c.values())} occurrences)") |
|
|
|
|
| def load_allow(path, X): |
| """{left word id: set(right word ids)} and the reverse, for words in the lexicon""" |
| import torch |
| d = json.load(open(path, encoding="utf-8"))["pairs"] |
| fwd, rev = collections.defaultdict(set), collections.defaultdict(set) |
| for k in d: |
| a, b = k.split("|") |
| ia, ib = X.wid.get(a), X.wid.get(b) |
| if ia is None or ib is None or ia < 4 or ib < 4: continue |
| fwd[ia].add(ib); rev[ib].add(ia) |
| t = lambda s: torch.tensor(sorted(s), dtype=torch.long) |
| return {k: t(v) for k, v in fwd.items()}, {k: t(v) for k, v in rev.items()} |
|
|
|
|
| |
|
|
| def make_decoder_class(): |
| import torch |
| import twolayer_canvas as TC |
|
|
| class StutterDecoder(TC.Decoder): |
| """TC.Decoder + the stutter ban, applied in lp() after every mask of the base decoder (so span admissibility |
| in span_scores / the tilings sees it). ALLOW = (fwd, rev) from load_allow, set on the class.""" |
| ALLOW = ({}, {}) |
| n_banned = 0 |
|
|
| def neighbour_tok(self, k): |
| if k in self.committed: return int(self.tok[k]) |
| return self.forced.get(k) |
|
|
| def ban_stutters(self, lp): |
| X = self.X; fwd, rev = self.ALLOW |
| ts, tw, tk, tn = X.tok_sound, X.tok_word, X.tok_k, X.tok_n |
| for i in range(TC.L): |
| for j in range(self.lens[i]): |
| k = self.idx(i, j) |
| if k in self.committed: continue |
| bad = None |
| if j > 0: |
| t = self.neighbour_tok(self.idx(i, j - 1)) |
| if t is not None and t >= 4 and int(ts[t]) != 0: |
| b = ts == int(ts[t]) |
| if int(tk[t]) < int(tn[t]) - 1: b[t + 1] = False |
| w = int(tw[t]) |
| if w in fwd: b &= ~torch.isin(tw, fwd[w]) |
| bad = b |
| if j + 1 < self.lens[i]: |
| t = self.neighbour_tok(self.idx(i, j + 1)) |
| if t is not None and t >= 4 and int(ts[t]) != 0: |
| b = ts == int(ts[t]) |
| if int(tk[t]) > 0: b[t - 1] = False |
| w = int(tw[t]) |
| if w in rev: b &= ~torch.isin(tw, rev[w]) |
| bad = b if bad is None else (bad | b) |
| if bad is not None: |
| self.n_banned += int((bad & (lp[k] > -1e8)).sum()) |
| lp[k] += torch.where(bad, -1e9, 0.0) |
| return lp |
|
|
| def lp(self, hold=True): |
| return self.ban_stutters(super().lp(hold)) |
|
|
| return StutterDecoder |
|
|
|
|
| def stutter_pairs(D): |
| """(line, j) of adjacent same-sound cells of a finished Decoder that are neither inside one word nor allowed""" |
| X = D.X; fwd, _ = type(D).ALLOW if hasattr(type(D), "ALLOW") else ({}, {}) |
| out = [] |
| for i in range(len(D.lens)): |
| for j in range(1, D.lens[i]): |
| a, b = int(D.tok[D.idx(i, j - 1)]), int(D.tok[D.idx(i, j)]) |
| if a < 4 or b < 4 or int(X.tok_sound[a]) != int(X.tok_sound[b]): continue |
| if b == a + 1 and int(X.tok_k[b]) == int(X.tok_k[a]) + 1: continue |
| wa, wb = int(X.tok_word[a]), int(X.tok_word[b]) |
| if wa in fwd and bool((fwd[wa] == wb).any()): continue |
| out.append((i, j)) |
| return out |
|
|
|
|
| def decode_cli(argv): |
| """= twolayer_canvas.py decode (same records, same RNG streams) + --stutter-fix; extra keys only with the flag""" |
| import torch, random as _r |
| import twolayer_canvas as TC |
| from klon8 import prosody as P |
| ap = argparse.ArgumentParser() |
| ap.add_argument("--model", required=True); ap.add_argument("--prompts", required=True); ap.add_argument("--out", required=True) |
| ap.add_argument("--seed", type=int, default=0); ap.add_argument("--temp", type=float, default=0.8); ap.add_argument("--limit", type=int, default=0) |
| ap.add_argument("--unit", choices=["syllable", "word", "wak", "bot"], default="word") |
| ap.add_argument("--refine", type=int, default=3); ap.add_argument("--topic-boost", type=float, default=1.5) |
| ap.add_argument("--stutter-fix", action="store_true", help="ban same-sound neighbours unless attested (default off)") |
| ap.add_argument("--allow", default="", help="allowlist from `allow` (required with --stutter-fix)") |
| a = ap.parse_args(argv) |
| model, X = TC.load(a.model) |
| if a.stutter_fix: |
| assert a.allow, "--stutter-fix needs --allow" |
| SD = make_decoder_class(); SD.ALLOW = load_allow(a.allow, X) |
| TC.Decoder = SD |
| rows_ = [json.loads(l) for l in open(a.prompts, encoding="utf-8")] |
| if a.limit: rows_ = rows_[:a.limit] |
| rng_len = _r.Random(a.seed) |
| with open(a.out + ".tmp", "w", encoding="utf-8") as f: |
| for n, r in enumerate(rows_): |
| text = r.get("text") or r.get("prompt") or r.get("title") or r.get("topic") |
| assert text, f"prompt row without text: {r}" |
| want = TC.pick_lens(rng_len) |
| lines, D, tries = TC.decode_one(model, X, text, unit=a.unit, temp=a.temp, lens=want, refine=a.refine, |
| topic_boost=a.topic_boost, gen=torch.Generator().manual_seed(a.seed * 100003 + n)) |
| poem = "\n".join(lines); sc = P.score(poem, n_bot_expected=1) |
| got = [len(P.g2p_line(x)) for x in lines] |
| rec = {"id": r.get("id"), "prompt": text, "sample": 0, "poem": poem, "unit": a.unit, "syl_target": want, |
| "syl": got, "retries": tries, "forward_passes": D.n_fwd, "all_pass": sc["all_pass"], |
| "structure": sc["structure"], "ext_rhyme": sc["ext_rhyme"], "tone": sc["tone"], "internal": sc["internal"]} |
| if a.stutter_fix: |
| rec["stutter_fix"] = True; rec["n_banned"] = D.n_banned; rec["stutters_left"] = len(stutter_pairs(D)) |
| f.write(json.dumps(rec, ensure_ascii=False) + "\n"); f.flush() |
| print(f"{n}/{len(rows_)} all_pass={sc['all_pass']} syl={got} want={want} retries={tries}", file=sys.stderr, flush=True) |
| os.replace(a.out + ".tmp", a.out) |
| P.save_cache() |
|
|
|
|
| |
|
|
| def merge(d, nt, s0, ns, fix=None): |
| nt, s0, ns = int(nt), int(s0), int(ns) |
| rs = [] |
| for s in range(s0, s0 + ns): |
| x = rows(f"{d}/cands/c{s}.jsonl"); assert len(x) == nt, f"{d}/cands/c{s}.jsonl: {len(x)} rows, want {nt}" |
| rs += [dict(r, seed=s) for r in x] |
| assert all(r["unit"] == "syllable" for r in rs), "not the syllable unit" |
| if fix is not None: |
| want = fix == "1" |
| assert all(bool(r.get("stutter_fix")) == want for r in rs), f"rows decoded with stutter_fix != {want}" |
| assert len({r["id"] for r in rs}) == nt and all(sum(1 for r in rs if r["id"] == i) == ns for i in {r["id"] for r in rs}) |
| wrows(f"{d}/cands.jsonl", rs) |
| print(f" {d}: {len(rs)} candidates for {nt} topics, seeds {s0}-{s0 + ns - 1}; form {100 * sum(r['all_pass'] for r in rs) / len(rs):.1f}%") |
|
|
|
|
| def picks(d, nt=49): |
| top = [r for r in rows(f"{d}/typhoon_top8.jsonl") if r["sel_rank"] == 0] |
| assert len({r["topic_id"] for r in top}) == len(top) == int(nt), "rank-0 picks: missing or duplicate topics" |
| wrows(f"{d}/pick.jsonl", [{"id": r["topic_id"], "prompt": r["prompt"], "poem": r["poem"], "seed": r["seed"], |
| "all_pass": r["all_pass"]} for r in sorted(top, key=lambda r: r["topic_id"])]) |
| print(f" {d}: {len(top)} Typhoon picks (no refill)") |
|
|
|
|
| def formgate(d, thr=98.0): |
| """pre-registered form gate, run right after merge (before Typhoon): candidate form >= 98% or the arm is dropped |
| (exit 1); writes D/FORMGATE""" |
| rs = rows(f"{d}/cands.jsonl"); f = 100 * sum(bool(r["all_pass"]) for r in rs) / len(rs) |
| ok = f >= float(thr) |
| open(f"{d}/FORMGATE", "w").write(f"{f:.2f} {'PASS' if ok else 'FAIL'}\n") |
| print(f" {d}: candidate form {f:.1f}% -> " + ("form gate PASS" if ok else "KILLED before Typhoon: form < 98%")) |
| return ok |
|
|
|
|
| def form_of(poem): |
| from klon8 import prosody as P |
| return bool(P.score(poem, n_bot_expected=1)["all_pass"]) |
|
|
|
|
| def attested_syllables(train, out, min_n=2): |
| """written syllables X (v1 segmenter, as every stutter gate) that the TRAINING TEXT repeats adjacently (XX, with ๆ |
| written out as the repeated word) >= min_n times: an independent text-level check, not the decoder's own word-pair |
| allowlist. Writes {syllable: count} for all repeats and the kept set.""" |
| from syl_align import syls_aligned as SA |
| import twolayer_data as TD |
| c = collections.Counter() |
| for l in open(train, encoding="utf-8"): |
| for ln in json.loads(l)["poem"].split("\n"): |
| s = SA("".join(TD.line_words(ln)), 40) |
| c.update(a for a, b in zip(s, s[1:]) if a == b) |
| keep = sorted(k for k, v in c.items() if v >= int(min_n)) |
| json.dump({"min": int(min_n), "source": train, "kept": keep, "counts": dict(c)}, open(out, "w", encoding="utf-8"), ensure_ascii=False) |
| print(f" attested reduplication syllables {out}: {len(keep)} of {len(c)} repeated syllables (>= {min_n}x)") |
|
|
|
|
| def stutters_unattested(poem, att): |
| import fixbase_pipe as FP |
| return FP.stutters(poem, att) |
|
|
|
|
| def gate_fail(c, f, stutter_gate): |
| """pre-registered (TWOLAYER_FULL_PREREG.md), on the 64 x 49 raw candidates: form >= 98% (every arm); UNATTESTED |
| stutters (adjacent identical written syllables, v1 segmenter, excluding syllables the training text repeats >= 2x) |
| <= control + 0.10 per poem (arm B only - arm A is reported, not killed). Raw stutters (186's metric) are reported.""" |
| dead = [] |
| if f["form"] < 98: dead.append("form < 98%") |
| if stutter_gate and f["stutters_unattested"] > c["stutters_unattested"] + 0.10: dead.append("unattested stutters") |
| return dead |
|
|
|
|
| def reported_186(c, f): |
| """186's other gates, REPORTED only (no kill here): repeats, distinct-2, rhyme-word variety, unseen, final-word freq""" |
| import twolayer_pipe as TP |
| return [x for x in TP.gate_fail(c, f) if x not in ("form < 98%", "stutters")] |
|
|
|
|
| def gates(out, att_json, ctl_cands, ctl_final, ctl_pick, *arms): |
| import twolayer_pipe as TP |
| A = dict(x.split("=", 1) for x in arms) |
| assert set(A) <= {"A", "B"} and A, "arms: A=DIR and/or B=DIR" |
| att = frozenset(json.load(open(att_json, encoding="utf-8"))["kept"]); assert len(att) > 100, "attested set too small" |
| def extra(R): |
| return {"stutters_unattested": sum(stutters_unattested(r["poem"], att) for r in R) / len(R)} |
| cand = {"CTL": rows(ctl_cands)}; cand.update({k: rows(f"{d}/cands.jsonl") for k, d in A.items()}) |
| fin = {"CTL_final": [dict(r, all_pass=r["form"]) for r in rows(ctl_final)], |
| "CTL_pick": [dict(r, all_pass=form_of(r["poem"])) for r in rows(ctl_pick)]} |
| fin.update({f"{k}_pick": rows(f"{d}/pick.jsonl") for k, d in A.items()}) |
| S = {k: dict(TP.stats(R), **extra(R)) for k, R in list(cand.items()) + list(fin.items())} |
| res = {"stats": S, "dead": {}, "reported_186": {}} |
| for n, s in S.items(): |
| print(f" {n:10s} n {s['n']:4d}, form {s['form']:.1f}%, stutters raw {s['stutters']:.3f} unattested {s['stutters_unattested']:.3f}, " |
| f"repeats {s['repeats']:.2f}, distinct-2 {s['distinct2']:.3f}, final bits {s['final_bits']:.2f}, " |
| f"final-word log-freq {s['final_word_logfreq']:.2f}, unseen {s['final_word_unseen']:.3f}") |
| for k in A: |
| res["dead"][k] = gate_fail(S["CTL"], S[k], stutter_gate=(k == "B")) |
| res["reported_186"][k] = reported_186(S["CTL"], S[k]) |
| for m in ("stutters", "stutters_unattested"): |
| if S[k][m] > S["CTL"][m] + 0.10: print(f" {k}: {m} {S[k][m]:.3f} above control + 0.10 ({S['CTL'][m]:.3f})" + (" - reported, not a kill" if k == "A" or m == "stutters" else "")) |
| print(f" {k}: " + ("KILLED before judging: " + ", ".join(res["dead"][k]) if res["dead"][k] else "pre-judge gates PASS") |
| + (f"; 186's other gates (reported): {', '.join(res['reported_186'][k])}" if res["reported_186"][k] else "; 186's other gates: all within")) |
| os.makedirs(out, exist_ok=True); json.dump(res, open(f"{out}/GATES.json", "w"), indent=1) |
| print("PASSING " + " ".join(k for k, d in res["dead"].items() if not d)) |
|
|
|
|
| def halves(ids): |
| ids = sorted(ids); h = (len(ids) + 1) // 2 |
| return [ids[:h], ids[h:]] |
|
|
|
|
| LABEL = {"A": "tl_a", "B": "tl_b"} |
|
|
|
|
| def pool_sources(fb, arms): |
| """{label: {topic: record}}: 170's final and pick, and each arm's pick; every label covers the same 49 topics""" |
| S = {"fixfinal": {r["id"]: r for r in rows(f"{fb}/final.jsonl")}, "fixpick": {r["id"]: r for r in rows(f"{fb}/pick.jsonl")}} |
| for x in arms: |
| k, d = x.split("=", 1) |
| dead = json.load(open(f"{os.path.dirname(d.rstrip('/'))}/GATES.json")).get("dead", {}).get(k) \ |
| if os.path.exists(f"{os.path.dirname(d.rstrip('/'))}/GATES.json") else None |
| if not os.path.exists(f"{d}/pick.jsonl") or dead: |
| print(f" arm {k}: killed by its gate ({dead or 'form, before Typhoon'}) - skipped"); continue |
| S[LABEL[k]] = {r["id"]: r for r in rows(f"{d}/pick.jsonl")} |
| ids = sorted(S["fixfinal"]) |
| for lab, R in S.items(): |
| assert sorted(R) == ids, f"{lab}: covers different topics" |
| assert all(R[i]["prompt"] == S["fixfinal"][i]["prompt"] for i in ids), f"{lab}: prompts differ" |
| assert all(len(R[i]["poem"].split("\n")) == 4 for i in ids), f"{lab}: poem without 4 lines" |
| return S, ids |
|
|
|
|
| def pools(out, fb, *arms): |
| """RAW text (no ๆ write-back), ALL 49 poems of every label (form-failing too; form enters the endpoint), topic halves""" |
| S, ids = pool_sources(fb, arms) |
| os.makedirs(out, exist_ok=True); n = 0 |
| for h, part in enumerate(halves(ids)): |
| for lab, R in S.items(): |
| wrows(f"{out}/jp_{lab}_h{h}.jsonl", [{"id": f"{i}_s0", "prompt": R[i]["prompt"], "poem": R[i]["poem"]} for i in part]); n += len(part) |
| same = sum(S["fixfinal"][i]["poem"] == S["fixpick"][i]["poem"] for i in ids) |
| print(f" pools: labels {list(S)}, {len(ids)} topics each, {n} system poems; fixpick == fixfinal (not refilled) for {same}") |
| return list(S) |
|
|
|
|
| def decide(allthree): |
| """pre-registered primary rule (B vs 170 final, all-three): CONTINUE if diff >= +5pp and the CI lower bound > 0; |
| STOP if the CI upper bound < +5pp; otherwise GREY""" |
| if allthree["diff"] >= 5 and allthree["ci"][0] > 0: return "CONTINUE" |
| if allthree["ci"][1] < 5: return "STOP" |
| return "GREY" |
|
|
|
|
| def score(out, fb, *arms, root_pools="runs"): |
| import fixon_raw, fixbase_pipe |
| S, ids = pool_sources(fb, arms) |
| lab, anchors = fixon_raw.labels(POOL_PREFIX, 0, 0, root_pools) |
| void = {d: v for d, v in anchors.items() if v < 0.70} |
| form = {(l, i, 0): (bool(S[l][i]["form"]) if l == "fixfinal" else form_of(S[l][i]["poem"])) for l in S for i in ids} |
| for kind in ("clean", "aligned"): |
| assert set(lab[kind]) == set(form), f"{kind}: judged poems differ from the pipeline output" |
| full = {k: {x: form[x] and lab[k][x] for x in form} for k in ("clean", "aligned")} |
| full["all-three"] = {x: full["clean"][x] and full["aligned"][x] for x in form} |
| full["form"] = form |
| R = {"anchor_clean_share": anchors, "void_pools": sorted(void), "contrasts": {}} |
| labs = [l for l in S if l.startswith("tl_")] |
| C_ = [(f"{l} vs fixfinal", "fixfinal", l) for l in labs] + [(f"{l} vs fixpick", "fixpick", l) for l in labs] + \ |
| [("fixfinal vs fixpick (refill)", "fixpick", "fixfinal")] + ([("tl_b vs tl_a (stutter fix)", "tl_a", "tl_b")] if len(labs) == 2 else []) |
| for name, a_, b_ in C_: |
| R["contrasts"][name] = {} |
| print(f" {name}") |
| for kind, L in full.items(): |
| a = {x[1:]: v for x, v in L.items() if x[0] == a_}; b = {x[1:]: v for x, v in L.items() if x[0] == b_} |
| st = R["contrasts"][name][kind] = fixbase_pipe.paired_stats(a, b) |
| print(f" {kind:9s} {a_} {st['a']:5.1f}% {b_} {st['b']:5.1f}% diff {st['diff']:+.1f}pp ({st['up']} up / {st['down']} down, " |
| f"n={st['n']}) 95% CI [{st['ci'][0]:+.1f}, {st['ci'][1]:+.1f}]") |
| same = [i for i in ids if S["fixfinal"][i]["poem"] == S["fixpick"][i]["poem"]] |
| agree = {k: sum(lab[k][("fixfinal", i, 0)] == lab[k][("fixpick", i, 0)] for i in same) for k in ("clean", "aligned")} |
| R["repeat_agreement_identical_poems"] = {"n": len(same), **agree} |
| print(f" judge repeatability on {len(same)} identical fixfinal/fixpick poems: clean {agree['clean']}/{len(same)}, aligned {agree['aligned']}/{len(same)}") |
| prim = "tl_b vs fixfinal" |
| R["decision"] = decide(R["contrasts"][prim]["all-three"]) if prim in R["contrasts"] else "B not judged (killed at the pre-judge gates)" |
| os.makedirs(out, exist_ok=True); json.dump(R, open(f"{out}/RESULT.json", "w"), indent=1, default=str) |
| print(f" anchors clean: {anchors}") |
| if void: |
| print(f" VOID pools (anchors < 70% clean): {void} - re-judge before any decision"); sys.exit(3) |
| print(f" DECISION (pre-registered, primary = B vs 170 final, all-three): {R['decision']}") |
|
|
|
|
| |
|
|
| def _test(): |
| import tempfile, torch |
| t = tempfile.mkdtemp() |
| |
| g = [[["นานา", 0, 2, "naa0"], ["นานา", 1, 2, "naa0"], ["นาน", 0, 1, "naan0"], ["นาน", 0, 1, "naan0"]], |
| [["นาน", 0, 1, "naan0"], ["นาน", 0, 1, "naan0"], ["เลย", 0, 1, "l@@j0"], ["เลย", 0, 1, "l@@j0"]], |
| [["ใจเย็น", 0, 2, "caj0"], ["ใจเย็น", 1, 2, "jen0"], ["เย็น", 0, 1, "jen0"], ["ดี", 0, 1, "dii0"]], |
| [["นาน", 0, 1, "naan0"], ["นาน", 0, 1, "naan0"], ["ใจเย็น", 0, 2, "caj0"], ["ใจเย็น", 1, 2, "jen0"]], |
| [["ใจเย็น", 0, 2, "caj0"], ["ใจเย็น", 1, 2, "jen0"], ["เย็น", 0, 1, "jen0"], ["กุ้ง", 0, 1, "kuN2"]]] |
| c = same_sound_pairs(g) |
| assert c == {("นาน", "นาน"): 3, ("เลย", "เลย"): 1, ("ใจเย็น", "เย็น"): 2}, c |
| os.makedirs(f"{t}/d"); wrows(f"{t}/d/poems.jsonl", [{"grid": g}]) |
| build_allow(f"{t}/d", f"{t}/allow.json", 2) |
| assert set(json.load(open(f"{t}/allow.json"))["pairs"]) == {"นาน|นาน", "ใจเย็น|เย็น"} |
| |
| import twolayer_canvas as TC |
| lex = {"words": ["นาน", "เลย", "ใจเย็น", "เย็น", "ดี", "นานา", "กุ้ง"], |
| "sounds": {"นาน": ["naan0"], "เลย": ["l@@j0"], "ใจเย็น": ["caj0", "jen0"], "เย็น": ["jen0"], "ดี": ["dii0"], |
| "นานา": ["naa0", "naa0"], "กุ้ง": ["kuN2"]}, "count": {}} |
| X = TC.Lex(lex); SD = make_decoder_class(); SD.ALLOW = load_allow(f"{t}/allow.json", X) |
| D = SD.__new__(SD); D.X = X; D.lens = [8] * TC.L; D.committed = set(); D.forced = {} |
| D.tok = torch.full((TC.T,), TC.MASK, dtype=torch.long); D.n_banned = 0 |
| tokof = lambda w, k=0: X.tok(w, k) |
| def put(i, j, w, k=0): kk = D.idx(i, j); D.tok[kk] = tokof(w, k); D.committed.add(kk) |
| put(0, 0, "เลย"); put(0, 2, "นาน"); put(0, 5, "ใจเย็น", 0); D.forced[D.idx(0, 6)] = tokof("ใจเย็น", 1) |
| put(1, 3, "กุ้ง") |
| lp = torch.zeros(TC.T, X.NT); D.ban_stutters(lp) |
| ok = lambda i, j, w, k=0: bool(lp[D.idx(i, j), tokof(w, k)] > -1e8) |
| assert not ok(0, 1, "เลย"), "เลย|เลย (unattested) must be banned" |
| assert ok(0, 1, "นาน"), "(0,1) sits between เลย and นาน: นาน|นาน is attested" |
| assert ok(0, 3, "นาน"), "นาน|นาน attested (right of นาน)" |
| assert ok(0, 1, "ดี") and ok(0, 3, "ดี"), "different sound never banned" |
| assert ok(0, 7, "เย็น"), "ใจเย็น|เย็น (left is a FORCED cell) attested >= 2 -> allowed" |
| assert not ok(0, 7, "ใจเย็น", 1), "a stray 2nd syllable jen0 after the forced jen0 -> banned" |
| assert ok(0, 4, "ดี") |
| assert not ok(1, 2, "กุ้ง") and not ok(1, 4, "กุ้ง"), "กุ้ง|กุ้ง unattested, both sides" |
| assert ok(1, 2, "เลย") and ok(1, 4, "นาน") |
| |
| put(2, 5, "นานา", 1) |
| lp = torch.zeros(TC.T, X.NT); D.ban_stutters(lp) |
| assert ok(2, 4, "นานา", 0), "consecutive syllables of one word are never banned" |
| assert not ok(2, 6, "นานา", 0), "a new นานา starting right after นานา's last syllable: same sound naa0 -> banned" |
| assert not ok(2, 4, "นานา", 1), "(นานา,1) at cell 4 is not the word's continuation -> banned" |
| |
| E = SD.__new__(SD); E.X = X; E.lens = [3, 3, 3, 3]; E.tok = torch.full((TC.T,), TC.MASK, dtype=torch.long) |
| for j, w in enumerate(["เลย", "เลย", "ดี"]): E.tok[E.idx(0, j)] = tokof(w) |
| for j, w in enumerate(["นาน", "นาน", "ดี"]): E.tok[E.idx(1, j)] = tokof(w) |
| E.tok[E.idx(2, 0)] = tokof("นานา", 0); E.tok[E.idx(2, 1)] = tokof("นานา", 1); E.tok[E.idx(2, 2)] = tokof("ดี") |
| for j, w in enumerate(["ดี", "กุ้ง", "กุ้ง"]): E.tok[E.idx(3, j)] = tokof(w) |
| assert stutter_pairs(E) == [(0, 1), (3, 2)], stutter_pairs(E) |
| |
| base = {"form": 99.0, "stutters": 0.06, "stutters_unattested": 0.02} |
| assert gate_fail(base, {"form": 98.4, "stutters": 0.28, "stutters_unattested": 0.20}, False) == [] |
| assert gate_fail(base, {"form": 98.4, "stutters": 0.28, "stutters_unattested": 0.13}, True) == ["unattested stutters"] |
| assert gate_fail(base, {"form": 98.4, "stutters": 0.28, "stutters_unattested": 0.11}, True) == [] |
| assert gate_fail(base, {"form": 97.9, "stutters": 0.10, "stutters_unattested": 0.0}, True) == ["form < 98%"] |
| |
| wrows(f"{t}/train.jsonl", [{"poem": "รักรักนานนาน\nรักๆเลยเลย"}]) |
| attested_syllables(f"{t}/train.jsonl", f"{t}/att.json", 2) |
| at = frozenset(json.load(open(f"{t}/att.json"))["kept"]) |
| assert "รัก" in at and "เลย" not in at, at |
| assert stutters_unattested("ฉันรักรักเธอเลยเลยนะ", at) == 1 |
| |
| assert decide({"diff": 6, "ci": (0.5, 12)}) == "CONTINUE" and decide({"diff": 6, "ci": (-1, 12)}) == "GREY" |
| assert decide({"diff": -3, "ci": (-12, 4.9)}) == "STOP" and decide({"diff": 2, "ci": (-8, 11)}) == "GREY" |
| |
| fb = f"{t}/fb"; os.makedirs(fb); os.makedirs(f"{t}/J/A"); os.makedirs(f"{t}/J/B") |
| rec = lambda i: {"id": i, "prompt": "p" + i, "poem": "ก\nข\nค\nง", "form": True} |
| for f_ in (f"{fb}/final.jsonl", f"{fb}/pick.jsonl", f"{t}/J/A/pick.jsonl"): wrows(f_, [rec("T1"), rec("T2")]) |
| json.dump({"dead": {"A": ["unattested stutters"]}}, open(f"{t}/J/GATES.json", "w")) |
| S_, _ = pool_sources(fb, [f"A={t}/J/A", f"B={t}/J/B"]); assert set(S_) == {"fixfinal", "fixpick"}, set(S_) |
| json.dump({"dead": {"A": []}}, open(f"{t}/J/GATES.json", "w")) |
| S_, _ = pool_sources(fb, [f"A={t}/J/A", f"B={t}/J/B"]); assert set(S_) == {"fixfinal", "fixpick", "tl_a"}, set(S_) |
| wrows(f"{t}/J/A/cands.jsonl", [{"all_pass": True}] * 49 + [{"all_pass": False}]); assert formgate(f"{t}/J/A") |
| wrows(f"{t}/J/B/cands.jsonl", [{"all_pass": True}] * 97 + [{"all_pass": False}] * 3); assert not formgate(f"{t}/J/B") |
| print("twolayer_full: tests pass") |
|
|
|
|
| if __name__ == "__main__": |
| cmd, args = sys.argv[1], sys.argv[2:] |
| if cmd == "decode": decode_cli(args) |
| elif cmd == "allow": build_allow(args[0], args[1], *(args[2:3] or [2])) |
| elif cmd == "attested": attested_syllables(args[0], args[1], *(args[2:3] or [2])) |
| elif cmd == "merge": merge(*args) |
| elif cmd == "picks": picks(*args) |
| elif cmd == "formgate": sys.exit(0 if formgate(*args) else 1) |
| elif cmd == "gates": gates(*args) |
| elif cmd == "pools": pools(*args) |
| elif cmd == "score": score(*args) |
| elif cmd == "test": _test() |
| else: sys.exit(f"unknown command {cmd}") |
|
|
