Download holo_program.py from zeechimp/holo-program: direct link, hf CLI and curl.
- Browser
- Download file 26.7 kB
-
https://huggingface.co/zeechimp/holo-program/resolve/main/holo_program.py
- Command line
-
hf download hf://zeechimp/holo-program/holo_program.py
-
curl -L -o holo_program.py https://huggingface.co/zeechimp/holo-program/resolve/main/holo_program.py
26.7 kB
| """ | |
| holo-program | |
| ============ | |
| A memory where programs are first-class items, membranes are nested | |
| regions, and writes can be cryptographically keyed. | |
| Three layers, one substrate: | |
| Program layer Operations are stored as key-value pairs: | |
| key = random vector derived from (op, in) | |
| value = bind(out, out_val) | |
| Execution is unbind-then-classify. Chaining is | |
| composition. | |
| Membrane layer Regions have parents. Reads inherit from child | |
| to parent. Writes isolate to the innermost | |
| membrane. Dissolution merges a child into its | |
| parent (P-systems operation). Movement transfers | |
| an object between membranes. | |
| Cryptographic layer | |
| Item vectors are derived from a key. Without | |
| the key, a trace is noise. A commitment scheme | |
| lets a store prove that an item was written | |
| without revealing the item. | |
| Everything is addressable through `ask()`. | |
| Requires: numpy. | |
| """ | |
| from __future__ import annotations | |
| import argparse | |
| import json | |
| import os | |
| import zlib | |
| from typing import Any, Dict, List, Optional, Tuple | |
| import numpy as np | |
| D_DEFAULT = 2048 | |
| # ============================================================ | |
| # Primitives | |
| # ============================================================ | |
| def rvec(seed: int, d: int = D_DEFAULT) -> np.ndarray: | |
| r = np.random.default_rng(seed) | |
| return np.exp(1j * r.uniform(0, 2 * np.pi, d)) | |
| def povec(seed: int, d: int = D_DEFAULT) -> np.ndarray: | |
| v = rvec(seed, d) | |
| F = np.fft.fft(v) | |
| return np.fft.ifft(F / np.abs(F)) | |
| def bind(a: np.ndarray, b: np.ndarray) -> np.ndarray: | |
| return np.fft.ifft(np.fft.fft(a) * np.fft.fft(b)) | |
| def unbind(a: np.ndarray, c: np.ndarray) -> np.ndarray: | |
| return np.fft.ifft(np.fft.fft(c) * np.conj(np.fft.fft(a))) | |
| def norm(v: np.ndarray) -> np.ndarray: | |
| F = np.fft.fft(v) | |
| return np.fft.ifft(F / (np.abs(F) + 1e-12)) | |
| def cos(a: np.ndarray, b: np.ndarray) -> float: | |
| na = np.linalg.norm(a) | |
| nb = np.linalg.norm(b) | |
| if na < 1e-12 or nb < 1e-12: | |
| return 0.0 | |
| return float(np.real(np.vdot(a, b)) / (na * nb)) | |
| def project(u: np.ndarray, v: np.ndarray) -> float: | |
| nv2 = float(np.real(np.vdot(v, v))) | |
| if nv2 < 1e-12: | |
| return 0.0 | |
| return float(np.real(np.vdot(u, v)) / nv2) | |
| def cos_batch(u: np.ndarray, X: np.ndarray) -> np.ndarray: | |
| X = X.reshape(len(X), -1) | |
| u = u.ravel() | |
| return np.real(X @ np.conj(u)) / ( | |
| np.linalg.norm(X, axis=1) * np.linalg.norm(u) + 1e-12) | |
| # ============================================================ | |
| # Cryptographic layer | |
| # ============================================================ | |
| def keyed_seed(key: Optional[str], label: str) -> int: | |
| s = f"{key}::{label}" if key is not None else f"_::{label}" | |
| return zlib.crc32(s.encode()) & 0xffffffff | |
| def make_vector(key: Optional[str], label: str, d: int) -> np.ndarray: | |
| return povec(keyed_seed(key, label), d) | |
| # ============================================================ | |
| # Global role vectors (cached per dimension) | |
| # ============================================================ | |
| _ROLES_CACHE: Dict[int, Dict[str, np.ndarray]] = {} | |
| def get_roles(d: int) -> Dict[str, np.ndarray]: | |
| if d not in _ROLES_CACHE: | |
| _ROLES_CACHE[d] = { | |
| "op": povec(1000, d), | |
| "in": povec(1001, d), | |
| "out": povec(1002, d), | |
| "salt": povec(1003, d), | |
| } | |
| return _ROLES_CACHE[d] | |
| # ============================================================ | |
| # Membrane | |
| # ============================================================ | |
| class Membrane: | |
| """ | |
| A region of the substrate with an optional parent. | |
| """ | |
| def __init__(self, name: str, d: int, key: Optional[str] = None, | |
| parent: Optional["Membrane"] = None, | |
| threshold: float = 0.05): | |
| self.name = name | |
| self.d = d | |
| self.key = key | |
| self.parent = parent | |
| self.threshold = threshold | |
| self.roles = get_roles(d) | |
| self.trace = np.zeros(d, dtype=complex) | |
| self.ops_trace = np.zeros(d, dtype=complex) | |
| self.weights: Dict[str, float] = {} | |
| self.labels: set = set() | |
| self.operations: List[Tuple[str, str, str]] = [] | |
| def _vec(self, label: str) -> np.ndarray: | |
| return make_vector(self.key, label, self.d) | |
| def _op_key(self, op_name: str, in_val: str) -> np.ndarray: | |
| """ | |
| Random key for an operation, derived from the pair. | |
| Fully random: no two distinct pairs share a component, so | |
| cross-talk between operations with the same op name is | |
| negligible. Same computation in add_op and apply. | |
| """ | |
| pair_id = f"__op__{op_name}::{in_val}" | |
| return self._vec(pair_id) | |
| # ---------------- items ---------------- | |
| def store(self, label: str, weight: float = 1.0) -> None: | |
| v = self._vec(label) | |
| self.trace += weight * bind(v, v) | |
| self.weights[label] = self.weights.get(label, 0.0) + weight | |
| self.labels.add(label) | |
| def similarity(self, label: str) -> float: | |
| v = self._vec(label) | |
| u = unbind(v, self.trace) | |
| return project(u, v) | |
| def query_inherited(self, label: str) -> float: | |
| sim = self.similarity(label) | |
| if sim >= self.threshold: | |
| return sim | |
| if self.parent is not None: | |
| return self.parent.query_inherited(label) | |
| return sim | |
| # ---------------- operations ---------------- | |
| def add_op(self, op_name: str, in_val: str, out_val: str) -> None: | |
| key = self._op_key(op_name, in_val) | |
| value = bind(self.roles["out"], self._vec(out_val)) | |
| self.ops_trace += bind(key, value) | |
| self.operations.append((op_name, in_val, out_val)) | |
| def apply(self, op_name: str, in_val: str) -> Optional[str]: | |
| if not self.operations: | |
| return None | |
| key = self._op_key(op_name, in_val) | |
| u = unbind(key, self.ops_trace) | |
| out = unbind(self.roles["out"], u) | |
| candidates = sorted({o[2] for o in self.operations}) | |
| stack = np.stack([self._vec(c) for c in candidates]) | |
| sims = np.abs(cos_batch(out, stack)) | |
| best = int(np.argmax(sims)) | |
| if sims[best] < self.threshold: | |
| return None | |
| return candidates[best] | |
| def find_applicable(self, in_val: str | |
| ) -> List[Tuple[str, str, float]]: | |
| hits: List[Tuple[str, str, float]] = [] | |
| for op_name, in_v, out_v in self.operations: | |
| if in_v != in_val: | |
| continue | |
| key = self._op_key(op_name, in_val) | |
| u = unbind(key, self.ops_trace) | |
| out = unbind(self.roles["out"], u) | |
| sim = project(out, self._vec(out_v)) | |
| if sim >= self.threshold: | |
| hits.append((op_name, out_v, float(sim))) | |
| hits.sort(key=lambda h: -h[2]) | |
| return hits | |
| # ---------------- cryptographic ---------------- | |
| def commit(self, label: str, salt: str) -> np.ndarray: | |
| v_label = self._vec(label) | |
| v_salt = self._vec(salt) | |
| return bind(v_label, v_salt) | |
| def verify(self, label: str, salt: str, | |
| commitment: np.ndarray) -> bool: | |
| expected = self.commit(label, salt) | |
| return cos(commitment, expected) > 0.99 | |
| # ---------------- diagnostics ---------------- | |
| def stats(self) -> Dict: | |
| return { | |
| "name": self.name, | |
| "d": self.d, | |
| "has_key": self.key is not None, | |
| "n_labels": len(self.labels), | |
| "n_operations": len(self.operations), | |
| "trace_magnitude": float(np.linalg.norm(self.trace)), | |
| "ops_magnitude": float(np.linalg.norm(self.ops_trace)), | |
| } | |
| # ============================================================ | |
| # HoloProgram | |
| # ============================================================ | |
| class HoloProgram: | |
| def __init__(self, d: int = D_DEFAULT, | |
| key: Optional[str] = None, | |
| threshold: float = 0.05): | |
| self.d = d | |
| self.key = key | |
| self.threshold = threshold | |
| self.membranes: Dict[str, Membrane] = {} | |
| self.add_membrane("root", parent=None) | |
| def add_membrane(self, name: str, | |
| parent: Optional[str] = "root") -> None: | |
| if name in self.membranes: | |
| return | |
| parent_m = self.membranes.get(parent) if parent else None | |
| self.membranes[name] = Membrane( | |
| name=name, d=self.d, key=self.key, | |
| parent=parent_m, threshold=self.threshold, | |
| ) | |
| def store(self, membrane: str, label: str, | |
| weight: float = 1.0) -> None: | |
| self.membranes[membrane].store(label, weight=weight) | |
| def add_op(self, membrane: str, op_name: str, | |
| in_val: str, out_val: str) -> None: | |
| self.membranes[membrane].add_op(op_name, in_val, out_val) | |
| def query(self, membrane: str, label: str) -> float: | |
| return self.membranes[membrane].query_inherited(label) | |
| def apply(self, membrane: str, op_name: str, | |
| in_val: str) -> Optional[str]: | |
| return self.membranes[membrane].apply(op_name, in_val) | |
| def run_chain(self, membrane: str, ops: List[str], | |
| start: str) -> Dict: | |
| cur = start | |
| path = [cur] | |
| for op in ops: | |
| nxt = self.apply(membrane, op, cur) | |
| if nxt is None: | |
| return {"path": path, "ok": False, "failed_at": op} | |
| path.append(nxt) | |
| cur = nxt | |
| return {"path": path, "ok": True} | |
| def run_data_driven(self, membrane: str, start: str, | |
| max_steps: int = 20) -> Dict: | |
| cur = start | |
| path = [cur] | |
| for _ in range(max_steps): | |
| hits = self.membranes[membrane].find_applicable(cur) | |
| if not hits: | |
| break | |
| op_name, out_val, sim = hits[0] | |
| path.append(out_val) | |
| cur = out_val | |
| return {"path": path, "steps": len(path) - 1} | |
| def commit(self, membrane: str, label: str, | |
| salt: str) -> np.ndarray: | |
| return self.membranes[membrane].commit(label, salt) | |
| def verify(self, membrane: str, label: str, salt: str, | |
| commitment: np.ndarray) -> bool: | |
| return self.membranes[membrane].verify(label, salt, commitment) | |
| def move(self, label: str, src: str, dst: str) -> bool: | |
| s = self.membranes[src] | |
| d = self.membranes[dst] | |
| if label not in s.labels: | |
| return False | |
| w = s.weights.get(label, 1.0) | |
| if w <= 0: | |
| return False | |
| v = s._vec(label) | |
| s.trace -= w * bind(v, v) | |
| s.weights[label] = 0.0 | |
| s.labels.discard(label) | |
| d.store(label, weight=w) | |
| return True | |
| def dissolve(self, child: str, into: str) -> None: | |
| c = self.membranes[child] | |
| t = self.membranes[into] | |
| t.trace = t.trace + c.trace | |
| t.ops_trace = t.ops_trace + c.ops_trace | |
| for lbl, w in c.weights.items(): | |
| t.weights[lbl] = t.weights.get(lbl, 0.0) + w | |
| t.labels.add(lbl) | |
| t.operations.extend(c.operations) | |
| for m in self.membranes.values(): | |
| if m.parent is c: | |
| m.parent = t | |
| del self.membranes[child] | |
| def ask(self, query: str, top_k: int = 8) -> List[Dict]: | |
| hits: List[Dict] = [] | |
| for m_name, mem in self.membranes.items(): | |
| if query in mem.labels: | |
| sim = mem.query_inherited(query) | |
| hits.append({ | |
| "type": "item", | |
| "membrane": m_name, | |
| "label": query, | |
| "similarity": sim, | |
| }) | |
| for m_name, mem in self.membranes.items(): | |
| for op_name, in_v, out_v in mem.operations: | |
| if query in (op_name, in_v, out_v): | |
| role = ("op" if query == op_name else | |
| "in" if query == in_v else "out") | |
| hits.append({ | |
| "type": "operation", | |
| "membrane": m_name, | |
| "role": role, | |
| "op": op_name, | |
| "in": in_v, | |
| "out": out_v, | |
| }) | |
| if query in self.membranes: | |
| hits.append({ | |
| "type": "membrane", | |
| "membrane": query, | |
| }) | |
| def sort_key(h): | |
| if h["type"] == "item": | |
| return (0, -h["similarity"]) | |
| return (1, 0) | |
| hits.sort(key=sort_key) | |
| return hits[:top_k] | |
| def stats(self) -> Dict: | |
| return { | |
| "d": self.d, | |
| "has_key": self.key is not None, | |
| "threshold": self.threshold, | |
| "n_membranes": len(self.membranes), | |
| "membranes": {n: m.stats() | |
| for n, m in self.membranes.items()}, | |
| } | |
| def save_json(self, path: str) -> None: | |
| out = self.stats() | |
| out["membrane_detail"] = { | |
| n: { | |
| "labels": sorted(m.labels), | |
| "operations": [ | |
| {"op": o, "in": i, "out": r} | |
| for o, i, r in m.operations | |
| ], | |
| "parent": m.parent.name if m.parent else None, | |
| } | |
| for n, m in self.membranes.items() | |
| } | |
| with open(path, "w") as f: | |
| json.dump(out, f, indent=2) | |
| print(f" saved: {path}") | |
| # ============================================================ | |
| # Helpers | |
| # ============================================================ | |
| def section(title: str) -> None: | |
| print() | |
| print("=" * 78) | |
| print(title) | |
| print("=" * 78) | |
| def self_test() -> bool: | |
| print(" self-test:") | |
| d = 2048 | |
| a = povec(1, d) | |
| b = povec(2, d) | |
| ok1 = cos(unbind(a, bind(a, b)), b) > 0.99 | |
| print(f" bind/unbind identity: {'PASS' if ok1 else 'FAIL'}") | |
| v1 = make_vector("key1", "label", d) | |
| v2 = make_vector("key2", "label", d) | |
| ok2 = cos(v1, v2) < 0.1 | |
| print(f" keys separate vectors: {'PASS' if ok2 else 'FAIL'}") | |
| bb = 0.7 * bind(a, a) | |
| recovered = project(unbind(a, bb), a) | |
| ok3 = abs(recovered - 0.7) < 0.02 | |
| print(f" raw projection: {'PASS' if ok3 else 'FAIL'}") | |
| # Fully random key-value storage: many pairs sharing nothing | |
| r = get_roles(d) | |
| trace = np.zeros(d, dtype=complex) | |
| pairs = [] | |
| for i in range(50): | |
| key = norm(bind(r["op"], povec(2000 + i, d))) | |
| val = povec(3000 + i, d) | |
| trace += bind(key, val) | |
| pairs.append((key, val)) | |
| correct = 0 | |
| for key, val in pairs: | |
| rec = unbind(key, trace) | |
| if cos(rec, val) > cos(rec, pairs[0][1]): | |
| correct += 1 | |
| ok4 = correct >= 48 | |
| print(f" random-key key-value: " | |
| f"{'PASS' if ok4 else 'FAIL'} ({correct}/50)") | |
| return ok1 and ok2 and ok3 and ok4 | |
| # ============================================================ | |
| # DEMO 1. Program execution (chains) | |
| # ============================================================ | |
| def demo_program_chains(): | |
| section("DEMO 1. PROGRAM EXECUTION — chained operations") | |
| prog = HoloProgram(d=2048) | |
| for i in range(20): | |
| prog.add_op("root", "inc", str(i), str(i + 1)) | |
| for i in range(20): | |
| prog.add_op("root", "double", str(i), str(2 * i)) | |
| for i in range(10): | |
| prog.add_op("root", "square", str(i), str(i * i)) | |
| print(" stored: 20 inc, 20 double, 10 square operations") | |
| print() | |
| chains = [ | |
| (["inc", "inc", "double"], "3"), | |
| (["square", "inc"], "4"), | |
| (["double", "inc", "inc"], "5"), | |
| (["inc", "square"], "2"), | |
| ] | |
| for ops, start in chains: | |
| result = prog.run_chain("root", ops, start) | |
| path_str = " -> ".join(result["path"]) | |
| ok = "OK" if result["ok"] else f"FAILED at {result['failed_at']}" | |
| print(f" [{', '.join(ops)}] from '{start}': {path_str} {ok}") | |
| print() | |
| print(" single apply:") | |
| for op_name, in_val in [("inc", "7"), ("double", "7"), ("square", "7")]: | |
| out = prog.apply("root", op_name, in_val) | |
| print(f" {op_name}({in_val}) -> {out}") | |
| # ============================================================ | |
| # DEMO 2. Data-driven execution | |
| # ============================================================ | |
| def demo_data_driven(): | |
| section("DEMO 2. DATA-DRIVEN EXECUTION — apply any applicable op") | |
| prog = HoloProgram(d=2048) | |
| transitions = [ | |
| ("t1", "cold", "warm"), | |
| ("t2", "warm", "hot"), | |
| ("t3", "hot", "boiling"), | |
| ("t4", "boiling", "evaporated"), | |
| ] | |
| for op, src, dst in transitions: | |
| prog.add_op("root", op, src, dst) | |
| result = prog.run_data_driven("root", "cold", max_steps=10) | |
| print(f" start at 'cold', walk using whichever op matches:") | |
| print(f" {' -> '.join(result['path'])}") | |
| print(f" steps: {result['steps']}") | |
| print() | |
| result = prog.run_data_driven("root", "evaporated", max_steps=5) | |
| print(f" start at 'evaporated' (no matching op):") | |
| print(f" {' -> '.join(result['path'])}") | |
| print(f" steps: {result['steps']}") | |
| # ============================================================ | |
| # DEMO 3. Membranes with inheritance | |
| # ============================================================ | |
| def demo_membranes(): | |
| section("DEMO 3. MEMBRANES — nested regions with inheritance") | |
| prog = HoloProgram(d=2048) | |
| prog.add_membrane("alice", parent="root") | |
| prog.add_membrane("alice_kitchen", parent="alice") | |
| prog.store("root", "earth_round") | |
| prog.store("root", "water_wet") | |
| prog.store("alice", "alice_lives_in_paris") | |
| prog.store("alice", "alice_works_at_cafe") | |
| prog.store("alice_kitchen", "kitchen_floor_5") | |
| prog.store("alice_kitchen", "manager_is_bob") | |
| print(" membranes: root <- alice <- alice_kitchen") | |
| print() | |
| print(f" {'query':>24} {'root':>8} {'alice':>8} {'kitchen':>8}") | |
| for q in ["earth_round", "alice_lives_in_paris", | |
| "kitchen_floor_5", "manager_is_bob"]: | |
| r_root = prog.query("root", q) | |
| r_alice = prog.query("alice", q) | |
| r_kit = prog.query("alice_kitchen", q) | |
| print(f" {q:>24} {r_root:>+8.3f} {r_alice:>+8.3f} {r_kit:>+8.3f}") | |
| # ============================================================ | |
| # DEMO 4. Communication and dissolution | |
| # ============================================================ | |
| def demo_communication(): | |
| section("DEMO 4. COMMUNICATION AND DISSOLUTION") | |
| prog = HoloProgram(d=2048) | |
| prog.add_membrane("inner", parent="root") | |
| prog.store("root", "common_knowledge") | |
| prog.store("inner", "private_note") | |
| print(" before move:") | |
| print(f" root sees private_note: {prog.query('root', 'private_note'):+.4f}") | |
| print(f" inner sees private_note: {prog.query('inner', 'private_note'):+.4f}") | |
| prog.move("private_note", "inner", "root") | |
| print() | |
| print(" after move (inner -> root):") | |
| print(f" root sees private_note: {prog.query('root', 'private_note'):+.4f}") | |
| print(f" inner sees private_note: {prog.query('inner', 'private_note'):+.4f}") | |
| prog.add_membrane("middle", parent="root") | |
| prog.store("middle", "middle_fact") | |
| print() | |
| print(" before dissolution:") | |
| print(f" root sees middle_fact: {prog.query('root', 'middle_fact'):+.4f}") | |
| print(f" middle sees middle_fact: {prog.query('middle', 'middle_fact'):+.4f}") | |
| prog.dissolve("middle", "root") | |
| print() | |
| print(" after dissolution (middle -> root):") | |
| print(f" root sees middle_fact: {prog.query('root', 'middle_fact'):+.4f}") | |
| print(f" membranes remaining: {list(prog.membranes.keys())}") | |
| # ============================================================ | |
| # DEMO 5. Cryptographic layer | |
| # ============================================================ | |
| def demo_crypto(): | |
| section("DEMO 5. CRYPTOGRAPHIC LAYER — keyed codebook") | |
| print(" Store items with key 'secret_A'. Copy the trace to two") | |
| print(" other programs: one with the same key, one with a wrong") | |
| print(" key. Query both.") | |
| print() | |
| p_correct = HoloProgram(d=2048, key="secret_A") | |
| p_correct.store("root", "classified_meeting_place") | |
| p_correct.store("root", "classified_meeting_time") | |
| p_wrong = HoloProgram(d=2048, key="wrong_key") | |
| p_wrong.membranes["root"].trace = \ | |
| p_correct.membranes["root"].trace.copy() | |
| candidates = ["classified_meeting_place", "classified_meeting_time", | |
| "paris", "3pm", "unrelated_word"] | |
| print(f" {'candidate':>30} {'correct key':>14} {'wrong key':>12}") | |
| for c in candidates: | |
| sim_c = p_correct.query("root", c) | |
| sim_w = p_wrong.query("root", c) | |
| print(f" {c:>30} {sim_c:>+14.4f} {sim_w:>+12.4f}") | |
| # ============================================================ | |
| # DEMO 6. Commitment scheme | |
| # ============================================================ | |
| def demo_commit(): | |
| section("DEMO 6. COMMITMENT SCHEME") | |
| prog = HoloProgram(d=2048, key="K") | |
| prog.store("root", "secret_value") | |
| commitment = prog.commit("root", "secret_value", salt="random_salt_123") | |
| print(f" commitment vector shape: {commitment.shape}") | |
| print(f" commitment magnitude: {np.linalg.norm(commitment):.4f}") | |
| print() | |
| print(" verification:") | |
| ok1 = prog.verify("root", "secret_value", "random_salt_123", commitment) | |
| ok2 = prog.verify("root", "secret_value", "wrong_salt", commitment) | |
| ok3 = prog.verify("root", "different_value", "random_salt_123", commitment) | |
| print(f" correct (label, salt): {ok1}") | |
| print(f" wrong salt: {ok2}") | |
| print(f" wrong label: {ok3}") | |
| # ============================================================ | |
| # DEMO 7. Unified ask() | |
| # ============================================================ | |
| def demo_ask(): | |
| section("DEMO 7. UNIFIED ask() — non-flat interface") | |
| prog = HoloProgram(d=2048) | |
| prog.add_membrane("alice", parent="root") | |
| prog.add_membrane("alice_kitchen", parent="alice") | |
| prog.store("root", "earth_round") | |
| prog.store("root", "water_wet") | |
| prog.store("alice", "alice_lives_in_paris") | |
| prog.store("alice", "alice_works_at_cafe") | |
| prog.store("alice_kitchen", "kitchen_floor_5") | |
| prog.store("alice_kitchen", "manager_is_bob") | |
| prog.add_op("root", "serve", "customer", "coffee") | |
| prog.add_op("root", "greet", "customer", "hello") | |
| prog.add_op("alice_kitchen", "brew", "coffee", "ready") | |
| for query in ["alice", "alice_lives_in_paris", "coffee", | |
| "alice_kitchen", "manager_is_bob"]: | |
| hits = prog.ask(query, top_k=6) | |
| print(f" ask('{query}'):") | |
| if not hits: | |
| print(f" (no hits)") | |
| for h in hits: | |
| if h["type"] == "item": | |
| print(f" [item] membrane={h['membrane']:<14} " | |
| f"sim={h['similarity']:+.3f}") | |
| elif h["type"] == "operation": | |
| print(f" [operation] membrane={h['membrane']:<14} " | |
| f"role={h['role']:<4} " | |
| f"op={h['op']}({h['in']}) -> {h['out']}") | |
| elif h["type"] == "membrane": | |
| print(f" [membrane] {h['membrane']}") | |
| print() | |
| # ============================================================ | |
| # DEMO 8. Programs as items | |
| # ============================================================ | |
| def demo_program_as_item(): | |
| section("DEMO 8. PROGRAMS AS ITEMS") | |
| prog = HoloProgram(d=2048) | |
| prog.add_op("root", "inc", "1", "2") | |
| prog.add_op("root", "inc", "2", "3") | |
| prog.add_op("root", "double", "3", "6") | |
| print(" The substrate holds operations alongside items.") | |
| print(" Query for 'inc' through ask():") | |
| hits = prog.ask("inc", top_k=10) | |
| for h in hits: | |
| if h["type"] == "operation": | |
| print(f" {h['op']}({h['in']}) -> {h['out']} " | |
| f"(membrane={h['membrane']})") | |
| print() | |
| print(" Query for '3' (input and output of operations):") | |
| hits = prog.ask("3", top_k=10) | |
| for h in hits: | |
| if h["type"] == "operation": | |
| print(f" {h['op']}({h['in']}) -> {h['out']} " | |
| f"role={h['role']}") | |
| # ============================================================ | |
| # Main | |
| # ============================================================ | |
| def run_all_demos(output_dir: str = "holo_program_out") -> None: | |
| os.makedirs(output_dir, exist_ok=True) | |
| print("=" * 78) | |
| print("HOLO-PROGRAM") | |
| print("Programs as items, membranes as regions, keys as gates.") | |
| print("=" * 78) | |
| demos = [ | |
| demo_program_chains, | |
| demo_data_driven, | |
| demo_membranes, | |
| demo_communication, | |
| demo_crypto, | |
| demo_commit, | |
| demo_ask, | |
| demo_program_as_item, | |
| ] | |
| for d in demos: | |
| try: | |
| d() | |
| except Exception as e: | |
| print(f" EXCEPTION in {d.__name__}: {e}") | |
| section("OUTPUT") | |
| prog = HoloProgram(d=2048, key="demo_key") | |
| prog.add_membrane("alice", parent="root") | |
| prog.store("root", "earth_round") | |
| prog.store("alice", "alice_lives_in_paris") | |
| prog.add_op("root", "inc", "1", "2") | |
| prog.save_json(os.path.join(output_dir, "state.json")) | |
| print() | |
| print("=" * 78) | |
| print(f"All outputs saved to: {output_dir}/") | |
| print("=" * 78) | |
| def main() -> None: | |
| parser = argparse.ArgumentParser( | |
| description="Programs as items, membranes as regions, " | |
| "keys as gates.", | |
| formatter_class=argparse.RawDescriptionHelpFormatter, | |
| epilog=""" | |
| Library usage: | |
| from holo_program import HoloProgram | |
| prog = HoloProgram(d=2048, key=None) | |
| prog.add_membrane("alice", parent="root") | |
| prog.store("root", "apple") | |
| prog.add_op("root", "inc", "0", "1") | |
| prog.run_chain("root", ["inc", "inc"], "0") | |
| """) | |
| parser.add_argument("--output", default="holo_program_out", | |
| help="output directory (default: holo_program_out)") | |
| args = parser.parse_args() | |
| print("=" * 78) | |
| print("SELF-TEST") | |
| print("=" * 78) | |
| if not self_test(): | |
| print(" primitive test failed; continuing anyway") | |
| run_all_demos(output_dir=args.output) | |
| if __name__ == "__main__": | |
| main() |