Download compile.py from Pranavz/pylexnet: direct link, hf CLI and curl.
- Browser
- Download file 43.2 kB
-
https://huggingface.co/Pranavz/pylexnet/resolve/main/compile.py
- Command line
-
hf download hf://Pranavz/pylexnet/compile.py
-
curl -L -o compile.py https://huggingface.co/Pranavz/pylexnet/resolve/main/compile.py
43.2 kB
| """pylexnet 'hand' lane: a Python syntax-highlighter network designed unit by unit, lexnet style. | |
| No automaton is compiled. Every unit below does one designed job: ~60 booleans for strings, comments, | |
| numbers, words and contexts, a length counter, two hash-gate registers, a sorted threshold family on the | |
| word hash, and one "distance to the nearest vocabulary word of category c" unit per category. | |
| Activation is clip(x, 0, hi) everywhere. | |
| Two vocabulary matchers, chosen by --matcher: | |
| add (default) additive retrieval: per-position value tables (positions 0..P-1 plus a constant each, a tail | |
| table, three last-byte tables) solved so that every vocabulary word's sum of values is its class target | |
| t*2^30 (t = 1 keyword, 2 def, 3 class, 4 builtin, 5 magic), rounded to integers (window +-64). A one-hot | |
| position chain gates the values into an accumulator; three delayed registers add the last bytes' values at | |
| the word's end; nine threshold units read the class windows. ~1 number per word. Not exact: a non-vocabulary | |
| word whose sum lands in a window is a false positive (fpcheck.py measures them on the fit split; pinning | |
| words to 0 via zero_words.txt is possible when the system stays solvable). | |
| hash exact: a polynomial hash h over the word, a sorted tied threshold family ge_k = clip1(h - T_k + 1) (the rank | |
| of h), and per category E_c = clip1(h - sum_k ge_k * D_c[k]) with D_c telescoping to the largest category-c | |
| threshold <= h, so E_c == 0 exactly when h is a category-c word. 2 numbers per word. | |
| Writes pylexnet_weights.bin (PYLX02) and pylexnet_kernel.h (architecture only, with a structure hash). | |
| usage: compile.py [--matcher add|hash] [--vocab N] [--addP P] [--sim TEXT] N = builtin/magic words kept, by fit frequency. | |
| """ | |
| import keyword | |
| import re | |
| import struct | |
| import sys | |
| from pathlib import Path | |
| import numpy as np | |
| HERE = Path(__file__).resolve().parent | |
| # ---------------------------------------------------------------- byte features (the embedding table) | |
| FEATS = ["digit", "nl", "dq", "sq", "bs", "hash", "dot", "at", "lbr", "rbr", | |
| "obr", "cbr", "fcch", "code", "pf", "po", "eE", "pm"] | |
| MATCHER = "add" # "add": additive retrieval (IDEAS.md idea 1, default). "hash": sorted threshold family on a polynomial hash | |
| ADD_P, ADD_L = 7, 3 # additive matcher: value tables (+ a constant) for positions 0..P-1, a tail table, 3 last-byte tables (L is fixed at 3) | |
| ADD_EPS = 64 # window half-width around each class target | |
| ADD_T = 2 ** 30 # class targets are t * ADD_T, t = 1 keyword, 2 def, 3 class, 4 builtin, 5 magic | |
| if "--matcher" in sys.argv: | |
| MATCHER = sys.argv[sys.argv.index("--matcher") + 1] | |
| if "--addP" in sys.argv: | |
| ADD_P = int(sys.argv[sys.argv.index("--addP") + 1]) | |
| if "--addL" in sys.argv: | |
| ADD_L = int(sys.argv[sys.argv.index("--addL") + 1]) | |
| if MATCHER == "add": | |
| FEATS += [f"v{p}" for p in range(ADD_P)] + ["vt"] + [f"e{j}" for j in range(1, ADD_L + 1)] | |
| NF = len(FEATS) | |
| FI = {f: i for i, f in enumerate(FEATS)} | |
| ADD_TABLES = {} # feature name -> {code: value}, filled by solve_additive | |
| def code_of(b): | |
| """Word-character code for the hash: a-z 1..26, A-Z 27..52, _ 53, digits 54, bytes >= 127 55, else 0.""" | |
| if 97 <= b <= 122: | |
| return b - 96 | |
| if 65 <= b <= 90: | |
| return b - 64 + 26 | |
| if b == 95: | |
| return 53 | |
| if 48 <= b <= 57: | |
| return 54 | |
| if b >= 127: | |
| return 55 | |
| return 0 | |
| BASE = 56 # hash base: one more than the largest code | |
| WIN = 7 # exact polynomial over the first WIN chars; later chars add code + 8*position | |
| HMAX = 2 ** 41 # ceiling of the hash registers (> BASE**WIN + tail) | |
| KG = 2 ** 42 # gate constant for the hash units: larger than any hash value | |
| def emb_table(): | |
| E = np.zeros((128, NF)) | |
| for b in range(128): | |
| r = E[b] | |
| c = code_of(b) | |
| isalpha = (65 <= b <= 90) or (97 <= b <= 122) or b == 95 or b == 127 | |
| isdigit = 48 <= b <= 57 | |
| r[FI["digit"]] = isdigit | |
| r[FI["nl"]] = b == 10 | |
| r[FI["dq"]] = b == 34 | |
| r[FI["sq"]] = b == 39 | |
| r[FI["bs"]] = b == 92 | |
| r[FI["hash"]] = b == 35 | |
| r[FI["dot"]] = b == 46 | |
| r[FI["at"]] = b == 64 | |
| r[FI["lbr"]] = b == 123 | |
| r[FI["rbr"]] = b == 125 | |
| r[FI["obr"]] = b in (40, 91, 123) | |
| r[FI["cbr"]] = b in (41, 93, 125) | |
| r[FI["fcch"]] = b in (125, 58, 33) # } : ! end a replacement field at depth 0 | |
| r[FI["code"]] = c | |
| r[FI["pf"]] = b in (102, 70) # f F | |
| r[FI["eE"]] = b in (101, 69) | |
| r[FI["pm"]] = b in (43, 45) | |
| r[FI["po"]] = b in (114, 82, 98, 66, 117, 85) # r R b B u U | |
| for f, tab in ADD_TABLES.items(): | |
| r[FI[f]] = tab.get(c, 0) if c else 0 | |
| return E | |
| def word_feats(w): | |
| """Additive matcher: the (table, code) terms of a word.""" | |
| cs = [code_of(b) for b in w.encode()] | |
| out = {} | |
| for i, c in enumerate(cs): | |
| k = (f"v{i}" if i < ADD_P else "vt", c) | |
| out[k] = out.get(k, 0) + 1 | |
| k = (k[0], 0) # the position's constant (the gate unit's bias) | |
| out[k] = out.get(k, 0) + 1 | |
| for j in range(1, ADD_L + 1): | |
| if len(cs) >= j: | |
| k = (f"e{j}", cs[-j]) | |
| out[k] = out.get(k, 0) + 1 | |
| return out | |
| def solve_additive(vocab): | |
| """Integer value tables such that every vocabulary word's sum lands on its class target (within ADD_EPS). | |
| Orthogonal matching pursuit picks a sparse support, least squares solves it, rounding keeps it integer.""" | |
| words = sorted(vocab) | |
| tgt = {"KW": 1, "BI": 4, "MAG": 5} | |
| zero = [w for w in (HERE / "zero_words.txt").read_text().split() if w not in vocab] if (HERE / "zero_words.txt").exists() else [] | |
| T = np.array([ADD_T * (2 if w == "def" else 3 if w == "class" else tgt[vocab[w]]) for w in words] + [0.0] * len(zero)) | |
| words = words + zero # frequent non-vocabulary words pinned to 0 (chosen on the fit split, see zero_words.txt) | |
| cols = {} | |
| rows = [word_feats(w) for w in words] | |
| for r in rows: | |
| for k in r: | |
| cols.setdefault(k, len(cols)) | |
| A = np.zeros((len(words), len(cols))) | |
| for i, r in enumerate(rows): | |
| for k, n in r.items(): | |
| A[i, cols[k]] = n | |
| S, x = [], np.zeros(len(cols)) | |
| for _ in range(len(cols)): | |
| res = T - A @ x | |
| if np.abs(res).max() < 1e-3: | |
| break | |
| corr = A.T @ res | |
| corr[S] = 0 | |
| j = int(np.argmax(np.abs(corr))) | |
| if corr[j] == 0: | |
| break | |
| S.append(j) | |
| xs, *_ = np.linalg.lstsq(A[:, S], T, rcond=None) | |
| x = np.zeros(len(cols)); x[S] = xs | |
| x = np.round(x) | |
| res = A @ x - T | |
| unsolved = int((np.abs(res) > ADD_EPS).sum()) | |
| for (f, c), j in cols.items(): | |
| if x[j] != 0: | |
| ADD_TABLES.setdefault(f, {})[c] = int(x[j]) | |
| nz = int((x != 0).sum()) | |
| vmax = max(abs(v) for tab in ADD_TABLES.values() for v in tab.values()) | |
| print(f"additive matcher: {len(words) - len(zero)} words + {len(zero)} zero-pinned, {len(cols)} candidate values, support {nz}, max |value| {vmax}, " | |
| f"unsolved {unsolved}, max residual {np.abs(res).max():g}") | |
| assert unsolved == 0, "additive system not solvable with these features" | |
| return vmax | |
| def word_hash(w): | |
| h = 0 | |
| for i, ch in enumerate(w.encode()): | |
| c = code_of(ch) | |
| h = BASE * h + c if i < WIN else h + c + 8 * i | |
| return h | |
| # ---------------------------------------------------------------- vocabulary | |
| import pygments.lexers.python as _pl | |
| _src = Path(_pl.__file__).read_text() | |
| _b = _src.index("'builtins': [") | |
| _a = _src.index("'magicfuncs': [") | |
| _v = _src.index("'magicvars': [") | |
| BUILTINS = set(re.findall(r"'(\w+)'", _src[_b:_a])) - {"builtins", "exceptions"} | {"self", "cls", "Ellipsis", "NotImplemented"} | |
| MAGIC = set(re.findall(r"'(__\w+__)'", _src[_a:_v])) - BUILTINS | |
| KEYWORDS = set(keyword.kwlist) | |
| # frequency order measured once on accuracy2's fit split (vocab_prio.txt); words below its cut-off come last, alphabetically | |
| PRIO = [w for w, _ in eval("[" + (HERE / "vocab_prio.txt").read_text() + "]")] | |
| def pick_vocab(limit): | |
| """Keywords always; builtins and magic words by fit-corpus frequency up to `limit`.""" | |
| ranked = [w for w in PRIO if w in BUILTINS or w in MAGIC] | |
| ranked += sorted((BUILTINS | MAGIC) - set(ranked)) | |
| chosen = ranked if limit is None else ranked[:limit] | |
| return {w: ("BI" if w in BUILTINS else "MAG") for w in chosen} | {w: "KW" for w in KEYWORDS} | |
| # ---------------------------------------------------------------- the network, unit by unit | |
| class Unit: | |
| def __init__(self, name, ins, b=0.0, hi=1.0, guard=None, fam=None): | |
| self.name, self.ins, self.b, self.hi, self.guard, self.fam = name, ins, float(b), float(hi), guard, fam | |
| UNITS = [] | |
| def u(name, b=0.0, hi=1.0, guard=None, fam=None, **ins): | |
| """ins: ref -> weight; refs written as x_feat (byte feature), p_name (previous state) or a unit name.""" | |
| d = {} | |
| for k, w in ins.items(): | |
| ref = "x." + k[2:] if k.startswith("x_") else "p." + k[2:] if k.startswith("p_") else "u." + k | |
| d[ref] = float(w) | |
| UNITS.append(Unit(name, d, b, hi, guard, fam)) | |
| return name | |
| STATE = ["S", "sq1", "T", "F", "E", "c", "fo", "FS", "FD", "oq1", "oT", "C", "N", "ee", "W", "d", "pf", "fp", | |
| "cD", "cC", "cAt", "atp", "cDot"] | |
| STATE += ["h1"] if MATCHER == "hash" else [f"pos{p}" for p in range(ADD_P)] + ["tail", "A", "e1", "e2", "e2d", "e3", "e3d", "e3dd"] | |
| STATE_UNIT = {s: s + "_new" for s in STATE} | |
| for _p in range(ADD_P): | |
| STATE_UNIT[f"pos{_p}"] = f"pos{_p}" | |
| STATE_UNIT["tail"] = "tail" | |
| CATS = ["KW", "BI", "MAG"] | |
| def build(vocab): | |
| UNITS.clear() | |
| # ---- layer 0: from the byte and the previous state only | |
| u("wordch", guard=["x.code"], x_code=1) # word character (any code > 0) | |
| u("qmd", guard=["x.dq"], x_dq=1, p_sq1=-1, p_E=-1) # unescaped matching " in a "-string | |
| u("qms", guard=["x.sq"], x_sq=1, p_sq1=1, p_E=-1, b=-1) # unescaped matching ' in a '-string | |
| u("c2", guard=["p.c"], p_c=1, b=-1) # quote run >= 2 | |
| if MATCHER == "hash": | |
| u("lt7", p_d=-1, b=WIN) # fewer than WIN chars of the word seen | |
| u("wend", guard=["p.W"], p_W=1, x_code=-1) # a word ended just before this byte | |
| u("fieldopen", guard=["p.fo"], p_fo=1, x_lbr=-1) # { was a field start: this byte is code | |
| u("pf_new", guard=["x.po", "x.pf"], x_po=3, x_pf=3, p_pf=2, p_W=-1, p_d=-1, b=-2) # word so far could be a string prefix | |
| # ---- strings. S_eff: this byte is inside a string (p.S unless "" just closed or a field just opened) | |
| u("S_eff", guard=["p.S"], p_S=4, fieldopen=-4, p_T=1, qmd=1, qms=1, c2=-1, b=-3) | |
| u("fp_new", guard=["x.pf", "p.fp"], x_pf=1, p_fp=1, wordch=1, b=-1) # word contains f | |
| u("sclose", guard=["u.S_eff"], S_eff=2, p_T=-1, qmd=1, qms=1, p_c=-1, b=-2) | |
| u("tclose", guard=["u.S_eff"], S_eff=3, p_T=1, qmd=1, qms=1, p_c=1, b=-6) | |
| u("tripleopen", guard=["u.S_eff"], S_eff=3, p_T=-1, qmd=1, qms=1, p_c=1, b=-5) | |
| u("nlclose", guard=["u.S_eff"], S_eff=1, p_T=-1, x_nl=1, p_E=-1, b=-1) | |
| u("E_new", guard=["u.S_eff"], S_eff=1, x_bs=1, p_E=-1, b=-1) | |
| u("fo_new", guard=["u.S_eff"], S_eff=1, p_F=1, x_lbr=1, p_E=-1, p_fo=-1, b=-2) | |
| u("C_new", guard=["x.hash", "p.C"], x_hash=1, S_eff=-1, p_C=1, x_nl=-1) | |
| u("open", guard=["x.dq", "x.sq"], x_dq=1, x_sq=1, S_eff=-1, p_C=-1, p_N=-1) | |
| u("open_f", guard=["x.dq", "x.sq"], x_dq=1, x_sq=1, S_eff=-1, p_C=-1, p_N=-1, p_pf=1, p_fp=1, b=-2) | |
| u("open_qs", guard=["x.sq"], x_dq=1, x_sq=2, S_eff=-1, p_C=-1, p_N=-1, b=-1) | |
| u("fieldclose", guard=["p.FS"], p_FS=1, S_eff=-1, p_C=-1, p_FD=-1, x_fcch=1, b=-1) | |
| u("inc", guard=["p.FS"], p_FS=1, S_eff=-1, p_C=-1, x_obr=1, b=-1) | |
| u("dec", guard=["p.FS"], p_FS=4, S_eff=-4, p_C=-4, x_cbr=1, p_FD=1, b=-5) | |
| u("S_new", guard=["u.S_eff", "u.open", "u.fieldclose"], S_eff=1, sclose=-1, tclose=-1, nlclose=-1, open=1, fieldclose=1) | |
| u("c_new", hi=3, guard=["u.S_eff", "u.open", "u.open_qs"], p_c=1, S_eff=10, qmd=5, qms=5, tripleopen=-5, tclose=-5, sclose=-5, | |
| open=10, open_qs=5, b=-14) | |
| u("T_new", guard=["p.T", "u.tripleopen", "u.fieldclose"], p_T=2, tripleopen=2, tclose=-2, open=-2, fieldclose=1, p_oT=1, b=-1) | |
| u("F_new", guard=["p.F", "u.open_f", "u.fieldclose"], p_F=1, open_f=2, open=-1, sclose=-1, tclose=-1, nlclose=-1, fieldclose=1) | |
| u("sq1_new", guard=["p.sq1", "u.open_qs", "u.S_eff", "u.fieldclose"], p_sq1=2, open_qs=4, sclose=-2, tclose=-2, nlclose=-2, | |
| p_S=-2, S_eff=2, fieldclose=1, p_oq1=1, b=-1) # quote type; cleared when the string dies (unless a nested one opens); restored from oq1 | |
| u("FS_new", guard=["p.FS", "u.fieldopen"], p_FS=1, fieldopen=1, fieldclose=-1) | |
| u("FD_new", hi=3, guard=["p.FD", "u.inc"], p_FD=1, inc=1, dec=-1) | |
| u("oq1_new", guard=["p.oq1", "u.fieldopen"], p_oq1=2, fieldopen=1, p_sq1=1, fieldclose=-2, b=-1) # outer quote type, saved at a field's start | |
| u("oT_new", guard=["p.oT", "u.fieldopen"], p_oT=2, fieldopen=1, p_T=1, fieldclose=-2, b=-1) | |
| u("lab_STR", guard=["u.S_eff", "u.open", "u.fieldclose"], S_eff=1, nlclose=-1, open=1, fieldclose=1) | |
| # ---- numbers and words | |
| u("N_new", guard=["x.digit", "p.N"], x_digit=6, p_W=-3, p_N=5, wordch=2, x_dot=2, x_pm=1, p_ee=1, S_eff=-2, p_C=-2, b=-6) | |
| u("ee_new", guard=["u.N_new"], x_eE=1, N_new=1, b=-1) # an exponent e: a sign may follow | |
| u("W_new", guard=["u.wordch", "p.W"], wordch=2, x_digit=-1, p_W=1, p_N=-2, S_eff=-2, p_C=-2, b=-1) | |
| u("d_new", hi=30, guard=["u.W_new"], p_d=1, W_new=40, b=-39) | |
| if MATCHER == "hash": | |
| u("gA", hi=HMAX, guard=["p.W"], p_h1=1, p_W=KG, wordch=KG, lt7=KG, b=-3 * KG) # old hash, still in the window | |
| u("gB", hi=HMAX, guard=["p.W"], p_h1=1, p_W=KG, wordch=KG, lt7=-KG, b=-2 * KG) # old hash, past the window | |
| u("cA", hi=HMAX, guard=["u.W_new"], x_code=1, W_new=KG, lt7=KG, b=-2 * KG) # this char's code, in window | |
| u("cB", hi=HMAX, guard=["u.W_new"], x_code=1, p_d=8, W_new=KG, lt7=-KG, b=-KG) # past the window: code + 8*position | |
| u("h1_new", hi=HMAX, gA=BASE, gB=1, cA=1, cB=1) | |
| else: | |
| build_additive(vocab) | |
| if MATCHER == "add": | |
| u("is_def", guard=["u.wend"], wend=1, ge2=1, le2=1, b=-2) | |
| u("is_class", guard=["u.wend"], wend=1, ge3=1, le3=1, b=-2) | |
| u("cD_new", guard=["p.cD", "u.is_def"], p_cD=1, is_def=2, wend=-1, x_nl=-1) | |
| u("cC_new", guard=["p.cC", "u.is_class"], p_cC=1, is_class=2, wend=-1, x_nl=-1) | |
| u("atp_new", guard=["x.at"], x_at=1, S_eff=-1, p_C=-1) | |
| u("cAt_new", guard=["p.cAt", "p.atp"], p_cAt=1, p_atp=1, wordch=1, wend=-1, b=-1) | |
| u("cDot_new", guard=["x.dot", "p.cDot"], x_dot=4, S_eff=-4, p_C=-4, p_N=-4, p_cDot=2, wordch=1, wend=-1, b=-2) | |
| u("ev_DEC", guard=["u.wend"], wend=1, p_cAt=1, b=-1) | |
| u("ev_CL", guard=["u.wend"], wend=1, p_cC=1, p_cAt=-1, b=-1) | |
| # keyword targets 1, 2 (def), 3 (class): a value in one of the three windows scores 4 on the sorted sum, 3 otherwise | |
| u("ev_KW", guard=["u.wend"], wend=3, ge1=1, le1=1, ge2=1, le2=1, ge3=1, le3=1, p_cAt=-1, p_cC=-1, p_cD=-1, b=-6) | |
| u("ev_FN", guard=["u.wend"], wend=3, ge5=1, le5=1, p_cD=2, p_cAt=-2, p_cC=-2, b=-4) # after def, or a magic method name | |
| u("ev_BI", guard=["u.wend"], wend=1, ge4=1, le4=1, p_cAt=-1, p_cC=-1, p_cD=-1, p_cDot=-1, b=-2) | |
| u("ev_STR", guard=["u.wend"], wend=1, open=1, p_pf=1, b=-2) | |
| u("lab_W", guard=["u.W_new", "u.atp_new"], W_new=1, atp_new=1) | |
| return [], [] | |
| # ---- vocabulary: sorted thresholds (one tied family), one distance unit per category | |
| byhash = {} | |
| for w, cat in sorted(vocab.items()): | |
| byhash.setdefault(word_hash(w), set()).add(cat) | |
| for h, cats in byhash.items(): | |
| assert len(cats) == 1, ("hash collision across categories", h, cats) | |
| hashes = sorted(byhash) | |
| cat_of = [next(iter(byhash[h])) for h in hashes] | |
| for k, h in enumerate(hashes): | |
| u(f"ge_{k}", p_h1=1, b=1 - h, fam="ge") | |
| for cat in CATS: | |
| prev = 0 | |
| deltas = {} | |
| for k, h in enumerate(hashes): | |
| if cat_of[k] == cat: | |
| deltas[f"ge_{k}"] = -(h - prev) | |
| prev = h | |
| u(f"E_{cat}", p_h1=1, **deltas) # 0 exactly when h is a category word, else 1 | |
| kdef, kclass = hashes.index(word_hash("def")), hashes.index(word_hash("class")) | |
| nxt = lambda k: {f"ge_{k + 1}": -1} if k + 1 < len(hashes) else {} | |
| u("is_def", guard=["u.wend"], wend=1, E_KW=-1, b=-1, **{f"ge_{kdef}": 1}, **nxt(kdef)) | |
| u("is_class", guard=["u.wend"], wend=1, E_KW=-1, b=-1, **{f"ge_{kclass}": 1}, **nxt(kclass)) | |
| # ---- contexts: after def, after class, after @, after a dot | |
| u("cD_new", guard=["p.cD", "u.is_def"], p_cD=1, is_def=2, wend=-1, x_nl=-1) | |
| u("cC_new", guard=["p.cC", "u.is_class"], p_cC=1, is_class=2, wend=-1, x_nl=-1) | |
| u("atp_new", guard=["x.at"], x_at=1, S_eff=-1, p_C=-1) | |
| u("cAt_new", guard=["p.cAt", "p.atp"], p_cAt=1, p_atp=1, wordch=1, wend=-1, b=-1) | |
| u("cDot_new", guard=["x.dot", "p.cDot"], x_dot=4, S_eff=-4, p_C=-4, p_N=-4, p_cDot=2, wordch=1, wend=-1, b=-2) | |
| # ---- events at a word's end (read by the backward pass) and the word label | |
| u("ev_DEC", guard=["u.wend"], wend=1, p_cAt=1, b=-1) | |
| u("ev_CL", guard=["u.wend"], wend=1, p_cC=1, p_cAt=-1, b=-1) | |
| u("ev_KW", guard=["u.wend"], wend=1, E_KW=-1, p_cAt=-1, p_cC=-1, p_cD=-1) | |
| u("ev_FN", guard=["u.wend"], wend=2, E_MAG=-1, p_cD=1, p_cAt=-1, p_cC=-1, b=-1) # after def, or a magic method name | |
| u("ev_BI", guard=["u.wend"], wend=1, E_BI=-1, p_cAt=-1, p_cC=-1, p_cD=-1, p_cDot=-1) | |
| u("ev_STR", guard=["u.wend"], wend=1, open=1, p_pf=1, b=-2) # the word was a string prefix | |
| u("lab_W", guard=["u.W_new", "u.atp_new"], W_new=1, atp_new=1) | |
| return hashes, cat_of | |
| def build_additive(vocab): | |
| """Additive retrieval: A accumulates shifted position values over the word, three delayed last-byte values are added | |
| at the end, and the class is read from windows around the targets.""" | |
| vmax = solve_additive(vocab) | |
| C = 2 * vmax # shift: every gated value (+ its position constant) is >= 0; the readout subtracts C per byte | |
| M = 4 * C + 4 # gate constant | |
| AMAX = 2 ** 45; K = 2 ** 46 | |
| u("pos0", guard=["u.W_new"], W_new=1, p_W=-1) # first byte of a word | |
| for p in range(1, ADD_P): | |
| u(f"pos{p}", guard=[f"p.pos{p - 1}"], **{f"p_pos{p - 1}": 1}, W_new=1, b=-1) | |
| u("tail", guard=[f"p.pos{ADD_P - 1}", "p.tail"], **{f"p_pos{ADD_P - 1}": 1}, p_tail=1, W_new=1, b=-1) | |
| for p in range(ADD_P): | |
| u(f"a{p}", hi=2 * C, guard=[f"u.pos{p}"], **{f"x_v{p}": 1, f"pos{p}": M}, b=C - M + ADD_TABLES.get(f"v{p}", {}).get(0, 0)) | |
| u("at_", hi=2 * C, guard=["u.tail"], x_vt=1, tail=M, b=C - M + ADD_TABLES.get("vt", {}).get(0, 0)) | |
| u("A_new", hi=AMAX, guard=["u.W_new"], p_A=1, W_new=K, b=-K, **{f"a{p}": 1 for p in range(ADD_P)}, at_=1) | |
| for j in range(1, ADD_L + 1): | |
| u(f"e{j}_new", hi=2 * C, guard=["u.W_new"], **{f"x_e{j}": 1}, W_new=M, b=C - M) | |
| u("e2d_new", hi=2 * C, guard=["p.e2"], p_e2=1) | |
| u("e3d_new", hi=2 * C, guard=["p.e3"], p_e3=1) | |
| u("e3dd_new", hi=2 * C, guard=["p.e3d"], p_e3d=1) | |
| u("m3", hi=3, guard=["p.d"], p_d=1) # min(length, 3): last-byte terms present | |
| u("SE", hi=AMAX, p_A=1, p_e1=1, p_e2d=1, p_e3dd=1, p_d=-C, m3=-C) # the word's value sum | |
| for t in (1, 2, 3, 4, 5): | |
| u(f"ge{t}", SE=1, b=1 - (t * ADD_T - ADD_EPS)) | |
| u(f"le{t}", SE=-1, b=1 + (t * ADD_T + ADD_EPS)) | |
| # Kernel blocks: a unit is computed inside the first block whose condition (any listed atom non-zero) provably | |
| # covers its guard, following guard chains down to byte features and previous state. Everything else is unconditional. | |
| BLOCKS = [("vocab", ["u.wend"]), | |
| ("str", ["p.S", "p.fo", "p.FS", "p.FD", "p.c", "p.T", "p.F", "p.sq1", "p.oq1", "p.oT", "x.dq", "x.sq"]), | |
| ("word", ["p.W", "x.code", "x.dot", "x.at", "p.atp", "p.cD", "p.cC", "p.cAt", "p.cDot", "x.po", "x.pf", "p.fp", "p.d"] | |
| + ([] if MATCHER == "hash" else [f"p.pos{p}" for p in range(ADD_P)] + ["p.tail", "p.e2", "p.e3", "p.e3d"])), | |
| ("num", ["x.digit", "p.N", "p.ee"]), | |
| ("cmt", ["x.hash", "p.C"])] | |
| BITS = ["lab_STR", "C_new", "N_new", "lab_W", "ev_KW", "ev_BI", "ev_FN", "ev_CL", "ev_DEC", "ev_STR"] | |
| EV = ["KW", "BI", "FN", "CL", "DEC", "STR"] | |
| CLASSES = ["text", "keyword", "builtin", "string", "comment", "number", "func", "class", "decorator"] | |
| BSTATE = ["u.reg_" + e for e in EV] + ["b.ev_" + e for e in EV] | |
| # class id read out by one integer unit: direct labels, or the carried word class on word bytes | |
| HEAD = {"b.lab_STR": 3, "b.C_new": 4, "b.N_new": 5, "u.reg_KW": 1, "u.reg_BI": 2, "u.reg_STR": 3, "u.reg_FN": 6, "u.reg_CL": 7, "u.reg_DEC": 8} | |
| # ---------------------------------------------------------------- layering and matrices | |
| def layer_units(units): | |
| """Layers by dependency on same-step units (guards count); a family gets a layer of its own (tied weights).""" | |
| byname = {x.name: x for x in units} | |
| lvl = {} | |
| def L(n): | |
| if n in lvl: | |
| return lvl[n] | |
| x = byname[n] | |
| deps = [r for r in x.ins if r.startswith("u.")] + [g for g in (x.guard or []) if g.startswith("u.")] | |
| lvl[n] = 1 + max([L(r[2:]) for r in deps], default=-1) | |
| return lvl[n] | |
| for x in units: | |
| L(x.name) | |
| layers = [] | |
| for l in range(1 + max(lvl.values())): | |
| xs = [x for x in units if lvl[x.name] == l] | |
| for fam in sorted({x.fam for x in xs if x.fam}): | |
| layers.append([x for x in xs if x.fam == fam]) | |
| rest = [x for x in xs if not x.fam] | |
| if rest: | |
| layers.append(rest) | |
| return layers | |
| def check_guards(units, emb): | |
| """Same proof as the C loader: with the guard inputs at 0, bias plus every other positive term at its ceiling must be <= 0.""" | |
| hi = {"u." + x.name: x.hi for x in units} | |
| for s_, un in STATE_UNIT.items(): | |
| if "u." + un in hi: # matcher-specific state units exist only in that matcher's build | |
| hi["p." + s_] = hi["u." + un] | |
| for f in FEATS: | |
| hi["x." + f] = float(emb[:, FI[f]].max()) | |
| bad = [] | |
| for x in units: | |
| if not x.guard or getattr(x, "inherited", False): | |
| continue | |
| bound = x.b + sum(w * hi[r] for r, w in x.ins.items() if w > 0 and r not in x.guard) | |
| if bound > 0: | |
| bad.append((x.name, bound)) | |
| assert not bad, f"guards not provable: {bad}" | |
| def inherit_guards(units): | |
| """A unit that is not an output and whose consumers all share one guard set inherits it (it is dead otherwise).""" | |
| cons = {x.name: [] for x in units} | |
| for x in units: | |
| for r in x.ins: | |
| if r.startswith("u."): | |
| cons[r[2:]].append(x) | |
| outputs = set(STATE_UNIT.values()) | set(BITS) | |
| changed = True | |
| while changed: | |
| changed = False | |
| for x in units: | |
| if x.guard is not None or x.name in outputs or not cons[x.name]: | |
| continue | |
| gs = {tuple(c.guard) if c.guard is not None else None for c in cons[x.name]} | |
| if len(gs) == 1 and None not in gs: | |
| x.guard = list(gs.pop()) | |
| x.inherited = True | |
| changed = True | |
| def build_cell(layers, inputs, outputs_needed): | |
| """Layered units -> (W, b, hi, skip, names, tied) over a growing vector, lexnet style. | |
| A tied layer stores one weight row and one ceiling for all its rows (a threshold family).""" | |
| vec = list(inputs) | |
| mats = [] | |
| later_needs = [] | |
| for l, xs in enumerate(layers): | |
| need = set(outputs_needed) | |
| for ys in layers[l + 1:]: | |
| for y in ys: | |
| need |= set(y.ins) | |
| later_needs.append(need) | |
| for l, xs in enumerate(layers): | |
| idx = {n: i for i, n in enumerate(vec)} | |
| tied = all(x.fam for x in xs) | |
| if tied: | |
| assert all(x.ins == xs[0].ins and x.hi == xs[0].hi for x in xs) | |
| W = np.zeros((1 if tied else len(xs), len(vec))) | |
| b = np.zeros(len(xs)); hi = np.zeros(1 if tied else len(xs)) | |
| for r, x in enumerate(xs): | |
| if not tied or r == 0: | |
| for ref, w in x.ins.items(): | |
| W[r, idx[ref]] = w | |
| hi[r] = x.hi | |
| b[r] = x.b | |
| skip = [i for i, n in enumerate(vec) if n in later_needs[l]] | |
| mats.append((W, b, hi, np.array(skip, dtype=np.int64), [x.name for x in xs], tied)) | |
| vec = ["u." + x.name for x in xs] + [vec[i] for i in skip] | |
| return mats, vec | |
| # ---------------------------------------------------------------- weights file | |
| def write_weights(path, emb, fwd, fvec, bwd, bvec, head_W, head_b, h0): | |
| out = bytearray(b"PYLX02\0\0") | |
| out += struct.pack("<6q", NF, len(STATE), len(BITS), len(BSTATE), len(CLASSES), head_W.shape[1]) | |
| out += emb.astype("<f8").tobytes() | |
| for cell in (fwd, bwd): | |
| out += struct.pack("<q", len(cell)) | |
| for W, b, hi, skip, names, tied in cell: | |
| out += struct.pack("<4q", len(names), W.shape[1], len(skip), int(tied)) | |
| out += W.astype("<f8").tobytes() + b.astype("<f8").tobytes() + hi.astype("<f8").tobytes() + skip.astype("<q").tobytes() | |
| f_state = [fvec.index("u." + STATE_UNIT[s]) for s in STATE] | |
| f_out = [fvec.index("u." + bname) for bname in BITS] | |
| b_state = [bvec.index(n) for n in BSTATE] | |
| out += np.array(f_state, dtype="<q").tobytes() + np.array(f_out, dtype="<q").tobytes() + np.array(b_state, dtype="<q").tobytes() | |
| out += h0.astype("<f8").tobytes() + head_W.astype("<f8").tobytes() + head_b.astype("<f8").tobytes() | |
| Path(path).write_bytes(out) | |
| return len(out) | |
| def count(emb, fwd, bwd, head_W, head_b, h0): | |
| """Every non-zero number the network uses: embedding, weights, biases, ceilings, head, initial state.""" | |
| nz = lambda a: int((np.asarray(a) != 0).sum()) | |
| tied = nz(emb) + nz(head_W) + nz(head_b) + nz(h0) + 1 # +1: the head ceiling | |
| untied = tied | |
| for cell in (fwd, bwd): | |
| for W, b, hi, skip, names, t in cell: | |
| tied += nz(W) + nz(b) + nz(hi) | |
| untied += (nz(W) + nz(hi)) * (len(names) if t else 1) + nz(b) | |
| return tied, untied | |
| # ---------------------------------------------------------------- kernel codegen | |
| def struct_hash(fwd, bwd, f_state, f_out, b_state, head_W, head_cols): | |
| """FNV-1a over shapes, ceilings, skips, sparsity masks and index maps; identical to the C loader's.""" | |
| h = 0xCBF29CE484222325 | |
| def by(bs): | |
| nonlocal h | |
| for c in bs: | |
| h ^= c; h = (h * 0x100000001B3) & 0xFFFFFFFFFFFFFFFF | |
| def i64(x): | |
| by(struct.pack("<q", x)) | |
| i64(NF); i64(len(STATE)); i64(len(BITS)); i64(len(BSTATE)); i64(len(CLASSES)); i64(head_cols) | |
| for cell in (fwd, bwd): | |
| i64(len(cell)) | |
| for W, b, hi, skip, names, tied in cell: | |
| i64(len(names)); i64(W.shape[1]); i64(len(skip)); i64(int(tied)) | |
| by(hi.astype("<f8").tobytes()) | |
| for s in skip: | |
| i64(int(s)) | |
| by(bytes((W != 0).astype(np.uint8).ravel())) | |
| for i in f_state: i64(i) | |
| for i in f_out: i64(i) | |
| for i in b_state: i64(i) | |
| by(bytes((head_W != 0).astype(np.uint8).ravel())) | |
| return h | |
| def cfmt(v): | |
| return repr(float(v)) if v != int(v) else f"{int(v)}.0" | |
| def tree(terms): | |
| """Balanced sum: every value is an exact integer, so the order does not change the result but halves the latency.""" | |
| while len(terms) > 1: | |
| terms = [f"({terms[i]} + {terms[i + 1]})" if i + 1 < len(terms) else terms[i] for i in range(0, len(terms), 2)] | |
| return terms[0] | |
| _FVEC = {} | |
| def fvec_index(cell, l, ref): | |
| return _FVEC[(id(cell), l)][ref] | |
| def gen_kernel(path, fwd, fvec, bwd, bvec, head_W, head_b, head_hi, shash, units, bunits): | |
| coef, bias, guards, families, prefixes = [], [], [], [], [] | |
| lines = [] | |
| A = lines.append | |
| A("/* Generated by compile.py: the network's ARCHITECTURE as C. No weight values live here: pylexnet.c") | |
| A(" * loads them, checks the structure hash and the guard inequalities, and gathers kc[]/kb[] by K_COEF/K_BIAS. */") | |
| A("#include <stdint.h>") | |
| A("#include <string.h>") | |
| A(f"#define K_STRUCT_HASH 0x{shash:016X}ULL") | |
| A(f"#define K_N_FEATS {NF}") | |
| A(f"#define K_N_CLASSES {len(CLASSES)}") | |
| A(f"#define K_N_BITS {len(BITS)}") | |
| A("#ifndef CLIP") | |
| A("#define CLIP(x, hi) clipn((x), (hi))") | |
| A("static inline double clipn(double x, double hi) { x = x < 0.0 ? 0.0 : x; return x > hi ? hi : x; }") | |
| A("#endif") | |
| A("/* A tied, sorted threshold family is a step vector: unit k is 1 iff kb[k] + w*h >= 1, true for k < rank.") | |
| A(" * The loader verifies integrality, tying and monotonicity, so a binary search gives the same values. */") | |
| A("static inline int k_rank(const double *kb, int n, double w, double h) {") | |
| A(" int lo = 0, hi = n;") | |
| A(" while (lo < hi) { int mid = (lo + hi) >> 1; if (kb[mid] + w * h >= 1.0) lo = mid + 1; else hi = mid; }") | |
| A(" return lo;") | |
| A("}") | |
| A("typedef struct { double " + ", ".join(STATE) + "; } FState;") | |
| A("typedef struct { double " + ", ".join(n.replace(".", "_") for n in BSTATE) + "; } BState;") | |
| A("static inline void k_init_state(FState *s, const double *h0) { memcpy(s, h0, sizeof *s); }") | |
| # ---- forward | |
| A("static inline __attribute__((always_inline)) uint16_t k_fwd_step(const double *restrict kc, const double *restrict kb, const double *restrict kp, const double *restrict x,") | |
| A(" const FState *restrict p, FState *restrict s) {") | |
| byname = {x.name: x for x in units} | |
| famidx, fams = {}, {} | |
| for x in units: | |
| if x.fam: | |
| famidx[x.name] = len(fams.setdefault(x.fam, [])) | |
| fams[x.fam].append(x) | |
| def ref(r): | |
| if r.startswith("x."): | |
| return f"x[{FI[r[2:]]}]" | |
| if r.startswith("p."): | |
| return f"p->{r[2:]}" | |
| x = byname[r[2:]] | |
| return f"(double)({famidx[x.name]} < r_{x.fam})" if x.fam else f"u_{x.name}" | |
| scalars = [x.name for x in units if not x.fam] | |
| A(" double " + ", ".join(f"u_{n} = 0.0" for n in scalars) + ";") | |
| for fam in fams: | |
| A(f" int r_{fam} = 0;") | |
| def covered(atom, cond, seen=()): | |
| if atom in cond: | |
| return True | |
| if not atom.startswith("u.") or atom in seen: | |
| return False | |
| g = byname[atom[2:]].guard | |
| return g is not None and all(covered(a, cond, seen + (atom,)) for a in g) | |
| blk = {} | |
| for x in units: | |
| blk[x.name] = None | |
| if x.guard: | |
| for bname, cond in BLOCKS: | |
| if all(covered(a, cond) for a in x.guard): | |
| blk[x.name] = bname | |
| break | |
| cond_of = dict(BLOCKS) | |
| layer_of = {} | |
| for l, (W, b, hi, skip, names, tied) in enumerate(fwd): | |
| for row, n in enumerate(names): | |
| layer_of[n] = (l, row) | |
| def deps(x): | |
| d = {r[2:] for r in x.ins if r.startswith("u.")} | |
| if blk[x.name]: | |
| d |= {a[2:] for a in cond_of[blk[x.name]] if a.startswith("u.")} | |
| return d | |
| remaining = {x.name for x in units} | |
| done = set() | |
| cur = None | |
| nseg = 0 | |
| while remaining: | |
| ready = [byname[n] for n in remaining if deps(byname[n]) <= done] | |
| assert ready, "cyclic dependency" | |
| bybl = {} | |
| for x in ready: | |
| bybl.setdefault(blk[x.name], []).append(x) | |
| if cur not in bybl: | |
| cur = max(bybl, key=lambda k: (len(bybl[k]), k is None)) | |
| if cur is not None: | |
| A(" if ((" + " + ".join(ref(a) for a in cond_of[cur]) + ") != 0.0) { /* all activations are >= 0 */") | |
| nseg += 1 | |
| batch = sorted(bybl[cur], key=lambda x: (layer_of[x.name], x.fam or "", famidx.get(x.name, 0))) | |
| indent = " " if cur is not None else " " | |
| i = 0 | |
| while i < len(batch): | |
| x = batch[i] | |
| l, row = layer_of[x.name] | |
| if x.fam: | |
| j = i | |
| while j < len(batch) and batch[j].fam == x.fam and famidx[batch[j].name] == famidx[x.name] + (j - i): | |
| j += 1 | |
| run = batch[i:j] | |
| assert len(run) == len(fams[x.fam]) and famidx[x.name] == 0, "a family is emitted whole" | |
| (inref, _), = run[0].ins.items() | |
| cb, bb = len(coef), len(bias) | |
| coef.append((0, l, 0, fvec_index(fwd, l, inref))) # tied: one weight for the family | |
| for y in run: | |
| bias.append((0,) + layer_of[y.name]) | |
| families.append((0, l, bb, len(run), cb)) | |
| A(f"{indent}r_{x.fam} = k_rank(kb + {bb}, {len(run)}, kc[{cb}], {ref(inref)});") | |
| i = j | |
| continue | |
| terms = [f"kb[{len(bias)}]"] | |
| bias.append((0, l, row)) | |
| famins = {} | |
| for inref, w in x.ins.items(): | |
| y = byname[inref[2:]] if inref.startswith("u.") else None | |
| if y is not None and y.fam: | |
| famins.setdefault(y.fam, []).append(inref) | |
| for inref, w in x.ins.items(): | |
| y = byname[inref[2:]] if inref.startswith("u.") else None | |
| if y is not None and y.fam and len(famins[y.fam]) >= 4: | |
| if inref == famins[y.fam][0]: | |
| # all of this family's weights into the unit, as a prefix sum indexed by the rank (built at load) | |
| cols = [fvec_index(fwd, l, "u." + z.name) for z in fams[y.fam]] | |
| assert cols == list(range(cols[0], cols[0] + len(cols))), "family columns are contiguous" | |
| pb = sum(n + 1 for _, _, _, _, n in prefixes) | |
| prefixes.append((0, l, row, cols[0], len(cols))) | |
| terms.append(f"kp[{pb} + r_{y.fam}]") | |
| continue | |
| terms.append(f"kc[{len(coef)}] * {ref(inref)}") | |
| coef.append((0, l, row, fvec_index(fwd, l, inref))) | |
| A(f"{indent}u_{x.name} = CLIP({tree(terms)}, {cfmt(x.hi)});") | |
| if x.guard and not getattr(x, "inherited", False): | |
| guards.append((0, l, row, [fvec_index(fwd, l, gr) for gr in x.guard])) | |
| i += 1 | |
| for x in batch: | |
| done.add(x.name); remaining.discard(x.name) | |
| nxt = [n for n in remaining if blk[n] == cur and deps(byname[n]) <= done] | |
| if not nxt and cur is not None: | |
| A(" }") | |
| cur = None | |
| print(f"kernel: {nseg} block segments; units per block " + str({b: sum(1 for x in units if blk[x.name] == b) for b in [None] + [b for b, _ in BLOCKS]})) | |
| for sname in STATE: | |
| A(f" s->{sname} = u_{STATE_UNIT[sname]};") | |
| A(" return (uint16_t)(" + " | ".join(f"((uint16_t)(u_{bn} != 0.0) << {i})" for i, bn in enumerate(BITS)) + ");") | |
| A("}") | |
| # ---- backward: carried word classes, then the integer class id | |
| A("static inline __attribute__((always_inline)) int k_bwd_step(const double *restrict kc, const double *restrict kb, uint16_t bits,") | |
| A(" const BState *restrict r, BState *restrict s) {") | |
| for i, bn in enumerate(BITS): | |
| A(f" const double f_{bn} = (double)((bits >> {i}) & 1);") | |
| def bref(rf): | |
| if rf.startswith("b."): | |
| return "f_" + rf[2:] | |
| if rf.startswith("r."): | |
| return "r->" + rf[2:].replace(".", "_") | |
| return "u_" + rf[2:] | |
| for l, (W, b, hi, skip, names, tied) in enumerate(bwd): | |
| for row, n in enumerate(names): | |
| x = bunits[n] | |
| terms = [f"kb[{len(bias)}]"] | |
| bias.append((1, l, row)) | |
| for inref, w in x.ins.items(): | |
| terms.append(f"kc[{len(coef)}] * {bref(inref)}") | |
| coef.append((1, l, row, fvec_index(bwd, l, inref))) | |
| A(f" const double u_{n} = CLIP({tree(terms)}, {cfmt(x.hi)});") | |
| for n in BSTATE: | |
| A(f" s->{n.replace('.', '_')} = {bref(n) if not n.startswith('u.') else 'u_' + n[2:]};") | |
| terms = [f"kb[{len(bias)}]"] | |
| bias.append((2, 0, 0)) | |
| for c in np.nonzero(head_W[0])[0]: | |
| terms.append(f"kc[{len(coef)}] * {bref(bvec[c])}") | |
| coef.append((2, 0, 0, int(c))) | |
| A(f" return (int)CLIP({tree(terms)}, {cfmt(head_hi)});") | |
| A("}") | |
| A(f"#define K_N_COEF {len(coef)}") | |
| A(f"#define K_N_BIAS {len(bias)}") | |
| A(f"#define K_N_GUARD {len(guards)}") | |
| A(f"#define K_N_FAMILY {len(families)}") | |
| A(f"#define K_N_PREFIX {len(prefixes)}") | |
| A(f"#define K_PREFIX_LEN {sum(n + 1 for *_, n in prefixes)}") | |
| A("/* sorted threshold families: {cell, layer, first bias index, rows, coef index of the tied weight} */") | |
| A("static const int32_t K_FAMILY[K_N_FAMILY + 1][5] = {" + "".join("{%d,%d,%d,%d,%d}," % f for f in families) + "{0,0,0,0,0}};") | |
| A("/* prefix sums of a unit's weights over a family's columns: {cell, layer, row, first column, rows} */") | |
| A("static const int32_t K_PREFIX[K_N_PREFIX + 1][5] = {" + "".join("{%d,%d,%d,%d,%d}," % f for f in prefixes) + "{0,0,0,0,0}};") | |
| A("#define K_MAX_GUARD 4") | |
| A("static const int32_t K_COEF[K_N_COEF][4] = {" + ",".join("{%d,%d,%d,%d}" % c for c in coef) + "};") | |
| A("static const int32_t K_BIAS[K_N_BIAS][3] = {" + ",".join("{%d,%d,%d}" % c for c in bias) + "};") | |
| A("/* units whose block is skipped when the listed input columns are all zero; the loader proves each one */") | |
| A("static const int32_t K_GUARD[K_N_GUARD][3 + K_MAX_GUARD] = {" + | |
| ",".join("{%d,%d,%d,%s}" % (c, l, r, ",".join(str(v) for v in (cols + [-1] * 4)[:4])) for c, l, r, cols in guards) + "};") | |
| Path(path).write_text("\n".join(lines) + "\n") | |
| return len(coef), len(bias), len(guards) | |
| # ---------------------------------------------------------------- simulator (debugging: a third evaluation of the same matrices) | |
| def run_layers(v, layers): | |
| for W, b, hi, skip, names, tied in layers: | |
| pre = (W @ v) | |
| if tied: | |
| pre = np.full(len(names), pre[0]) | |
| new = np.clip(pre + b, 0, hi if not tied else hi[0]) | |
| v = np.concatenate([new, v[skip]]) | |
| return v | |
| def simulate(text, emb, fwd, fvec, f_state, f_out, bwd, b_state, head_W, head_b, head_hi, h0): | |
| data = text.encode() + b"\n" | |
| st = h0.copy() | |
| allbits = [] | |
| for t, byte in enumerate(data): | |
| v = run_layers(np.concatenate([emb[byte if byte < 128 else 127], st]), fwd) | |
| st = v[f_state] | |
| bits = [int(v[i] != 0) for i in f_out] | |
| allbits.append(bits) | |
| on = [n[2:] + ("" if v[i] == 1 else f"={v[i]:g}") for i, n in enumerate(fvec) | |
| if n.startswith("u.") and v[i] != 0 and not n.startswith("u.ge_")] | |
| print(f"{t:3} {chr(byte)!r:6} bits={''.join(map(str, bits))} " + " ".join(on)) | |
| r = np.zeros(len(b_state)) | |
| out = [] | |
| for t in range(len(data) - 1, -1, -1): | |
| v = run_layers(np.concatenate([np.array(allbits[t], dtype=float), r]), bwd) | |
| r = v[b_state] | |
| out.append(int(np.clip(head_W @ v + head_b, 0, head_hi)[0])) | |
| print("classes:", "".join(str(c) for c in out[::-1][:-1])) | |
| # ---------------------------------------------------------------- main | |
| class BUnit: | |
| def __init__(self, name, ins, b=0.0, hi=1.0): | |
| self.name, self.ins, self.b, self.hi = name, ins, b, hi | |
| def main(): | |
| limit = None | |
| if "--vocab" in sys.argv: | |
| limit = int(sys.argv[sys.argv.index("--vocab") + 1]) | |
| vocab = pick_vocab(limit) | |
| hashes, cat_of = build(vocab) | |
| check_guards(UNITS, emb_table()) | |
| inherit_guards(UNITS) | |
| layers = layer_units(UNITS) | |
| inputs = ["x." + f for f in FEATS] + ["p." + s for s in STATE] | |
| needed = {"u." + STATE_UNIT[s] for s in STATE} | {"u." + b for b in BITS} | |
| fwd, fvec = build_cell(layers, inputs, needed) | |
| vec = list(inputs) | |
| for l, (W, b, hi, skip, names, tied) in enumerate(fwd): | |
| _FVEC[(id(fwd), l)] = {n: i for i, n in enumerate(vec)} | |
| vec = ["u." + n for n in names] + [vec[i] for i in skip] | |
| # backward cell: reg_X(i) = clip1(ev_X(i+1) + reg_X(i+1) + lab_W(i) - 1): set at a word's end, carried over | |
| # the word's bytes, zero elsewhere. Carries this byte's event bits for the next step. | |
| bunits = {} | |
| for e in EV: | |
| bunits["reg_" + e] = BUnit("reg_" + e, {"r.b.ev_" + e: 1.0, "r.u.reg_" + e: 1.0, "b.lab_W": 1.0}, b=-1.0) | |
| binputs = ["b." + bn for bn in BITS] + ["r." + n for n in BSTATE] | |
| bidx = {n: i for i, n in enumerate(binputs)} | |
| Wb = np.zeros((len(EV), len(binputs))); bb = -np.ones(len(EV)); hb = np.ones(len(EV)) | |
| for r, e in enumerate(EV): | |
| for ref, w in bunits["reg_" + e].ins.items(): | |
| Wb[r, bidx[ref]] = w | |
| bskip = [bidx["b." + bn] for bn in BITS] | |
| bwd = [(Wb, bb, hb, np.array(bskip, dtype=np.int64), ["reg_" + e for e in EV], False)] | |
| _FVEC[(id(bwd), 0)] = bidx | |
| bvec_out = ["u.reg_" + e for e in EV] + ["b." + bn for bn in BITS] | |
| hcol = {n: i for i, n in enumerate(bvec_out)} | |
| head_W = np.zeros((1, len(bvec_out))); head_b = np.zeros(1); head_hi = 8.0 | |
| for n, w in HEAD.items(): | |
| head_W[0, hcol[n]] = w | |
| emb = emb_table() | |
| h0 = np.zeros(len(STATE)) | |
| size = write_weights(HERE / "pylexnet_weights.bin", emb, fwd, fvec, bwd, bvec_out, head_W, head_b, h0) | |
| f_state = [fvec.index("u." + STATE_UNIT[s]) for s in STATE] | |
| f_out = [fvec.index("u." + bn) for bn in BITS] | |
| b_state = [bvec_out.index(n) for n in BSTATE] | |
| shash = struct_hash(fwd, bwd, f_state, f_out, b_state, head_W, head_W.shape[1]) | |
| ncoef, nbias, nguard = gen_kernel(HERE / "pylexnet_kernel.h", fwd, fvec, bwd, bvec_out, head_W, head_b, head_hi, shash, UNITS, bunits) | |
| if "--sim" in sys.argv: | |
| simulate(sys.argv[sys.argv.index("--sim") + 1], emb, fwd, fvec, f_state, f_out, bwd, b_state, head_W, head_b, head_hi, h0) | |
| tied, untied = count(emb, fwd, bwd, head_W, head_b, h0) | |
| nlog = sum(1 for x in UNITS if not x.fam and not x.name.startswith("E_")) | |
| vocab_numbers = 2 * len(hashes) if MATCHER == "hash" else sum(len(t) for t in ADD_TABLES.values()) | |
| print(f"units {len(UNITS)} forward in {len(fwd)} layers ({nlog} logic), state {len(STATE)}, bits {len(BITS)}, vocabulary {len(vocab)} words " | |
| f"({sum(1 for c in vocab.values() if c == 'KW')} keywords, {sum(1 for c in vocab.values() if c == 'BI')} builtins, " | |
| f"{sum(1 for c in vocab.values() if c == 'MAG')} magic) = {len(hashes)} thresholds; kernel coef {ncoef} bias {nbias} guards {nguard}") | |
| print(f"non-zero weights {tied} (every number: embedding {int((emb != 0).sum())}, vocabulary {vocab_numbers}; " | |
| f"{untied} if the threshold family's weight and ceiling were not tied)") | |
| print(f"weights file {size // 1024} KiB") | |
| if __name__ == "__main__": | |
| main() | |