"""Synthetic-only tests; no recorded scientific data, GPU, or file writes.""" import importlib.util import json from pathlib import Path import unittest import numpy as np from mbon_accounting import MBONAccounting, bit_equal _source = Path(__file__).resolve().parents[1] / "kc-drive-audit-v1" / "replay_core.py" _spec = importlib.util.spec_from_file_location("_mbon_test_frozen_ah_replay", _source) ah = importlib.util.module_from_spec(_spec) _spec.loader.exec_module(ah) def coeff(**changes): values = dict(gd=.5, c0=0., cg=1., cv=1., threshold=100., reset=0., seed_volt=101.) values.update(changes) return values def events(rows=()): array = np.asarray(rows, np.int32).reshape(-1, 2) return dict(tick=array[:, 0], neuron_idx=array[:, 1]) def graph(weights=(), destinations=None, n=1): e = len(weights) return ah.IncomingIndex(np.arange(10, 10 + e, dtype=np.int32), np.zeros(e, np.int32) if destinations is None else np.array(destinations, np.int32), np.arange(1000, 1000 + e, dtype=np.int64), np.array(weights, np.float64), n) def run(index, *, groups=None, names=("plastic_KC",), steps=25, constants=None, spikes=(), requests=(), v=None, g=None, observer_hook=None): constants = coeff() if constants is None else constants groups = np.zeros(len(index.edge_id), np.int32) if groups is None else np.array(groups, np.int32) obs = MBONAccounting(index, groups, names, constants, steps) seen = [] def callback(tick, stage): # Read-only views also catch accidental assignments to callback data. original = {key: value.copy() for key, value in stage.items()} for value in stage.values(): value.setflags(write=False) obs(tick, stage) for key in original: if not bit_equal(stage[key], original[key]): raise AssertionError("Observer modified callback field " + key) if observer_hook: observer_hook(tick, stage) seen.append(original) result = ah.replay(np.zeros(index.target_count, np.float64) if v is None else np.asarray(v, np.float64), np.zeros(index.target_count, np.float64) if g is None else np.asarray(g, np.float64), constants, steps, np.arange(100, 100 + index.target_count, dtype=np.int32), events(requests), events(spikes), index, np.zeros((steps, index.target_count), np.float32), callback) data, edges, summary = obs.finish() return data, edges, summary, result, seen, obs class AccountingTests(unittest.TestCase): def test_empty_graph_initial_baseline_and_json_summary(self): data, edge, summary, result, _, _ = run(graph(), steps=3, g=[2.], v=[-.052]) self.assertEqual(summary["due_count"], 0) self.assertEqual(len(edge["edge_id"]), 0) np.testing.assert_array_equal(data["history_group_g"], 0.) np.testing.assert_array_equal(data["history_g"], data["history_baseline_g"]) self.assertTrue(bit_equal(data["final_g"], result["g"])) json.dumps(summary, allow_nan=False) self.assertFalse(summary["functional_transfer_claim"]) self.assertFalse(summary["original_events_verified_by_observer"]) def test_real_delay_old_g_for_voltage_and_end_tick_integral(self): data, edge, summary, _, _, _ = run(graph([2.]), steps=21, spikes=[(0, 10)]) np.testing.assert_array_equal(data["history_g"][:18], 0.) np.testing.assert_array_equal(data["history_g"][18:, 0], [2., 1., .5]) np.testing.assert_array_equal(data["history_v"][18:, 0], [0., 2., 3.]) self.assertAlmostEqual(data["g_integral_V_ms"][0, 0], .35) np.testing.assert_array_equal(edge["accepted_count"], [1]) self.assertEqual(summary["max_g_lane_residual_V"], 0.) def test_arrival_rejected_on_fire_and_refractory_release(self): data, edge, summary, result, _, _ = run(graph([2., -1.]), steps=41, spikes=[(t, source) for t in (0, 1, 22) for source in (10, 11)], requests=[(18, 100)]) np.testing.assert_array_equal(edge["due_count"], [3, 3]) np.testing.assert_array_equal(edge["accepted_count"], [1, 1]) np.testing.assert_array_equal(edge["rejected_count"], [2, 2]) np.testing.assert_array_equal(data["history_g"][18:40], 0.) self.assertEqual(data["history_g"][40, 0], 1.) self.assertEqual(data["rejected_positive_V"][0, 0], 4.) self.assertEqual(data["rejected_negative_V"][0, 0], -2.) self.assertEqual(data["accepted_negative_V"][0, 0], -1.) np.testing.assert_array_equal(data["spike_count"], result["counts"]) self.assertEqual(summary["accepted_count"], 2) def test_firing_clears_source_g_and_voltage_lanes(self): data, _, _, result, _, _ = run(graph([2.]), steps=21, constants=coeff(threshold=1., seed_volt=2.), spikes=[(0, 10)]) self.assertEqual(data["first_spike_tick"][0], 19) np.testing.assert_array_equal(data["history_group_g"][19:], 0.) np.testing.assert_array_equal(data["final_group_v"], 0.) self.assertEqual(data["history_ref"][19, 0], 22) self.assertEqual(data["history_ref"][20, 0], 21) self.assertTrue(bit_equal(data["final_ref"], result["ref"])) def test_group_sign_totals_and_zero_weight_event_counts(self): index = graph([2., -3., 0., 4.], destinations=[0, 0, 1, 1], n=2) data, edge, summary, _, _, _ = run(index, groups=[0, 1, 0, 1], names=("plastic_KC", "other"), steps=19, spikes=[(0, x) for x in range(10, 14)]) np.testing.assert_array_equal(data["accepted_count"], [[1, 1], [1, 1]]) np.testing.assert_array_equal(data["accepted_zero_count"], [[0, 0], [1, 0]]) np.testing.assert_array_equal(data["accepted_positive_V"], [[2., 0.], [0., 4.]]) np.testing.assert_array_equal(data["accepted_negative_V"], [[0., -3.], [0., 0.]]) self.assertEqual(summary["groups"]["other"]["accepted_negative_V"], -3.) self.assertEqual(int(edge["due_count"].sum()), 4) def test_cross_lane_roundoff_is_visible_not_reassigned(self): data, _, summary, _, _, _ = run(graph([1.e16, 1., -1.e16]), groups=[0, 1, 0], names=("plastic_KC", "other"), steps=19, constants=coeff(threshold=1.e30, seed_volt=2.e30), spikes=[(0, 10), (0, 11), (0, 12)]) self.assertEqual(data["history_g"][18, 0], 0.) self.assertEqual(data["history_group_g"][18, 0, 1, 0], 1.) self.assertEqual(data["history_g_residual"][18, 0], -1.) self.assertEqual(summary["max_g_lane_residual_V"], 1.) def test_callback_and_constructor_inputs_never_modified(self): index = graph([1., -.25]) snapshots = {name: getattr(index, name).copy() for name in ("src_global", "dst_local", "edge_id", "weights")} run(index, spikes=[(0, 10), (0, 11)], requests=[(1, 100)]) for name, original in snapshots.items(): self.assertTrue(bit_equal(getattr(index, name), original)) def test_incomplete_duplicate_tick_and_repeated_finish_fail(self): index = graph() with self.assertRaises(ValueError): MBONAccounting(index, np.empty(0, np.int32), ("other",), coeff(), 2).finish() _, _, _, _, stages, finished = run(index, steps=2) with self.assertRaises(ValueError): finished.finish() obs = MBONAccounting(index, np.empty(0, np.int32), ("other",), coeff(), 2) obs(0, stages[0]) with self.assertRaises(ValueError): obs(0, stages[0]) def test_bad_dtype_groups_coefficients_and_nonfinite_fail(self): index = graph([1.]) for codes in (np.array([0], np.int64), np.array([1], np.int32), np.array([], np.int32)): with self.assertRaises(ValueError): MBONAccounting(index, codes, ("other",), coeff(), 2) for bad in (coeff(gd=np.float32(.5)), coeff(gd=float("nan")), coeff(dt_ms=.2)): with self.assertRaises(ValueError): MBONAccounting(index, np.array([0], np.int32), ("other",), bad, 2) def test_stage_guard_apl_and_canonical_order_corruption_fail(self): index = graph([1., 2.]) _, _, _, _, stages, _ = run(index, steps=19, spikes=[(0, 10), (0, 11)]) for corruption in ("guard", "order", "apl", "state"): obs = MBONAccounting(index, np.array([0, 0], np.int32), ("other",), coeff(), 19) for tick in range(18): obs(tick, stages[tick]) stage = {k: v.copy() for k, v in stages[18].items()} if corruption == "guard": stage["arrival_accepted"][0] = False elif corruption == "order": stage["due_edge_indices"] = stage["due_edge_indices"][::-1].copy() elif corruption == "apl": stage["h32"][0] = 1. else: stage["g_pre_apl"][0] += 1. with self.subTest(corruption=corruption), self.assertRaises(ValueError): obs(18, stage) if __name__ == "__main__": unittest.main()