SPIRA v2: entropy-patch tokenization + hierarchical dopamine-gated spiking cortex + hippocampal one-shot memory with replay consolidation
032a961 verified | #!/usr/bin/env python3 | |
| """Build the SPIRA training corpus: mixed natural language + source code. | |
| Sources: | |
| - English books from Project Gutenberg (public domain) | |
| - French books from Project Gutenberg (flavour / multilingual robustness) | |
| - Python standard library source files (code domain) | |
| Outputs (under python/data/ next to the scripts): | |
| - train.bin : ~95% of the interleaved stream (text + code) | |
| - test.bin : held-out tail slice (~5%) | |
| - oneshot.bin : a full book never used in training (one-shot / fresh-text tests) | |
| - corpus_manifest.json | |
| """ | |
| import glob | |
| import json | |
| import os | |
| import shutil | |
| import subprocess | |
| import sys | |
| BASE = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) | |
| DATA = f"{BASE}/data" | |
| BOOKS = f"{DATA}/corpus/books" | |
| CODE = f"{DATA}/corpus/code" | |
| # (gutenberg_id, name, holdout?) | |
| ENGLISH_BOOKS = [ | |
| (11, "alice", False), | |
| (84, "frankenstein", False), | |
| (120, "treasure_island", False), | |
| (174, "dorian_gray", False), | |
| (98, "two_cities", False), | |
| (345, "dracula", False), | |
| (5200, "metamorphosis", True), # held out entirely -> oneshot.bin | |
| (1342, "pride_prejudice", False), | |
| (1400, "great_expectations", False), | |
| (1661, "sherlock", False), | |
| (2701, "moby_dick", False), | |
| ] | |
| FRENCH_BOOKS = [ | |
| (19942, "candide_fr", False), | |
| (17489, "les_mis_fr", False), | |
| ] | |
| CODE_BUDGET_BYTES = 6 * 1024 * 1024 | |
| SKIP_CODE_DIRS = {"test", "tests", "__pycache__", "lib2to3", "idlelib", "site-packages", "turtledemo"} | |
| SKIP_CODE_FILES = {"setup.py"} | |
| DOC_SEP = b"\n\n" | |
| def strip_gutenberg_header(raw: bytes) -> bytes: | |
| """Remove the Project Gutenberg boilerplate header/footer when present.""" | |
| marker = b"*** START OF" | |
| idx = raw.find(marker) | |
| if idx != -1: | |
| end = raw.find(b"\n", idx) | |
| if end != -1: | |
| raw = raw[end + 1:] | |
| marker = b"*** END OF" | |
| idx = raw.find(marker) | |
| if idx != -1: | |
| raw = raw[:idx] | |
| return raw | |
| def french_ratio(raw: bytes) -> float: | |
| """Fraction of letters that carry French diacritics (cheap language check).""" | |
| letters = 0 | |
| for b in raw: | |
| if 65 <= b <= 90 or 97 <= b <= 122: | |
| letters += 1 | |
| accented = raw.count(0xC3) # UTF-8 lead byte for most accented Latin letters | |
| if letters == 0: | |
| return 0.0 | |
| return accented / max(letters, 1) | |
| def fetch_book(gid: int, name: str) -> str | None: | |
| os.makedirs(BOOKS, exist_ok=True) | |
| dst = f"{BOOKS}/{name}.txt" | |
| if os.path.exists(dst) and os.path.getsize(dst) > 20_000: | |
| return dst | |
| urls = [ | |
| f"https://www.gutenberg.org/cache/epub/{gid}/pg{gid}.txt", | |
| f"https://www.gutenberg.org/files/{gid}/{gid}-0.txt", | |
| f"https://www.gutenberg.org/files/{gid}/{gid}.txt", | |
| ] | |
| for url in urls: | |
| r = subprocess.run(["curl", "-sL", "-m", "120", "-o", dst, url]) | |
| if r.returncode == 0 and os.path.exists(dst) and os.path.getsize(dst) > 20_000: | |
| return dst | |
| if os.path.exists(dst): | |
| os.remove(dst) | |
| return None | |
| def collect_code_files(stdlib_dir: str, budget: int) -> list[str]: | |
| files = [] | |
| total = 0 | |
| for root, dirs, fnames in os.walk(stdlib_dir): | |
| dirs[:] = [d for d in sorted(dirs) if d not in SKIP_CODE_DIRS] | |
| for fn in sorted(fnames): | |
| if not fn.endswith(".py") or fn in SKIP_CODE_FILES: | |
| continue | |
| path = os.path.join(root, fn) | |
| try: | |
| sz = os.path.getsize(path) | |
| except OSError: | |
| continue | |
| if sz < 200: # skip stubs | |
| continue | |
| files.append(path) | |
| total += sz | |
| if total >= budget: | |
| return files | |
| return files | |
| def main() -> None: | |
| os.makedirs(DATA, exist_ok=True) | |
| os.makedirs(CODE, exist_ok=True) | |
| docs = [] # list of (domain, bytes) | |
| holdout_docs = [] | |
| print("== Downloading books ==") | |
| for gid, name, holdout in ENGLISH_BOOKS + FRENCH_BOOKS: | |
| path = fetch_book(gid, name) | |
| if path is None: | |
| print(f" !! failed: {name} (gid={gid})") | |
| continue | |
| raw = open(path, "rb").read() | |
| raw = strip_gutenberg_header(raw) | |
| if len(raw) < 20_000: | |
| print(f" !! too small after strip: {name}") | |
| continue | |
| domain = "fr" if name.endswith("_fr") else "en" | |
| if holdout: | |
| holdout_docs.append((domain, raw)) | |
| else: | |
| docs.append((domain, raw)) | |
| print(f" ok: {name:18s} {len(raw)/1024:8.1f} KB ({'holdout' if holdout else 'train'})") | |
| print("== Collecting Python stdlib code ==") | |
| import sysconfig | |
| stdlib_dir = sysconfig.get_paths()["stdlib"] | |
| code_files = collect_code_files(stdlib_dir, CODE_BUDGET_BYTES) | |
| print(f" {len(code_files)} files from {stdlib_dir}") | |
| for path in code_files: | |
| try: | |
| raw = open(path, "rb").read() | |
| except OSError: | |
| continue | |
| if not raw: | |
| continue | |
| docs.append(("code", raw)) | |
| dst = os.path.join(CODE, os.path.basename(path)) | |
| if not os.path.exists(dst): | |
| shutil.copyfile(path, dst) | |
| # ---- interleave documents deterministically (seeded shuffle) ---- | |
| import random | |
| rng = random.Random(20261022) | |
| rng.shuffle(docs) | |
| stream = DOC_SEP.join(d[1] for d in docs) | |
| n_test = int(len(stream) * 0.05) | |
| # cut at a document boundary close to the tail | |
| cut = len(stream) - n_test | |
| boundary = stream.rfind(DOC_SEP, 0, cut + n_test // 2) | |
| if boundary == -1 or boundary < len(stream) // 2: | |
| boundary = cut | |
| train = stream[:boundary] | |
| test = stream[boundary + len(DOC_SEP):] | |
| with open(f"{DATA}/train.bin", "wb") as f: | |
| f.write(train) | |
| with open(f"{DATA}/test.bin", "wb") as f: | |
| f.write(test) | |
| oneshot = DOC_SEP.join(d[1] for d in holdout_docs) | |
| with open(f"{DATA}/oneshot.bin", "wb") as f: | |
| f.write(oneshot) | |
| en_bytes = sum(len(d[1]) for d in docs if d[0] == "en") | |
| fr_bytes = sum(len(d[1]) for d in docs if d[0] == "fr") | |
| code_bytes = sum(len(d[1]) for d in docs if d[0] == "code") | |
| manifest = { | |
| "train_bytes": len(train), | |
| "test_bytes": len(test), | |
| "oneshot_bytes": len(oneshot), | |
| "composition": {"english": en_bytes, "french": fr_bytes, "code": code_bytes}, | |
| "n_documents": len(docs), | |
| "n_holdout_documents": len(holdout_docs), | |
| "code_files": len(code_files), | |
| } | |
| with open(f"{DATA}/corpus_manifest.json", "w") as f: | |
| json.dump(manifest, f, indent=2) | |
| print(json.dumps(manifest, indent=2)) | |
| print("DONE") | |
| if __name__ == "__main__": | |
| main() | |