Digital Drosophila Archive
153 lines
8.3 KiB
Python
153 lines
8.3 KiB
Python
"""Synthetic CPU-only tests, including the untouched sibling AH callback."""
|
|
import importlib.util
|
|
from pathlib import Path
|
|
import unittest
|
|
|
|
import numpy as np
|
|
|
|
from pn_observer import PNObserver
|
|
|
|
|
|
path = Path(__file__).resolve().parent.parent / "kc-drive-audit-v1/replay_core.py"
|
|
spec = importlib.util.spec_from_file_location("frozen_ah_observer_test_core", path)
|
|
core = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(core)
|
|
|
|
|
|
def event(rows=()):
|
|
values = np.array(rows, np.int32).reshape(-1, 2)
|
|
return {"step": values[:, 0], "neuron_idx": values[:, 1]}
|
|
|
|
|
|
def run(*, edges=(), events=(), requests=(), n=1, design=(), steps=25, coeff=None,
|
|
initial_v=None, initial_g=None, sink=None, cap=1000, h=None, hook=None):
|
|
targets = np.arange(100, 100 + n, dtype=np.int32)
|
|
constants = dict(gd=1., c0=0., cg=1., cv=1., threshold=1., reset=0., seed_volt=2.)
|
|
if coeff: constants.update(coeff)
|
|
arrays = np.asarray(edges).reshape(-1, 4)
|
|
incoming = core.IncomingIndex(arrays[:, 0].astype(np.int32), arrays[:, 1].astype(np.int32),
|
|
np.arange(len(arrays), dtype=np.int64) + 81, arrays[:, 2].astype(np.float64), n)
|
|
groups = arrays[:, 3].astype(np.int16)
|
|
observer = PNObserver(targets, np.array(design, np.int32), incoming, groups, ("ALPN", "other"), constants,
|
|
steps, max_first_event_rows=cap, event_sink=sink)
|
|
def callback(tick, state):
|
|
if hook: hook(tick, state, observer)
|
|
observer(tick, state)
|
|
result = core.replay(np.zeros(n, np.float64) if initial_v is None else np.array(initial_v, np.float64),
|
|
np.zeros(n, np.float64) if initial_g is None else np.array(initial_g, np.float64), constants, steps, targets,
|
|
event(requests), event(events), incoming, np.zeros((steps, n), np.float32) if h is None else h, callback)
|
|
data, report = observer.finish(result)
|
|
return observer, result, data, report
|
|
|
|
|
|
class PNObserverTests(unittest.TestCase):
|
|
def test_18_step_arrival_then_19_step_first_spike(self):
|
|
observer, result, data, report = run(edges=[(10, 0, 2., 0)], events=[(0, 10)])
|
|
self.assertEqual(data["first_spike_tick"].tolist(), [19])
|
|
self.assertEqual(len(observer.first_arrival_records), 1)
|
|
row = observer.first_arrival_records[0]
|
|
self.assertEqual((row["source_spike_tick"], row["arrival_tick"]), (0, 18))
|
|
self.assertEqual(row["v_prethreshold_SI"], 0.)
|
|
self.assertEqual(row["threshold_margin_SI"], 1.)
|
|
self.assertEqual(data["first_integrated_V_from_ALPN/positive_weight"].tolist(), [2.])
|
|
self.assertEqual(observer.edge_summary()["target_first_spike_tick"].tolist(), [19])
|
|
self.assertTrue(report["pass_all"])
|
|
|
|
def test_same_tick_first_spike_arrival_is_rejected_not_antecedent(self):
|
|
observer, _, data, _ = run(edges=[(10, 0, 2., 0)], events=[(0, 10)], requests=[(18, 100)], design=[100])
|
|
self.assertEqual(data["first_spike_tick"].tolist(), [18])
|
|
self.assertEqual(observer.first_arrival_records, [])
|
|
self.assertEqual(data["ALPN_rejected_positive_edges"].tolist(), [1])
|
|
self.assertEqual(data["ALPN_accepted_positive_before_first_edges"].tolist(), [0])
|
|
self.assertEqual(data["first_prethreshold_V_from_request_or_reset"].tolist(), [2.])
|
|
|
|
def test_all_positive_and_inhibitory_events_are_retained(self):
|
|
observer, _, data, _ = run(edges=[(10, 0, 2., 0), (11, 0, -.5, 1), (12, 0, 0., 1)],
|
|
events=[(0, 10), (0, 11), (0, 12)])
|
|
self.assertEqual([r["weight_sign"] for r in observer.first_arrival_records], [1, -1])
|
|
self.assertEqual(data["first_spike_tick"].tolist(), [19])
|
|
self.assertEqual(data["other_accepted_zero_edges"].tolist(), [1])
|
|
self.assertEqual(data["other_accepted_negative_before_first_abs_V"].tolist(), [.5])
|
|
self.assertEqual(data["first_integrated_V_from_other/negative_weight"].tolist(), [-.5])
|
|
|
|
def test_never_active_targets_retain_all_prefirst_records(self):
|
|
observer, _, data, report = run(edges=[(10, 0, .25, 0)], events=[(0, 10), (1, 10)],
|
|
coeff={"cg": 0.})
|
|
self.assertEqual(data["first_spike_tick"].tolist(), [-1])
|
|
self.assertTrue(np.isnan(data["first_v_threshold_SI"][0]))
|
|
self.assertEqual(observer.edge_summary()["accepted_event_count"].tolist(), [2])
|
|
self.assertEqual(observer.edge_summary()["target_first_spike_tick"].tolist(), [-1])
|
|
self.assertEqual(report["first_arrival_record_count"], 2)
|
|
|
|
def test_multiple_targets_independent_first_times_and_design(self):
|
|
observer, _, data, _ = run(n=2, edges=[(10, 0, 2., 0), (10, 1, .25, 0)],
|
|
events=[(0, 10)], requests=[(23, 101)], design=[101])
|
|
self.assertEqual(data["first_spike_tick"].tolist(), [19, 23])
|
|
self.assertEqual([r["target_is_design"] for r in observer.first_arrival_records], [False, True])
|
|
self.assertEqual(data["first_spike_coincident_with_accepted_request"].tolist(), [False, True])
|
|
|
|
def test_old_g_not_decayed_g_used_for_first_voltage(self):
|
|
_, _, data, _ = run(edges=[(10, 0, 4., 0)], events=[(0, 10)],
|
|
coeff={"gd": .1, "cg": .5, "cv": 0.})
|
|
self.assertEqual(data["first_integrated_V_from_ALPN/positive_weight"].tolist(), [2.])
|
|
self.assertEqual(data["first_spike_tick"].tolist(), [19])
|
|
|
|
def test_initial_state_is_not_assigned_to_source_class(self):
|
|
_, _, data, _ = run(initial_g=[2.])
|
|
self.assertEqual(data["first_spike_tick"].tolist(), [0])
|
|
self.assertEqual(data["first_integrated_V_from_initial_and_affine_baseline"].tolist(), [2.])
|
|
self.assertEqual(data["first_integrated_V_from_ALPN/positive_weight"].tolist(), [0.])
|
|
|
|
def test_streaming_sink_and_row_budget_fail_closed(self):
|
|
received = []
|
|
observer, _, _, report = run(edges=[(10, 0, .2, 0)], events=[(0, 10)],
|
|
sink=lambda rows: received.extend(rows))
|
|
self.assertEqual(observer.first_arrival_records, [])
|
|
self.assertEqual(len(received), report["first_arrival_record_count"])
|
|
with self.assertRaisesRegex(ValueError, "resource limit"):
|
|
run(edges=[(10, 0, .2, 0), (11, 0, .2, 1)], events=[(0, 10), (0, 11)], cap=1)
|
|
|
|
def test_nonzero_PN_APL_and_non_design_request_are_rejected(self):
|
|
h = np.zeros((25, 1), np.float32); h[0, 0] = 1
|
|
with self.assertRaisesRegex(ValueError, "zero graded APL"):
|
|
run(h=h)
|
|
with self.assertRaisesRegex(ValueError, "outside frozen odor design"):
|
|
run(requests=[(0, 100)])
|
|
|
|
def test_callback_arrays_not_modified(self):
|
|
snapshots = []
|
|
def hook(tick, state, observer):
|
|
if tick == 0:
|
|
before = {key: value.copy() for key, value in state.items() if isinstance(value, np.ndarray)}
|
|
observer(tick, state)
|
|
for key, value in before.items():
|
|
self.assertEqual(state[key].tobytes(), value.tobytes())
|
|
snapshots.append(True)
|
|
# Already consumed this tick; stop this synthetic test directly.
|
|
raise RuntimeError("checked immutability")
|
|
with self.assertRaisesRegex(RuntimeError, "checked immutability"):
|
|
run(hook=hook)
|
|
self.assertEqual(snapshots, [True])
|
|
|
|
def test_bad_queue_guard_and_tick_fail_closed(self):
|
|
def wrong_guard(tick, state, observer):
|
|
if tick == 18: state["arrival_accepted"][:] = False
|
|
with self.assertRaisesRegex(ValueError, "Arrival guard"):
|
|
run(edges=[(10, 0, 2., 0)], events=[(0, 10)], hook=wrong_guard)
|
|
def wrong_tick(tick, state, observer):
|
|
observer(tick + 1, state)
|
|
with self.assertRaisesRegex(ValueError, "out-of-order"):
|
|
run(hook=wrong_tick)
|
|
|
|
def test_sequential_arrival_roundoff_has_own_lane(self):
|
|
_, _, data, report = run(edges=[(10, 0, 1e16, 0), (11, 0, 1., 0), (12, 0, -1e16, 1)],
|
|
events=[(0, 10), (0, 11), (0, 12)], coeff={"cg": 0.})
|
|
if np.finfo(np.longdouble).nmant > np.finfo(np.float64).nmant:
|
|
self.assertEqual(data["arrival_roundoff_g_SI"].tolist(), [-1.])
|
|
self.assertTrue(report["voltage_decomposition"]["roundoff_lane_explicit"])
|
|
self.assertFalse(report["unique_cause_claim"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|