"""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()