"""PALIMPSESTE — Serialization (save/load to disk). The whole substrate is an append-only log of ``Trace`` records plus a small amount of config/encoder state. Serialization is therefore straightforward: - Memory traces -> a compact binary blob (one record per line: id, weight, meta flag, packed bits of address + value) - Encoder atoms -> JSON index + a packed-bits blob - Config -> JSON We avoid pickle for security/portability: everything is numpy arrays + JSON. A saved model directory looks like:: model_dir/ config.json memory.bin # traces (address + value packed bits, weights, flags) memory_index.bin # LSH projection positions (for exact reload) encoder.json # atom/role/level tables (keys -> ids) encoder.bin # packed bits for all atom/role/level HVs tokenizer.json # if a tokenizer is attached This format is deterministic, auditable, and loads on any machine without the original Python objects. """ from __future__ import annotations from dataclasses import asdict import json import os from pathlib import Path import numpy as np from .hv import HV, DEFAULT_D from .memory import Memory, Trace from .phi import Phi, KernelConfig from .learner import Encoder from .lsh import LSHConfig, LSHIndex __all__ = ["save_memory", "load_memory", "save_encoder", "load_encoder", "save_config", "load_config"] # ----------------------------------------------------------------- config def save_config(path: str | Path, config: dict) -> None: p = Path(path) p.parent.mkdir(parents=True, exist_ok=True) with open(p, "w", encoding="utf-8") as f: json.dump(config, f, indent=2, ensure_ascii=False) def load_config(path: str | Path) -> dict: with open(path, "r", encoding="utf-8") as f: return json.load(f) # ----------------------------------------------------------------- memory def save_memory(mem: Memory, path: str | Path) -> None: """Serialize a :class:`Memory` to ``memory.bin`` + ``memory_index.bin``. Format of ``memory.bin`` (numpy ``.npy``-style, but custom for compactness): Header (JSON, one line): {"D": D, "n_traces": N, "n_meta": M, "decay": {...}, "version": 1} Body (raw bytes): For each trace (non-meta first, then meta): id (int64), weight (float64), t_insert (float64), tag_len (int32), tag (utf-8 bytes), address_bits (D/8 packed), value_bits (D/8 packed) """ p = Path(path) p.parent.mkdir(parents=True, exist_ok=True) header = { "D": mem.D, "n_traces": len(mem._traces), "n_meta": len(mem._meta_traces), "decay": mem.decay, "version": 1, } chunks: list[bytes] = [] chunks.append((json.dumps(header) + "\n").encode("utf-8")) # Vectorized serialization: for large |M|, skip tags entirely (version 2 # format) and use fixed-width records. Each record is: # id(int64) + weight(float64) + t_insert(float64) + addr_bits + val_bits # = 24 + 2*packed_len bytes per trace, all stackable as a numpy array. packed_len = (mem.D + 7) // 8 record_size = 24 + 2 * packed_len def _emit_batch_streaming(traces: list[Trace], f, chunk_size: int = 50000): """Write traces to file in chunks to avoid MemoryError on large |M|.""" if not traces: return n = len(traces) for start in range(0, n, chunk_size): end = min(start + chunk_size, n) batch = traces[start:end] bn = len(batch) buf = np.empty((bn, record_size), dtype=np.uint8) ids = np.array([tr.id for tr in batch], dtype=np.int64) weights = np.array([tr.weight for tr in batch], dtype=np.float64) t_ins = np.array([tr.t_insert for tr in batch], dtype=np.float64) buf[:, :8] = np.frombuffer(ids.tobytes(), dtype=np.uint8).reshape(bn, 8) buf[:, 8:16] = np.frombuffer(weights.tobytes(), dtype=np.uint8).reshape(bn, 8) buf[:, 16:24] = np.frombuffer(t_ins.tobytes(), dtype=np.uint8).reshape(bn, 8) buf[:, 24:24+packed_len] = np.stack([tr.address.bits for tr in batch]) buf[:, 24+packed_len:] = np.stack([tr.value.bits for tr in batch]) f.write(buf.tobytes()) with open(p, "wb") as f: f.write((json.dumps(header) + "\n").encode("utf-8")) _emit_batch_streaming(mem._traces, f) _emit_batch_streaming(mem._meta_traces, f) # Save LSH projection positions so the index is reproducible on reload. # np.save appends .npy, so we name the file accordingly. idx_path = p.with_name(p.stem + ".index.npy") assert mem._index is not None proj = mem._index._positions np.save(idx_path, proj) def load_memory(path: str | Path, rng: np.random.Generator | None = None) -> Memory: """Load a :class:`Memory` from ``save_memory`` output. Rebuilds the LSH index with the *saved* projection positions so retrieval is bit-identical to the saved model. """ p = Path(path) with open(p, "rb") as f: raw = f.read() # parse header nl = raw.index(b"\n") header = json.loads(raw[:nl].decode("utf-8")) D = header["D"] n_traces = header["n_traces"] n_meta = header["n_meta"] decay = header.get("decay", {"half_life": float("inf"), "floor": 1e-3}) body = raw[nl + 1:] packed_len = (D + 7) // 8 record_size = 24 + 2 * packed_len # id(8) + weight(8) + t_insert(8) + addr + val # Vectorized read: interpret body as a (n_total, record_size) uint8 array n_total = n_traces + n_meta if n_total == 0: if rng is None: rng = np.random.default_rng() return Memory(D=D, decay=decay, rng=rng) buf = np.frombuffer(body, dtype=np.uint8, count=n_total * record_size).reshape(n_total, record_size) # Extract fixed fields ids = buf[:, :8].copy().view(np.int64).reshape(-1) weights = buf[:, 8:16].copy().view(np.float64).reshape(-1) t_ins = buf[:, 16:24].copy().view(np.float64).reshape(-1) addr_bits_all = buf[:, 24:24+packed_len].copy() val_bits_all = buf[:, 24+packed_len:].copy() if rng is None: rng = np.random.default_rng() mem = Memory(D=D, decay=decay, rng=rng) # Restore LSH projections BEFORE inserting traces idx_path = p.with_name(p.stem + ".index.npy") if idx_path.exists(): proj = np.load(idx_path) assert mem._index is not None mem._index._positions = proj # Insert traces (non-meta first, then meta) for i in range(n_traces): tr = Trace( id=int(ids[i]), address=HV(bits=addr_bits_all[i], D=D), value=HV(bits=val_bits_all[i], D=D), weight=float(weights[i]), t_insert=float(t_ins[i]), meta=False, tag=None, ) mem._traces.append(tr) assert mem._index is not None mem._index.insert(tr.id, tr.address) for i in range(n_traces, n_total): tr = Trace( id=int(ids[i]), address=HV(bits=addr_bits_all[i], D=D), value=HV(bits=val_bits_all[i], D=D), weight=float(weights[i]), t_insert=float(t_ins[i]), meta=True, tag=None, ) mem._meta_traces.append(tr) mem._ensure_meta_index().insert(tr.id, tr.address) return mem # ----------------------------------------------------------------- chunked I/O def split_file(path: str | Path, chunk_size: int = 4_500_000_000, suffix: str = ".part") -> list[Path]: """Split a large binary file into chunks under ``chunk_size`` bytes. Creates files: ``path.part0``, ``path.part1``, ... Returns list of chunk paths. """ p = Path(path) fsize = p.stat().st_size n_chunks = (fsize + chunk_size - 1) // chunk_size chunks: list[Path] = [] with open(p, "rb") as f: for i in range(n_chunks): chunk_path = p.with_suffix(f"{suffix}{i}") remaining = min(chunk_size, fsize - i * chunk_size) with open(chunk_path, "wb") as out: written = 0 while written < remaining: block = f.read(min(1024 * 1024, remaining - written)) if not block: break out.write(block) written += len(block) chunks.append(chunk_path) print(f" {chunk_path.name}: {chunk_path.stat().st_size / 1024**3:.2f} GB") return chunks def merge_files(chunk_paths: list[str | Path], output_path: str | Path) -> Path: """Merge chunk files back into a single file. Parameters ---------- chunk_paths : list Ordered list of chunk file paths (part0, part1, ...). output_path : str | Path Where to write the merged file. """ out = Path(output_path) out.parent.mkdir(parents=True, exist_ok=True) with open(out, "wb") as f: for cp in chunk_paths: cp = Path(cp) with open(cp, "rb") as chunk: while True: block = chunk.read(1024 * 1024) if not block: break f.write(block) return out def find_chunks(directory: str | Path, stem: str, suffix: str = ".part") -> list[Path]: """Find all chunk files for a given stem in a directory. Returns them sorted by part number. """ d = Path(directory) chunks = sorted(d.glob(f"{stem}{suffix}*")) # Sort numerically by part index def _part_idx(p: Path) -> int: try: return int(p.name.rsplit(suffix, 1)[1]) except (ValueError, IndexError): return 0 return sorted(chunks, key=_part_idx) def load_memory_chunked(directory: str | Path, base_name: str = "palimpseste_memory", rng: np.random.Generator | None = None) -> Memory: """Load memory from chunked files. If ``directory/base_name.bin`` exists, loads directly. Otherwise, looks for ``directory/base_name.bin.part0``, ``.part1``, etc., merges them to a temp file, then loads. """ d = Path(directory) direct = d / f"{base_name}.bin" if direct.exists(): return load_memory(direct, rng=rng) # Find chunks chunks = find_chunks(d, f"{base_name}.bin") if not chunks: # Try without .bin extension chunks = find_chunks(d, base_name) if not chunks: raise FileNotFoundError(f"No memory file or chunks found in {d} for {base_name}") print(f"Merging {len(chunks)} chunks...", flush=True) import tempfile with tempfile.NamedTemporaryFile(suffix=".bin", delete=False) as tmp: tmp_path = Path(tmp.name) merge_files(chunks, tmp_path) mem = load_memory(tmp_path, rng=rng) tmp_path.unlink() # cleanup temp return mem # ----------------------------------------------------------------- encoder def save_encoder(enc: Encoder, path: str | Path) -> None: """Serialize an :class:`Encoder` to JSON index + packed-bits blob.""" p = Path(path) p.parent.mkdir(parents=True, exist_ok=True) # Index: key -> row in the .bin blob atoms_keys = list(enc._atoms.keys()) roles_keys = list(enc._roles.keys()) header = { "D": enc.D, "n_atoms": len(atoms_keys), "n_roles": len(roles_keys), "n_levels": enc._n_levels, "has_levels": enc._levels is not None, "atoms": [_key_to_json(k) for k in atoms_keys], "roles": [_key_to_json(k) for k in roles_keys], "version": 1, } with open(p, "w", encoding="utf-8") as f: json.dump(header, f, indent=2, ensure_ascii=False) # Pack all HVs into one blob: atoms, then roles, then levels packed_len = (enc.D + 7) // 8 rows: list[np.ndarray] = [] for k in atoms_keys: rows.append(enc._atoms[k].bits) for k in roles_keys: rows.append(enc._roles[k].bits) if enc._levels is not None: for lv in enc._levels: rows.append(lv.bits) if rows: blob = np.stack(rows) # (N, packed_len) uint8 else: blob = np.zeros((0, packed_len), dtype=np.uint8) np.save(p.with_name(p.stem + ".bin.npy"), blob) def load_encoder(path: str | Path, rng: np.random.Generator | None = None) -> Encoder: p = Path(path) with open(p, "r", encoding="utf-8") as f: header = json.load(f) D = header["D"] enc = Encoder(D=D, rng=rng if rng else np.random.default_rng()) enc._n_levels = header["n_levels"] blob = np.load(p.with_name(p.stem + ".bin.npy")) packed_len = (D + 7) // 8 row = 0 for i in range(header["n_atoms"]): key = _key_from_json(header["atoms"][i]) enc._atoms[key] = HV(bits=blob[row].copy(), D=D) row += 1 for i in range(header["n_roles"]): key = _key_from_json(header["roles"][i]) enc._roles[key] = HV(bits=blob[row].copy(), D=D) row += 1 if header["has_levels"]: levels = [] for _ in range(header["n_levels"]): levels.append(HV(bits=blob[row].copy(), D=D)) row += 1 enc._levels = levels return enc # ----------------------------------------------------------------- helpers def _key_to_json(k: object) -> list: """JSON-safe encoding of an encoder key. Encoder keys are either: - tuples like ('i', 5), ('s', 'cat'), ('b', True), ('__char__a') - plain ints (role positions: 0, 1, 2, ...) """ if isinstance(k, tuple): # encode the elements: first element is a str tag, rest are scalars elems = [k[0]] + [e for e in k[1:]] return ["t"] + [_scalar_to_json(e) for e in elems] if isinstance(k, (int, np.integer)): return ["i", int(k)] return ["s", str(k)] def _key_from_json(j: list) -> object: if j[0] == "t": elems = [_scalar_from_json(e) for e in j[1:]] return tuple(elems) if j[0] == "i": return int(j[1]) return j[1] # "s" -> str def _scalar_to_json(e: object) -> object: if isinstance(e, (int, np.integer)): return {"i": int(e)} if isinstance(e, (float, np.floating)): return {"f": float(e)} if isinstance(e, bool): return {"b": e} return {"s": str(e)} def _scalar_from_json(e: object) -> object: if isinstance(e, dict): if "i" in e: return int(e["i"]) if "f" in e: return float(e["f"]) if "b" in e: return bool(e["b"]) if "s" in e: return str(e["s"]) return e