import unittest from types import SimpleNamespace import numpy as np from accounting import DriveAccounting, bit_equal, fraction GROUPS = ('KC', 'APL', 'ALPN', 'MBON', 'anatomical_DAN', 'dopamine_annotation_only', 'other', 'unclassified') def _events(ticks=(), nodes=()): return {'step': np.asarray(ticks, np.int32), 'neuron_idx': np.asarray(nodes, np.int32)} def _account(targets, destinations=(), weights=(), raw=(), groups=(), expected=None, counts=None, gd=.5, histories=None): targets = np.asarray(targets, np.int32) incoming = SimpleNamespace(dst_local=np.asarray(destinations, np.int32), weights=np.asarray(weights, np.float64)) coeff = {'gd': np.float64(gd), 'threshold': np.float64(-.045)} return DriveAccounting(targets, incoming, np.asarray(raw, np.int64), np.asarray(groups, np.int32), GROUPS, _events() if expected is None else expected, coeff, np.zeros(len(targets), np.int32) if counts is None else np.asarray(counts, np.int32), histories=histories) def _stage(old_g, g_decayed, g_after_reset, g_pre_apl, delta, *, available, fired, ref, due=(), accepted=(), v_threshold=None): """Manually specified accounting callback; no runtime or replay dependency.""" old_g = np.asarray(old_g, np.float64) n = len(old_g) delta = np.asarray(delta, np.float64) pre = np.asarray(g_pre_apl, np.float64) ref = np.asarray(ref, np.int32) threshold_v = np.full(n, -.05, np.float64) if v_threshold is None else np.asarray(v_threshold, np.float64) return {'old_g': old_g, 'g_decayed': np.asarray(g_decayed, np.float64), 'g_after_reset': np.asarray(g_after_reset, np.float64), 'g_pre_apl': pre, 'delta_si': delta, 'g': np.subtract(pre, delta), 'available': np.asarray(available, bool), 'fired': np.asarray(fired, bool), 'ref': ref, 'ref_pre_apl': ref.copy(), 'v': np.full(n, -.052, np.float64), 'v_pre_apl': np.full(n, -.052, np.float64), 'v_threshold': threshold_v, 'due_edge_indices': np.asarray(due, np.int64), 'arrival_accepted': np.asarray(accepted, bool)} def _two_tick_fixture(): # KC 10 fires immediately, KC 20 first fires one tick later, KC 30 never # fires. All contacts are positive anatomical counts; signed effective # SI increments are an independent vector, including a zeroed APL edge. account = _account([10, 20, 30], [0, 1, 2, 2, 2, 0], [1., 2., 3., -1., 0., -2.], [4, 8, 12, 4, 16, 8], [2, 2, 0, 3, 1, 3], _events([0, 1], [10, 20]), [1, 1, 0]) first = _stage([2., 4., 8.], [1., 2., 4.], [0., 2., 4.], [0., 4., 6.], [.25, .5, 1.], available=[1, 1, 1], fired=[1, 0, 0], ref=[22, 0, 0], due=np.arange(6), accepted=[0, 1, 1, 1, 1, 0], v_threshold=[-.044, -.049, -.05]) second = _stage([-.25, 3.5, 5.], [-.25, 1.75, 2.5], [-.25, 0., 2.5], [-.25, 0., 2.5], [.25, .5, 1.], available=[0, 1, 1], fired=[0, 1, 0], ref=[21, 22, 0], v_threshold=[-.052, -.044, -.05]) return account, first, second class AccountingTests(unittest.TestCase): def test_bitwise_signed_zero(self): self.assertFalse(bit_equal(np.array([0.0]), np.array([-0.0]))) def test_dtype_is_part_of_evidence(self): self.assertFalse(bit_equal(np.array([1], np.int32), np.array([1], np.int64))) def test_shape_is_part_of_evidence(self): self.assertFalse(bit_equal(np.array([1.0]), np.array([[1.0]]))) def test_equal(self): self.assertTrue(bit_equal(np.array([1.0, -2.0]), np.array([1.0, -2.0]))) def test_zero_denominator_is_unknown(self): self.assertIsNone(fraction(1, 0)) self.assertEqual(fraction(1, 2), .5) def test_actual_stage_ledger_telescopes_without_source_feedback(self): account, first, second = _two_tick_fixture() account(0, first) account(1, second) data, report = account.finish(second['g']) for name, expected in { 'decay': [-1., -3.75, -6.5], 'reset': [-1., -1.75, 0.], 'arrivals': [0., 2., 2.], 'APL': [-.5, -1., -2.]}.items(): np.testing.assert_array_equal(data['actual_g_delta_' + name + '_V'], expected) np.testing.assert_array_equal(data['g_ledger_residual_V'], np.zeros(3)) np.testing.assert_array_equal(data['initial_g_state_area_V_ms'], [0., .2, .4 + .2]) self.assertTrue(report['ledger_roundoff_bound_pass']) self.assertFalse(report['APL_cause_of_CAL_failure_proven']) self.assertFalse(report['full_CAL_all_tick_subset_state_independently_verified']) def test_ref_write_includes_new_spike_and_already_refractory(self): account, first, second = _two_tick_fixture() account(0, first) account(1, second) data, _ = account.finish(second['g']) np.testing.assert_array_equal(data['APL_write_ref_SI'], [.5, .5, 0.]) np.testing.assert_array_equal(data['APL_write_available_SI'], [0., .5, 2.]) np.testing.assert_array_equal(data['APL_write_SI'], data['APL_write_ref_SI'] + data['APL_write_available_SI']) np.testing.assert_array_equal(data['refractory_ticks'], [2., 1., 0.]) def test_first_spike_tick_arrival_and_apl_are_not_prior_causes(self): account, first, second = _two_tick_fixture() account(0, first) account(1, second) data, _ = account.finish(second['g']) np.testing.assert_array_equal(data['first_spike_tick'], [0, 1, -1]) np.testing.assert_array_equal(data['APL_pre_first_spike_SI'], [0., .5, 2.]) np.testing.assert_array_equal(data['APL_same_tick_first_spike_SI'], [.25, .5, 0.]) np.testing.assert_array_equal(data['accepted_exc_before_first_SI'], [0., 2., 3.]) np.testing.assert_array_equal(data['accepted_inh_before_first_abs_SI'], [0., 0., 1.]) def test_group_contact_sign_and_rejected_edge_accounting(self): account, first, second = _two_tick_fixture() account(0, first) account(1, second) data, report = account.finish(second['g']) for name in GROUPS: np.testing.assert_array_equal(data[name + '_due_edges'], data[name + '_accepted_edges'] + data[name + '_rejected_edges']) np.testing.assert_array_equal(data[name + '_due_contacts'], data[name + '_accepted_contacts'] + data[name + '_rejected_contacts']) np.testing.assert_array_equal(data['ALPN_accepted_exc_SI'], [0., 2., 0.]) np.testing.assert_array_equal(data['ALPN_rejected_exc_SI'], [1., 0., 0.]) np.testing.assert_array_equal(data['MBON_accepted_inh_abs_SI'], [0., 0., 1.]) np.testing.assert_array_equal(data['MBON_rejected_inh_abs_SI'], [2., 0., 0.]) self.assertEqual(report['strata']['never_active']['neurons'], 1) self.assertEqual(report['strata']['never_active']['no_accepted_ALPN_exc_neurons'], 1) def test_zeroed_apl_keeps_anatomical_edge_and_contact_counts(self): account, first, second = _two_tick_fixture() account(0, first) account(1, second) data, _ = account.finish(second['g']) np.testing.assert_array_equal(data['APL_zeroed_due_edges'], [0., 0., 1.]) np.testing.assert_array_equal(data['APL_zeroed_due_contacts'], [0., 0., 16.]) np.testing.assert_array_equal(data['APL_accepted_edges'], [0, 0, 1]) np.testing.assert_array_equal(data['APL_accepted_exc_SI'], np.zeros(3)) np.testing.assert_array_equal(data['APL_accepted_inh_abs_SI'], np.zeros(3)) def test_observer_never_modifies_supplied_callback_arrays(self): account, first, _ = _two_tick_fixture() original = {name: value.copy() for name, value in first.items()} account(0, first) for name, value in original.items(): self.assertTrue(bit_equal(first[name], value), name) def test_independent_spike_disagreement_stops_before_accounting(self): account, first, _ = _two_tick_fixture() first['fired'][0] = False with self.assertRaisesRegex(ValueError, 'KC spike divergence at tick 0'): account(0, first) self.assertEqual(account.callback_count, 0) def test_dense_fixture_history_disagreement_is_not_tolerated(self): account, first, _ = _two_tick_fixture() account.histories = {'pre_g': first['g_pre_apl'][None, :].copy()} account.histories['pre_g'][0, 1] = np.nextafter(account.histories['pre_g'][0, 1], np.inf) with self.assertRaisesRegex(ValueError, 'Fixture dense pre_g divergence'): account(0, first) def test_cancellation_reports_roundoff_instead_of_forcing_source_sum(self): account = _account([10], [0, 0], [2. ** 53, -(2. ** 53)], [1, 1], [2, 3], gd=1.) stage = _stage([1.], [1.], [1.], [0.], [0.], available=[1], fired=[0], ref=[0], due=[0, 1], accepted=[1, 1]) account(0, stage) data, report = account.finish(stage['g']) np.testing.assert_array_equal(data['arrival_rounding_V'], [-1.]) self.assertEqual(report['source_shadow_max_residual_V'], 1.) np.testing.assert_array_equal(data['initial_g_state_area_V_ms'], [.1]) np.testing.assert_array_equal(data['source_g_area_residual_V_ms'], [.1]) self.assertEqual(report['source_area_max_residual_V_ms'], .1) np.testing.assert_array_equal(data['g_ledger_residual_V'], [0.]) def test_initial_g_is_not_misattributed_to_a_source_group(self): account = _account([10]) stage = _stage([2.], [1.], [1.], [1.], [0.], available=[1], fired=[0], ref=[0]) account(0, stage) data, report = account.finish(stage['g']) for key, value in data.items(): if key.startswith('source_g_area_'): np.testing.assert_array_equal(value, [0.]) self.assertEqual(report['source_shadow_max_residual_V'], 0.) np.testing.assert_array_equal(data['actual_g_delta_decay_V'], [-1.]) np.testing.assert_array_equal(data['initial_g_state_area_V_ms'], [.1]) np.testing.assert_array_equal(data['source_g_area_residual_V_ms'], [0.]) for stratum in ('all', 'never_active'): self.assertEqual(report['strata'][stratum]['initial_g_state_area_V_ms'], .1) self.assertEqual(report['strata'][stratum]['source_g_area_residual_V_ms'], 0.) self.assertEqual(report['strata']['ever_active']['initial_g_state_area_V_ms'], 0.) self.assertEqual(report['ledger_accumulator_mantissa_bits'], np.finfo(np.longdouble).nmant) self.assertIn('eps(longdouble)', report['ledger_bound_formula']) def test_wrong_final_g_fails_ledger_closure(self): account, first, second = _two_tick_fixture() account(0, first) account(1, second) with self.assertRaisesRegex(AssertionError, 'g ledger residual'): account.finish(second['g'] + 1.) def test_synthetic_replay_accounting_integration_keeps_delay_and_stage_order(self): from replay_core import IncomingIndex, replay targets = np.asarray([10, 20], np.int32) expected = _events([0, 19], [10, 20]) coeff = {'gd': .5, 'c0': 0., 'cg': 1., 'cv': 1., 'threshold': 1., 'reset': 0., 'seed_volt': 2.} incoming = IncomingIndex(np.asarray([10], np.int32), np.asarray([1], np.int32), np.asarray([123], np.int64), np.asarray([2.], np.float64), 2) account = DriveAccounting(targets, incoming, np.asarray([5], np.int64), np.asarray([0], np.int32), GROUPS, expected, coeff, np.asarray([1, 1], np.int32)) result = replay(np.zeros(2, np.float64), np.zeros(2, np.float64), coeff, 24, targets, _events([0], [10]), expected, incoming, np.zeros((24, 2), np.float32), callback=account) data, report = account.finish(result['g']) np.testing.assert_array_equal(result['step'], [0, 19]) np.testing.assert_array_equal(result['neuron_idx'], targets) np.testing.assert_array_equal(result['counts'], [1, 1]) np.testing.assert_array_equal(result['ref'], [0, 18]) np.testing.assert_array_equal(data['first_spike_tick'], [0, 19]) np.testing.assert_array_equal(data['KC_accepted_contacts'], [0, 5]) np.testing.assert_array_equal(data['actual_g_delta_arrivals_V'], [0., 2.]) np.testing.assert_array_equal(data['actual_g_delta_decay_V'], [0., -1.]) np.testing.assert_array_equal(data['actual_g_delta_reset_V'], [0., -1.]) np.testing.assert_array_equal(data['source_g_area_KC_exc_V_ms'], [0., .2]) self.assertEqual(account.callback_count, 24) self.assertTrue(report['ledger_roundoff_bound_pass']) if __name__ == '__main__': unittest.main()