| import unittest |
|
|
| from hdc.hypervec import hvc, sim |
| from hdc.ring import ( |
| PHASES, |
| HistoricLedger, |
| LedgerEntry, |
| RingController, |
| drifting_world, |
| evolve, |
| periodic_world, |
| report, |
| rewind, |
| tick_key, |
| ) |
|
|
| D = 2000 |
|
|
|
|
| class TestAlgebraicRing(unittest.TestCase): |
| def test_evolution_is_invertible(self): |
| key = hvc("ring/test-key", D) |
| for i in range(4): |
| state = hvc("ring/state/%d" % i, D) |
| for k in (1, 2, 7): |
| self.assertEqual(rewind(evolve(state, key, k, D), key, k, D), state) |
|
|
| def test_evolution_actually_moves_the_state(self): |
| key = hvc("ring/test-key", D) |
| state = hvc("ring/state/0", D) |
| self.assertLess(abs(sim(evolve(state, key, 1, D), state, D)), 0.1) |
|
|
| def test_tick_keys_are_distinct(self): |
| self.assertNotEqual(tick_key(0, D), tick_key(1, D)) |
| self.assertLess(abs(sim(tick_key(0, D), tick_key(1, D), D)), 0.1) |
|
|
|
|
| class TestLedger(unittest.TestCase): |
| def test_append_and_lookup(self): |
| ledger = HistoricLedger(d=D, capacity=4) |
| state = hvc("ledger/0", D) |
| ledger.append(LedgerEntry(0, "past", state, None, 0.5, 1.0)) |
| self.assertIs(ledger.at(0).state, state) |
| self.assertEqual(ledger.recall_state(0)[0], state) |
| self.assertIsNone(ledger.at(9)) |
|
|
| def test_eviction_recycles_into_the_phase_summary(self): |
| ledger = HistoricLedger(d=D, capacity=2) |
| for i in range(5): |
| ledger.append(LedgerEntry(i, PHASES[i % 3], hvc("ledger/%d" % i, D), None, 0.0, 1.0)) |
| self.assertEqual(len(ledger), 2) |
| self.assertEqual(ledger.recycled, 3) |
| self.assertEqual(sum(1 for p in PHASES if ledger.summaries[p].keys), 3) |
|
|
| def test_entropy_of_uniform_surprise_is_zero(self): |
| ledger = HistoricLedger(d=D, capacity=8) |
| for i in range(4): |
| ledger.append(LedgerEntry(i, "past", hvc("l/%d" % i, D), None, 0.0, 1.0)) |
| self.assertEqual(ledger.entropy(), 0.0) |
|
|
| def test_entropy_rises_when_surprise_spreads(self): |
| ledger = HistoricLedger(d=D, capacity=8) |
| for i, surprise in enumerate([0.0, 0.5, 1.0, 0.0]): |
| ledger.append(LedgerEntry(i, "past", hvc("l/%d" % i, D), None, surprise, 1.0)) |
| self.assertGreater(ledger.entropy(), 1.0) |
|
|
| def test_timeline_is_a_superposition_of_the_history(self): |
| ledger = HistoricLedger(d=D, capacity=4) |
| states = [hvc("timeline/%d" % i, D) for i in range(4)] |
| for i, state in enumerate(states): |
| ledger.append(LedgerEntry(i, PHASES[i % 3], state, None, 0.0, 1.0)) |
| record = ledger.timeline_record() |
| self.assertNotEqual(record, 0) |
| got, score = ledger.recall_timeline(2) |
| self.assertEqual(got, states[2]) |
| self.assertGreater(score, 0.1) |
|
|
| def test_empty_ledger_is_well_behaved(self): |
| ledger = HistoricLedger(d=D, capacity=2) |
| self.assertEqual(ledger.entropy(), 0.0) |
| self.assertEqual(ledger.timeline_record(), 0) |
| self.assertEqual(ledger.recall_timeline(0), (None, 0.0)) |
|
|
| def test_zero_capacity_rejected(self): |
| with self.assertRaises(ValueError): |
| HistoricLedger(d=D, capacity=0) |
|
|
|
|
| class TestRingController(unittest.TestCase): |
| def test_converges_on_a_periodic_world(self): |
| ring = RingController(d=D, capacity=8) |
| ring.run(periodic_world(4, 12, D)) |
| tail = ring.ledger.entries[-4:] |
| self.assertTrue(all(e.surprise < 1e-6 for e in tail), msg=[e.surprise for e in tail]) |
| self.assertEqual(ring.energy, 1.0) |
| self.assertEqual(ring.mode(), "stable") |
| self.assertEqual(ring.horizon(), 3) |
|
|
| def test_stays_surprised_on_a_drifting_world(self): |
| ring = RingController(d=D, capacity=8) |
| ring.run(drifting_world(4, 12, D)) |
| self.assertGreater(ring.mean_surprise(), 0.9) |
| self.assertEqual(ring.energy, 0.0) |
| self.assertEqual(ring.mode(), "explore") |
| self.assertEqual(ring.horizon(), 1) |
|
|
| def test_energy_stays_in_range(self): |
| ring = RingController(d=D, capacity=8) |
| for entry in ring.run(periodic_world(4, 20, D)): |
| self.assertGreaterEqual(entry.energy, 0.0) |
| self.assertLessEqual(entry.energy, 1.0) |
|
|
| def test_first_tick_has_no_prediction(self): |
| ring = RingController(d=D, capacity=4) |
| entry = ring.step(hvc("ring/first", D)) |
| self.assertEqual(entry.surprise, 0.5) |
|
|
| def test_phases_rotate(self): |
| ring = RingController(d=D, capacity=8) |
| phases = [e.phase for e in ring.run(periodic_world(4, 6, D))] |
| self.assertEqual(phases, [PHASES[i % 3] for i in range(6)]) |
|
|
| def test_horizon_follows_energy(self): |
| ring = RingController(d=D, capacity=8) |
| ring.energy = 0.0 |
| self.assertEqual(ring.horizon(), 1) |
| ring.energy = 1.0 |
| self.assertEqual(ring.horizon(), 3) |
|
|
| def test_predictions_become_exact(self): |
| ring = RingController(d=D, capacity=8) |
| ring.run(periodic_world(4, 10, D)) |
| predicted, _ = ring.model.predict(ring.observations[-1], "tick") |
| self.assertIsNotNone(predicted) |
| self.assertEqual(predicted, ring.observations[-4]) |
|
|
| def test_report_mentions_the_mode(self): |
| ring = RingController(d=D, capacity=4) |
| ring.run(periodic_world(4, 5, D)) |
| self.assertIn("mode", ring.report()) |
| self.assertIn("ring controller", report(D)) |
|
|
|
|
| if __name__ == "__main__": |
| unittest.main() |
|
|
