Digital Drosophila Archive
238 lines
12 KiB
Python
238 lines
12 KiB
Python
"""Synthetic CPU-only contract tests; no real trajectories, GPU, or file writes."""
|
|
import sys
|
|
import unittest
|
|
|
|
import numpy as np
|
|
|
|
from replay_core import IncomingIndex, replay, state_step
|
|
|
|
|
|
def coefficients(**changes):
|
|
result = dict(gd=1., c0=0., cg=0., cv=1., threshold=1., reset=0., seed_volt=2.)
|
|
result.update(changes)
|
|
return result
|
|
|
|
|
|
def record(rows=(), key="tick"):
|
|
values = np.array(rows, np.int32).reshape(-1, 2)
|
|
return {key: values[:, 0], "neuron_idx": values[:, 1]}
|
|
|
|
|
|
def empty_index(count=1):
|
|
return IncomingIndex(np.empty(0, np.int32), np.empty(0, np.int32),
|
|
np.empty(0, np.int64), np.empty(0, np.float64), count)
|
|
|
|
|
|
def one_edge(weight=2.):
|
|
return IncomingIndex(np.array([10], np.int32), np.array([0], np.int32),
|
|
np.array([81], np.int64), np.array([weight], np.float64), 1)
|
|
|
|
|
|
def shadow(steps=25, *, initial_v=None, initial_g=None, coeff=None, requests=None,
|
|
events=None, incoming=None, h32=None, callback=None, targets=None):
|
|
targets = np.array([100], np.int32) if targets is None else targets
|
|
n = len(targets)
|
|
return replay(np.zeros(n, np.float64) if initial_v is None else initial_v,
|
|
np.zeros(n, np.float64) if initial_g is None else initial_g,
|
|
coefficients() if coeff is None else coeff, steps, targets,
|
|
record() if requests is None else requests,
|
|
record() if events is None else events,
|
|
empty_index(n) if incoming is None else incoming,
|
|
np.zeros((steps, n), np.float32) if h32 is None else h32, callback)
|
|
|
|
|
|
class StepTests(unittest.TestCase):
|
|
def test_strict_threshold_and_positive_zero_reset(self):
|
|
v = np.array([1., np.nextafter(1., np.inf)], np.float64)
|
|
result = state_step(v, np.array([3., -3.]), np.zeros(2, np.int32), np.zeros(2, bool), coefficients())
|
|
np.testing.assert_array_equal(result["fired"], [False, True])
|
|
np.testing.assert_array_equal(result["ref"], [0, 22])
|
|
self.assertEqual(result["g"].view(np.uint64)[1], 0)
|
|
self.assertEqual(result["v_threshold"][1], v[1])
|
|
|
|
def test_voltage_uses_old_g_and_separate_expression_order(self):
|
|
c = coefficients(gd=.4, cg=.2, cv=.3, c0=.1, threshold=100., seed_volt=101.)
|
|
result = state_step(np.array([3.]), np.array([2.]), np.array([0], np.int32), np.array([False]), c)
|
|
expected = np.float64(c["c0"]) + ((np.float64(c["cg"]) * np.float64(2.)) +
|
|
(np.float64(c["cv"]) * np.float64(3.)))
|
|
self.assertEqual(result["v"][0].hex(), expected.hex())
|
|
self.assertEqual(result["g"][0].hex(), (np.float64(.4) * np.float64(2.)).hex())
|
|
self.assertNotEqual(result["v"][0], .1 + (.2 * .8 + .3 * 3.))
|
|
|
|
def test_negative_g_and_voltage_freeze_when_refractory(self):
|
|
result = state_step(np.array([-.052]), np.array([-.123]), np.array([22], np.int32),
|
|
np.array([True]), coefficients(gd=.5, cg=1., c0=.3))
|
|
self.assertEqual(result["v"][0], -.052)
|
|
self.assertEqual(result["g"][0], -.123)
|
|
self.assertEqual(result["ref"][0], 21)
|
|
self.assertFalse(result["fired"][0])
|
|
self.assertTrue(result["request_rejected"][0])
|
|
|
|
def test_ref_one_releases_before_request_and_reset(self):
|
|
result = state_step(np.array([-.052]), np.array([-.123]), np.array([1], np.int32),
|
|
np.array([True]), coefficients())
|
|
self.assertTrue(result["available"][0])
|
|
self.assertTrue(result["request_accepted"][0])
|
|
self.assertEqual(result["v_threshold"][0], 2.)
|
|
self.assertTrue(result["fired"][0])
|
|
self.assertEqual(result["ref"][0], 22)
|
|
self.assertEqual(result["g"][0], 0.)
|
|
|
|
def test_state_inputs_unchanged(self):
|
|
v, g, ref, request = np.array([2.]), np.array([5.]), np.array([0], np.int32), np.array([False])
|
|
inputs = [x.copy() for x in (v, g, ref, request)]
|
|
state_step(v, g, ref, request, coefficients())
|
|
for actual, expected in zip((v, g, ref, request), inputs):
|
|
np.testing.assert_array_equal(actual, expected)
|
|
|
|
|
|
class ReplayTests(unittest.TestCase):
|
|
def test_logged_target_spikes_are_not_forced(self):
|
|
result = shadow(events=record([(0, 100), (22, 100)]))
|
|
self.assertEqual(len(result["step"]), 0)
|
|
np.testing.assert_array_equal(result["first_spike"], [-1])
|
|
|
|
def test_predicted_events_exist_without_logged_target_spikes(self):
|
|
result = shadow(requests=record([(0, 100), (1, 100), (21, 100), (22, 100), (23, 100)]))
|
|
np.testing.assert_array_equal(result["step"], [0, 22])
|
|
np.testing.assert_array_equal(result["neuron_idx"], [100, 100])
|
|
np.testing.assert_array_equal(result["counts"], [2])
|
|
np.testing.assert_array_equal(result["first_spike"], [0])
|
|
|
|
def test_anatomical_arrival_is_tick18_then_voltage_tick19(self):
|
|
history = []
|
|
shadow(steps=20, coeff=coefficients(cg=1., threshold=100., seed_volt=101.),
|
|
events=record([(0, 10)]), incoming=one_edge(), callback=lambda t, s: history.append(s))
|
|
self.assertEqual(history[17]["g"][0], 0.)
|
|
self.assertEqual(history[18]["g"][0], 2.)
|
|
self.assertEqual(history[18]["v"][0], 0.)
|
|
self.assertEqual(history[19]["v"][0], 2.)
|
|
np.testing.assert_array_equal(history[18]["due_edge_id"], [81])
|
|
|
|
def test_arrival_rejected_on_same_tick_target_fire(self):
|
|
history = []
|
|
result = shadow(steps=19, events=record([(0, 10)]), incoming=one_edge(),
|
|
requests=record([(18, 100)]), callback=lambda t, s: history.append(s))
|
|
np.testing.assert_array_equal(result["step"], [18])
|
|
self.assertFalse(history[18]["arrival_accepted"][0])
|
|
self.assertEqual(history[18]["g_pre_apl"][0], 0.)
|
|
|
|
def test_refractory_rejects18_and_release22_accepts(self):
|
|
history = []
|
|
shadow(steps=23, events=record([(0, 10), (4, 10)]), incoming=one_edge(),
|
|
requests=record([(0, 100)]), callback=lambda t, s: history.append(s))
|
|
self.assertFalse(history[18]["arrival_accepted"][0])
|
|
self.assertTrue(history[22]["arrival_accepted"][0])
|
|
self.assertEqual(history[22]["ref"][0], 0)
|
|
self.assertEqual(history[22]["g_pre_apl"][0], 2.)
|
|
|
|
def test_apl_unconditional_during_freeze_and_release_decay(self):
|
|
history = []
|
|
h = np.ones((23, 1), np.float32)
|
|
shadow(steps=23, coeff=coefficients(gd=.5), requests=record([(0, 100)]), h32=h,
|
|
callback=lambda t, s: history.append(s))
|
|
self.assertEqual(history[0]["g_after_reset"][0], 0.)
|
|
self.assertEqual(history[0]["g"][0], -.001)
|
|
for tick in range(1, 22):
|
|
self.assertEqual(history[tick]["g_decayed"][0].hex(), history[tick]["old_g"][0].hex())
|
|
self.assertEqual(history[tick]["g"][0].hex(), (history[tick]["old_g"][0] - np.float64(.001)).hex())
|
|
self.assertEqual(history[22]["g_decayed"][0].hex(), (.5 * history[22]["old_g"][0]).hex())
|
|
|
|
def test_si_promotion_order_is_not_early_float32_scaling(self):
|
|
h = np.array([[.01]], np.float32)
|
|
result = shadow(steps=1, h32=h)
|
|
expected = -(np.float64(h[0, 0]) * np.float64(.001))
|
|
wrong = -np.float64(h[0, 0] * np.float32(.001))
|
|
self.assertEqual(result["g"][0].hex(), expected.hex())
|
|
self.assertNotEqual(result["g"][0].hex(), wrong.hex())
|
|
|
|
def test_arrivals_start_at_existing_g_in_original_edge_order(self):
|
|
incoming = IncomingIndex(np.array([10, 11], np.int32), np.array([0, 0], np.int32),
|
|
np.array([2, 81], np.int64), np.array([-1e16, 1.], np.float64), 1)
|
|
result = shadow(steps=19, initial_g=np.array([1e16]), incoming=incoming, events=record([(0, 10), (0, 11)]))
|
|
self.assertEqual(result["g"][0], 1.)
|
|
self.assertNotEqual(result["g"][0], np.float64(1e16) + np.sum(incoming.weights))
|
|
|
|
def test_zero_weight_edges_still_present_in_due_trace(self):
|
|
history = []
|
|
shadow(steps=19, events=record([(0, 10)]), incoming=one_edge(0.), callback=lambda t, s: history.append(s))
|
|
np.testing.assert_array_equal(history[18]["due_edge_id"], [81])
|
|
np.testing.assert_array_equal(history[18]["arrival_accepted"], [True])
|
|
|
|
def test_non_target_requests_ignored_and_step_alias_supported(self):
|
|
result = shadow(requests=record([(0, 10), (0, 100)], key="step"), events=record(key="step"))
|
|
np.testing.assert_array_equal(result["step"], [0])
|
|
np.testing.assert_array_equal(result["neuron_idx"], [100])
|
|
|
|
def test_callback_mutation_cannot_change_replay(self):
|
|
args = dict(steps=23, requests=record([(0, 100)]), h32=np.ones((23, 1), np.float32))
|
|
expected = shadow(**args)
|
|
|
|
def mutate(tick, stage):
|
|
for value in stage.values():
|
|
value[...] = 0
|
|
|
|
actual = shadow(**args, callback=mutate)
|
|
for name in expected:
|
|
np.testing.assert_array_equal(actual[name].view(np.uint8), expected[name].view(np.uint8))
|
|
|
|
def test_callback_can_fail_fast_without_later_ticks(self):
|
|
visited = []
|
|
|
|
def stop(tick, stage):
|
|
visited.append(tick)
|
|
if tick == 2:
|
|
raise RuntimeError("intentional mismatch")
|
|
|
|
with self.assertRaisesRegex(RuntimeError, "intentional mismatch"):
|
|
shadow(callback=stop)
|
|
self.assertEqual(visited, [0, 1, 2])
|
|
|
|
def test_no_dense_history_and_exact_output_dtypes(self):
|
|
result = shadow()
|
|
self.assertFalse(any(value.ndim > 1 for value in result.values()))
|
|
for name in ("v", "g"):
|
|
self.assertEqual(result[name].dtype, np.float64)
|
|
for name in ("ref", "counts", "step", "neuron_idx", "first_spike", "target_globals"):
|
|
self.assertEqual(result[name].dtype, np.int32)
|
|
|
|
|
|
class ValidationTests(unittest.TestCase):
|
|
def test_index_copies_inputs_and_reuses_source_lookup(self):
|
|
src, dst, ids, w = np.array([3, 3, 8], np.int32), np.array([1, 0, 1], np.int32), np.array([8, 20, 99], np.int64), np.array([1., 0., -2.])
|
|
index = IncomingIndex(src, dst, ids, w, 2)
|
|
w[:] = 99
|
|
np.testing.assert_array_equal(index.weights, [1., 0., -2.])
|
|
np.testing.assert_array_equal(index.due(np.array([3, 7, 8], np.int32)), [0, 1, 2])
|
|
self.assertFalse(index.weights.flags.writeable)
|
|
|
|
def test_rejects_bad_index_dtype_range_order_and_nonfinite(self):
|
|
base = dict(src_global=np.array([10, 11], np.int32), dst_local=np.array([0, 0], np.int32),
|
|
edge_id=np.array([3, 8], np.int64), weights=np.array([1., 2.]), target_count=1)
|
|
changes = [dict(src_global=np.array([10, 11], np.int64)), dict(src_global=np.array([11, 10], np.int32)),
|
|
dict(src_global=np.array([10, 166700], np.int32)), dict(dst_local=np.array([0, 1], np.int32)),
|
|
dict(edge_id=np.array([3, 3], np.int64)), dict(weights=np.array([1., np.nan])),
|
|
dict(weights=np.array([1., 2.], np.float32))]
|
|
for change in changes:
|
|
with self.subTest(change=change), self.assertRaises(ValueError):
|
|
IncomingIndex(**(base | change))
|
|
|
|
def test_rejects_invalid_replay_arrays(self):
|
|
variants = [dict(initial_v=np.array([0.], np.float32)), dict(initial_g=np.array([np.inf])),
|
|
dict(targets=np.array([100], np.int64)), dict(targets=np.array([100, 100], np.int32)),
|
|
dict(h32=np.zeros((25, 1), np.float64)), dict(h32=np.full((25, 1), np.nan, np.float32)),
|
|
dict(h32=np.full((25, 1), -1, np.float32)), dict(coeff=coefficients(gd=np.float32(.9))),
|
|
dict(requests=record([(0, 100), (0, 100)])), dict(events=record([(25, 10)])),
|
|
dict(events=record([(0, 166700)])), dict(steps=0)]
|
|
for change in variants:
|
|
with self.subTest(change=change), self.assertRaises(ValueError):
|
|
shadow(**change)
|
|
|
|
def test_no_scientific_runtime_import(self):
|
|
self.assertNotIn("torch", sys.modules)
|
|
self.assertNotIn("brian2", sys.modules)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main(verbosity=2)
|