File size: 11,298 Bytes
ad474c9 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 | """Content-addressable hyperdimensional storage.
A cell is one hypervector of ``d`` bits, identified by a *name*. The address
of a name is itself a hypervector (``hvc("cell/" + name)``), so the store has
no linear address bus: you hand it a vector and it answers with the cell whose
address vector it most resembles. Exact-name access is served by a dictionary
index to keep the VM deterministic; near-miss access falls through to a
similarity scan, which is the behaviour that makes it content-addressable.
Two structures live here:
``cells`` the data store — program instructions, register shadows, tool
records, structure records, all the same kind of object
``cleanup`` a symbol table mapping hypervectors back to names, which is what
turns a noisy decoded field into an opcode / register / literal
The file format (``.hdc``) is deliberately boring and byte-stable: a header,
the packed bit-vectors, then the symbol table. Vectors are regenerated from
names where they are symbolic, so an image written on one machine loads on
another at the same ``d``.
"""
from __future__ import annotations
import os
import struct
from typing import Iterable, Iterator
from .hypervec import DEFAULT_DIM, hvc, mask, popcount, sim
MAGIC = b"HDCV1\x00"
#: A fuzzy address match must beat this to resolve without an exact hit.
FUZZY_FLOOR = 0.20
def _nbytes(d: int) -> int:
return (d + 7) // 8
def pack(hv: int, d: int = DEFAULT_DIM) -> bytes:
"""Pack a hypervector into fixed-width little-endian bytes."""
return (hv & mask(d)).to_bytes(_nbytes(d), "little")
def unpack(raw: bytes, d: int = DEFAULT_DIM) -> int:
"""Inverse of :func:`pack`."""
return int.from_bytes(raw, "little") & mask(d)
class HDCStorage:
"""A hyperdimensional store: named cells plus a cleanup symbol table."""
def __init__(self, d: int = DEFAULT_DIM, capacity: int = 8192, label: str = "hdc0"):
self.d = d
self.capacity = capacity
self.label = label
self.cells: dict[str, int] = {}
self.order: list[str] = []
self._by_addr: dict[int, str] = {}
self.cleanup: dict[int, str] = {}
self._cleanup_names: list[str] = []
#: Monotonic counters, surfaced by ``stat`` and the IDE.
self.reads = 0
self.writes = 0
self.fuzzy_reads = 0
# ---------------------------------------------------------------- cells
def addr(self, name: str) -> int:
"""The address hypervector of a cell name."""
return hvc("cell/" + name, self.d)
def write(self, name: str, hv: int, symbol: bool = False) -> None:
"""Store ``hv`` at ``name``. Returns nothing; overwrites in place."""
if name not in self.cells:
if len(self.order) >= self.capacity:
raise MemoryError(
"HDC storage full (%d cells)" % self.capacity
)
self.order.append(name)
self.cells[name] = hv & mask(self.d)
self._by_addr[self.addr(name)] = name
self.writes += 1
if symbol:
self.define(name)
def read(self, name: str, default: int | None = None) -> int:
"""Read a cell by name."""
if name in self.cells:
self.reads += 1
return self.cells[name]
if default is not None:
return default
raise KeyError("no such HDC cell: %r" % (name,))
def remove(self, name: str) -> bool:
"""Delete a cell and its symbol entry; False when it was not present.
A filesystem has to be able to take cells away when a file is deleted,
otherwise ``len(storage)`` grows forever and every stat lies about it.
"""
if name not in self.cells:
return False
del self.cells[name]
try:
self.order.remove(name)
except ValueError:
pass
address = self.addr(name)
self._by_addr.pop(address, None)
self.cleanup.pop(address, None)
try:
self._cleanup_names.remove(name)
except ValueError:
pass
return True
def define(self, name: str) -> int:
"""Add a name to cleanup memory; returns its address vector."""
hv = self.addr(name)
if hv not in self.cleanup:
self.cleanup[hv] = name
self._cleanup_names.append(name)
return hv
def __contains__(self, name: object) -> bool:
return name in self.cells
def __len__(self) -> int:
return len(self.cells)
def __iter__(self) -> Iterator[tuple[str, int]]:
for name in self.order:
yield name, self.cells[name]
# -------------------------------------------------- content addressing
def resolve(self, addr_hv: int, floor: float = FUZZY_FLOOR) -> tuple[str, float] | None:
"""Map an address vector to a cell name.
Exact hit first (confidence 1.0). Otherwise the nearest cell address
by similarity, provided it clears ``floor``. This is the read path
that gives the machine its HDC character: a corrupted address still
lands on the right cell and reports how sure it was.
"""
name = self._by_addr.get(addr_hv & mask(self.d))
if name is not None:
self.reads += 1
return name, 1.0
best_name: str | None = None
best = -2.0
for a, n in self._by_addr.items():
s = sim(a, addr_hv, self.d)
if s > best:
best, best_name = s, n
if best_name is not None and best >= floor:
self.reads += 1
self.fuzzy_reads += 1
return best_name, best
return None
def recall(
self, probe: int, k: int = 5, prefix: str = "", floor: float = 0.0
) -> list[tuple[str, int, float]]:
"""Associative search: the ``k`` cell *values* most similar to ``probe``.
This is the compute-in-memory primitive — no index, no address, the
search happens by comparing against every stored vector.
"""
scored: list[tuple[str, int, float]] = []
for name in self.order:
if prefix and not name.startswith(prefix):
continue
hv = self.cells[name]
s = sim(hv, probe, self.d)
if s >= floor:
scored.append((name, hv, s))
scored.sort(key=lambda t: t[2], reverse=True)
return scored[:k]
def symbols(self) -> list[str]:
"""Cleanup memory contents, in registration order."""
return list(self._cleanup_names)
def nearest_symbol(self, hv: int, k: int = 1, prefix: str = "") -> list[tuple[str, float]]:
"""Cleanup: nearest named symbol(s) for a noisy hypervector."""
scored: list[tuple[str, float]] = []
for name in self._cleanup_names:
if prefix and not name.startswith(prefix):
continue
scored.append((name, sim(hvc("cell/" + name, self.d), hv, self.d)))
scored.sort(key=lambda t: t[1], reverse=True)
return scored[:k]
# -------------------------------------------------------------- chex
def bitplanes(self, name: str, width: int = 64) -> list[str]:
"""Bit planes of a cell, WSB first — the IDE's raw view."""
from .hypervec import bitplanes
return bitplanes(self.read(name), self.d, width)
def stat(self) -> dict:
"""Storage statistics (cheap; used by ``MEMSTAT`` and the IDE)."""
used = len(self.order)
total_bits = used * self.d
return {
"label": self.label,
"d": self.d,
"cells_used": used,
"cells_free": max(0, self.capacity - used),
"bytes": used * (_nbytes(self.d) + 24),
"symbols": len(self.cleanup),
"reads": self.reads,
"writes": self.writes,
"fuzzy_reads": self.fuzzy_reads,
"occupancy": used / self.capacity if self.capacity else 0.0,
"stored_bits": total_bits,
"total_popcount": sum(popcount(v) for v in self.cells.values()),
}
# --------------------------------------------------------- persistence
def save(self, path: str) -> int:
"""Write the whole store as a ``.hdc`` image. Returns bytes written."""
nb = _nbytes(self.d)
with open(path, "wb") as fh:
fh.write(MAGIC)
fh.write(struct.pack("<IIII", self.d, self.capacity, len(self.order), len(self._cleanup_names)))
for name in self.order:
raw = name.encode("utf-8")
fh.write(struct.pack("<H", len(raw)))
fh.write(raw)
fh.write(pack(self.cells[name], self.d))
for name in self._cleanup_names:
raw = name.encode("utf-8")
fh.write(struct.pack("<H", len(raw)))
fh.write(raw)
return os.path.getsize(path)
@classmethod
def load(cls, path: str, label: str | None = None) -> "HDCStorage":
"""Load a ``.hdc`` image."""
with open(path, "rb") as fh:
if fh.read(len(MAGIC)) != MAGIC:
raise ValueError("not an HDC image: %s" % path)
d, capacity, n_cells, n_syms = struct.unpack("<IIII", fh.read(16))
store = cls(d=d, capacity=max(capacity, n_cells), label=label or os.path.basename(path))
nb = _nbytes(d)
for _ in range(n_cells):
(ln,) = struct.unpack("<H", fh.read(2))
name = fh.read(ln).decode("utf-8")
hv = unpack(fh.read(nb), d)
store.write(name, hv)
for _ in range(n_syms):
(ln,) = struct.unpack("<H", fh.read(2))
store.define(fh.read(ln).decode("utf-8"))
return store
# ------------------------------------------------------------- helpers
def clone(self) -> "HDCStorage":
other = HDCStorage(d=self.d, capacity=self.capacity, label=self.label)
for name in self.order:
other.write(name, self.cells[name])
for name in self._cleanup_names:
other.define(name)
return other
def dump(self, limit: int = 24, width: int = 48) -> str:
"""Human-readable listing of the store."""
lines = [
"HDC storage %r d=%d cells=%d/%d symbols=%d"
% (self.label, self.d, len(self.order), self.capacity, len(self.cleanup))
]
for i, name in enumerate(self.order[:limit]):
hv = self.cells[name]
from .hypervec import bits
b = bits(hv, self.d)
lines.append(
" %4d %-22s %s... pop=%d"
% (i, name, b[:width], popcount(hv))
)
if len(self.order) > limit:
lines.append(" ... %d more cells" % (len(self.order) - limit))
return "\n".join(lines)
def storage_from_pairs(pairs: Iterable[tuple[str, int]], d: int = DEFAULT_DIM) -> HDCStorage:
"""Build a store from ``(name, hypervector)`` pairs."""
items = list(pairs)
store = HDCStorage(d=d, capacity=max(64, len(items)))
for name, hv in items:
store.write(name, hv)
return store
|