| """Tests for content-addressable HDC storage and its file format.""" |
|
|
| import os |
| import random |
| import sys |
| import tempfile |
| import unittest |
|
|
| sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) |
|
|
| from hdc.hypervec import hvc, random_hv, sim |
| from hdc.memory import HDCStorage, pack, unpack |
|
|
| D = 2000 |
|
|
|
|
| class TestCells(unittest.TestCase): |
| def test_write_read(self): |
| s = HDCStorage(d=D) |
| hv = random_hv(D, random.Random(1)) |
| s.write("a", hv) |
| self.assertEqual(s.read("a"), hv) |
| self.assertIn("a", s) |
| self.assertEqual(len(s), 1) |
|
|
| def test_overwrite_does_not_grow(self): |
| s = HDCStorage(d=D) |
| s.write("a", 1) |
| s.write("a", 2) |
| self.assertEqual(len(s), 1) |
| self.assertEqual(s.read("a"), 2) |
|
|
| def test_missing_cell(self): |
| s = HDCStorage(d=D) |
| with self.assertRaises(KeyError): |
| s.read("nope") |
| self.assertEqual(s.read("nope", default=7), 7) |
|
|
| def test_capacity_overflow(self): |
| s = HDCStorage(d=200, capacity=2) |
| s.write("x", 1) |
| s.write("y", 2) |
| with self.assertRaises(MemoryError): |
| s.write("z", 3) |
|
|
| def test_pack_unpack(self): |
| s = random_hv(D, random.Random(2)) |
| self.assertEqual(unpack(pack(s, D), D), s) |
| self.assertEqual(len(pack(s, D)), (D + 7) // 8) |
|
|
|
|
| class TestCleanup(unittest.TestCase): |
| def test_nearest_symbol_exact(self): |
| s = HDCStorage(d=D) |
| s.write("alpha", 1, symbol=True) |
| s.write("beta", 2, symbol=True) |
| self.assertEqual(s.symbols(), ["alpha", "beta"]) |
| hits = s.nearest_symbol(s.addr("beta"), k=1) |
| self.assertEqual(hits[0][0], "beta") |
| self.assertEqual(hits[0][1], 1.0) |
|
|
| def test_nearest_symbol_noisy(self): |
| s = HDCStorage(d=D) |
| s.write("alpha", 1, symbol=True) |
| s.write("beta", 2, symbol=True) |
| probe = s.addr("alpha") ^ (1 << 4) |
| name, score = s.nearest_symbol(probe, k=1)[0] |
| self.assertEqual(name, "alpha") |
| self.assertLess(score, 1.0) |
| self.assertGreater(score, 0.9) |
|
|
| def test_prefix_filter(self): |
| s = HDCStorage(d=D) |
| s.write("code/0", 1, symbol=True) |
| s.write("tool/x", 2, symbol=True) |
| hits = s.nearest_symbol(s.addr("code/0"), k=5, prefix="tool/") |
| self.assertEqual([h[0] for h in hits], ["tool/x"]) |
|
|
|
|
| class TestContentAddressing(unittest.TestCase): |
| def setUp(self): |
| self.s = HDCStorage(d=D) |
| self.names = ["r0", "r1", "data/x", "data/y"] |
| for i, n in enumerate(self.names): |
| self.s.write(n, random_hv(D, random.Random(100 + i))) |
|
|
| def test_exact_hit_confidence_is_one(self): |
| name, conf = self.s.resolve(self.s.addr("data/x")) |
| self.assertEqual(name, "data/x") |
| self.assertEqual(conf, 1.0) |
|
|
| def test_fuzzy_hit(self): |
| addr = self.s.addr("data/y") |
| rng = random.Random(7) |
| |
| for pos in rng.sample(range(D), 10): |
| addr ^= 1 << pos |
| got = self.s.resolve(addr) |
| self.assertIsNotNone(got) |
| name, conf = got |
| self.assertEqual(name, "data/y") |
| self.assertLess(conf, 1.0) |
| self.assertGreater(conf, 0.9) |
| self.assertEqual(self.s.fuzzy_reads, 1) |
|
|
| def test_unrelated_address_returns_none(self): |
| alien = random_hv(D, random.Random(999)) |
| self.assertIsNone(self.s.resolve(alien)) |
|
|
| def test_empty_store_returns_none(self): |
| self.assertIsNone(HDCStorage(d=D).resolve(random_hv(D, random.Random(3)))) |
|
|
|
|
| class TestRecall(unittest.TestCase): |
| def test_sorted_descending_and_self_first(self): |
| s = HDCStorage(d=D) |
| vecs = {} |
| for i in range(6): |
| vecs["c%d" % i] = random_hv(D, random.Random(200 + i)) |
| s.write("c%d" % i, vecs["c%d" % i]) |
| hits = s.recall(vecs["c3"], k=4) |
| self.assertEqual(len(hits), 4) |
| self.assertEqual(hits[0][0], "c3") |
| self.assertEqual(hits[0][2], 1.0) |
| sims = [h[2] for h in hits] |
| self.assertEqual(sims, sorted(sims, reverse=True)) |
|
|
| def test_prefix_scopes_search(self): |
| s = HDCStorage(d=D) |
| probe = random_hv(D, random.Random(5)) |
| s.write("code/0", probe) |
| s.write("data/0", random_hv(D, random.Random(6))) |
| hits = s.recall(probe, k=5, prefix="data/") |
| self.assertEqual([h[0] for h in hits], ["data/0"]) |
|
|
| def test_floor_filters(self): |
| s = HDCStorage(d=D) |
| s.write("a", random_hv(D, random.Random(7))) |
| self.assertEqual(s.recall(random_hv(D, random.Random(8)), k=5, floor=0.5), []) |
|
|
|
|
| class TestPersistence(unittest.TestCase): |
| def test_save_load_roundtrip(self): |
| s = HDCStorage(d=D, capacity=64, label="unit") |
| for i in range(5): |
| s.write("cell%d" % i, random_hv(D, random.Random(300 + i)), symbol=True) |
| s.write("code/0", 0) |
| with tempfile.TemporaryDirectory() as tmp: |
| path = os.path.join(tmp, "unit.hdc") |
| size = s.save(path) |
| self.assertGreater(size, 0) |
| loaded = HDCStorage.load(path) |
| self.assertEqual(loaded.d, D) |
| self.assertEqual(dict(loaded.cells), dict(s.cells)) |
| self.assertEqual(loaded.symbols(), s.symbols()) |
| self.assertEqual(loaded.order, s.order) |
| self.assertEqual(loaded.read("cell3"), s.read("cell3")) |
|
|
| def test_reject_bad_magic(self): |
| with tempfile.TemporaryDirectory() as tmp: |
| path = os.path.join(tmp, "junk.hdc") |
| with open(path, "wb") as fh: |
| fh.write(b"NOTHDC!!" + b"\x00" * 32) |
| with self.assertRaises(ValueError): |
| HDCStorage.load(path) |
|
|
|
|
| class TestStat(unittest.TestCase): |
| def test_keys_and_counts(self): |
| s = HDCStorage(d=D, capacity=10) |
| s.write("a", random_hv(D, random.Random(9)), symbol=True) |
| s.read("a") |
| st = s.stat() |
| for key in ( |
| "label", |
| "d", |
| "cells_used", |
| "cells_free", |
| "occupancy", |
| "symbols", |
| "reads", |
| "writes", |
| "stored_bits", |
| "total_popcount", |
| ): |
| self.assertIn(key, st) |
| self.assertEqual(st["cells_used"], 1) |
| self.assertEqual(st["cells_free"], 9) |
| self.assertEqual(st["writes"], 1) |
| self.assertEqual(st["reads"], 1) |
| self.assertEqual(st["stored_bits"], D) |
| self.assertAlmostEqual(st["occupancy"], 0.1) |
|
|
| def test_clone_is_independent(self): |
| s = HDCStorage(d=D) |
| s.write("a", 1, symbol=True) |
| c = s.clone() |
| c.write("b", 2) |
| self.assertIn("a", c) |
| self.assertNotIn("b", s) |
| self.assertEqual(s.symbols(), ["a"]) |
|
|
|
|
| if __name__ == "__main__": |
| unittest.main() |
|
|
