"""Synthetic CPU-only tests. No real records, files, Torch, GPU, or model runs.""" import json import sys import unittest import numpy as np from apl_sources import N, GROUPS, DECAY32, GAIN32, decompose_apl_sources def fixture(events=((0, 1), (1, 4), (2, 10), (4, 20)), steps=6): I = np.zeros((2, N), np.float32) I[:, 1] = [1., .3] I[:, 4] = [.5, 2.] I[:, 10] = [1.1, .2] I[:, 20] = [.7, 1.5] O = np.array([[1., .7, .2, 0.], [.3, .5, 1., 2.]], np.float32) pairs = np.asarray(events, np.int32).reshape(-1, 2) tick, node = pairs[:, 0], pairs[:, 1] # This is merely one legal ordinary f32 reduction implementation for # exercising the mathematical bound, not a CUDA-equality oracle. state = np.zeros(2, np.float32) drive_log, state_log, h_log = [], [], [] for t in range(steps): drive = I[:, node[tick == t]].sum(axis=1, dtype=np.float32) state = np.add(np.multiply(DECAY32, state), drive) h = np.multiply(np.matmul(state, O), GAIN32) drive_log.append(drive) state_log.append(state.copy()) h_log.append(h) return dict(I32=I, O32=O, pn_globals=np.array([1, 4, 6], np.int32), kc_globals=np.array([10, 20, 30, 40], np.int32), design_pn_globals=np.array([1], np.int32), event_step=tick, event_node=node, drive32=np.asarray(drive_log), state32=np.asarray(state_log), h32=np.asarray(h_log), decay32=DECAY32, gain32=GAIN32, standard_fp32=True) class AttributionTests(unittest.TestCase): def test_three_groups_cover_drive_and_pass_prior_bound(self): f = fixture() calls = [] out = decompose_apl_sources(**f, callback=lambda t, s: calls.append(s)) summary = out['summary'] self.assertTrue(summary['pass_all']) self.assertTrue(summary['input_support_partition_exact']) np.testing.assert_array_equal(out['first_kc_tick'], [2, 4, -1, -1]) self.assertEqual(len(out['tick_rows']), 6 * 3 * 2) self.assertEqual([summary['groups'][g]['actual_spikes'] for g in GROUPS], [1, 1, 2]) for s in calls: np.testing.assert_array_equal(s['drive_total'], (s['drive_by_group'][0] + s['drive_by_group'][1]) + s['drive_by_group'][2]) np.testing.assert_array_equal(s['state_total'], (s['state_by_group'][0] + s['state_by_group'][1]) + s['state_by_group'][2]) self.assertFalse(summary['CUDA_float32_bitwise_replay']) self.assertFalse(summary['independent_brain_oracle']) self.assertFalse(summary['theoretical_bound']['tolerance_fitted_to_observed_error']) self.assertFalse(summary['cumulative_source_fraction_error_bounds_verified']) self.assertFalse(summary['source_majority_robustness_proven']) json.dumps(summary, allow_nan=False) json.dumps(out['tick_rows'], allow_nan=False) def test_early_excludes_same_tick_and_never_fired_columns(self): f = fixture(events=((0, 1), (2, 10))) out = decompose_apl_sources(**f) rows = [r for r in out['tick_rows'] if r['group'] == 'design_ALPN' and r['apl_index'] == 0] d, q = float(DECAY32), float(GAIN32) self.assertAlmostEqual(rows[0]['h_before_first_KC_sum_mV'], q * 1.) self.assertAlmostEqual(rows[1]['h_before_first_KC_sum_mV'], q * d) self.assertEqual(rows[2]['h_before_first_KC_sum_mV'], 0.) self.assertGreater(rows[2]['h_never_fired_KC_sum_mV'], 0.) self.assertGreater(rows[1]['h_before_first_any_KC_sum_mV'], 0.) self.assertEqual(rows[2]['h_before_first_any_KC_sum_mV'], 0.) def test_first_kc_at_zero_has_no_early_output(self): out = decompose_apl_sources(**fixture(events=((0, 1), (0, 10)))) self.assertEqual(out['summary']['first_any_KC_spike_tick'], 0) self.assertTrue(all(r['h_before_first_any_KC_sum_mV'] == 0 for r in out['tick_rows'])) self.assertTrue(all(r['h_before_first_KC_sum_mV'] == 0 for r in out['tick_rows'])) def test_no_kc_spikes_keep_never_output_separate(self): out = decompose_apl_sources(**fixture(events=((0, 1),))) self.assertIsNone(out['summary']['first_any_KC_spike_tick']) np.testing.assert_array_equal(out['first_kc_tick'], [-1] * 4) for row in out['tick_rows']: self.assertEqual(row['h_sum_mV'], row['h_never_fired_KC_sum_mV']) self.assertEqual(row['h_before_first_KC_sum_mV'], 0.) self.assertEqual(row['h_before_first_any_KC_sum_mV'], 0.) def test_zero_input_exact_zeros_and_undefined_fractions(self): out = decompose_apl_sources(**fixture(events=())) self.assertTrue(out['summary']['pass_all']) for group in out['summary']['groups'].values(): self.assertEqual(group['h_sum_mV'], 0.) self.assertIsNone(group['generated_h_fraction']) self.assertIsNone(group['drive_fraction']) def test_source_first_spike_is_not_first_nonzero_drive(self): out = decompose_apl_sources(**fixture(events=((0, 6), (2, 4), (3, 100)))) group = out['summary']['groups']['non_design_ALPN'] self.assertEqual(group['actual_spikes'], 2) self.assertEqual(group['input_supported_spikes'], 1) self.assertEqual(group['first_actual_spike_tick'], 0) self.assertEqual(group['first_nonzero_drive_tick'], 2) self.assertEqual(out['summary']['actual_events'], 3) self.assertTrue(out['summary']['pass_all']) def test_empty_design_set_reclassifies_pn_not_drops_them(self): f = fixture() f['design_pn_globals'] = np.empty(0, np.int32) out = decompose_apl_sources(**f) self.assertTrue(out['summary']['pass_all']) self.assertEqual(out['summary']['groups']['design_ALPN']['actual_spikes'], 0) self.assertEqual(out['summary']['groups']['non_design_ALPN']['actual_spikes'], 2) def test_chunk_size_does_not_change_source_accounting(self): f = fixture() a = decompose_apl_sources(**f, chunk_size=1) b = decompose_apl_sources(**f, chunk_size=7) self.assertTrue(a['summary']['pass_all'] and b['summary']['pass_all']) self.assertEqual(a['tick_rows'], b['tick_rows']) self.assertEqual(a['summary']['groups'], b['summary']['groups']) for name in ('drive', 'state', 'inhibition'): for key in ('max_abs_error', 'max_error_bound', 'max_bound_utilization', 'bound_violations', 'elements'): self.assertEqual(a['summary']['comparison'][name][key], b['summary']['comparison'][name][key]) def test_callback_and_core_never_modify_inputs_or_running_state(self): f = fixture() inputs = {k: v.copy() for k, v in f.items() if isinstance(v, np.ndarray)} expected = decompose_apl_sources(**f) def mutate(t, s): for value in s.values(): value[:] = 100. actual = decompose_apl_sources(**f, callback=mutate) self.assertEqual(actual['summary'], expected['summary']) self.assertEqual(actual['tick_rows'], expected['tick_rows']) for name, before in inputs.items(): np.testing.assert_array_equal(np.ascontiguousarray(f[name]).view(np.uint8), np.ascontiguousarray(before).view(np.uint8)) def test_per_kc_strict_pre_first_and_never_sums(self): f = fixture(events=((0, 1), (2, 10))) out = decompose_apl_sources(**f, first_kc_tick=np.array([2, -1, -1, -1], np.int32)) per = out['per_kc'] d, q = float(DECAY32), float(GAIN32) first_two = q * (1. + d) * (float(f['I32'][0, 1]) * float(f['O32'][0, 0]) + float(f['I32'][1, 1]) * float(f['O32'][1, 0])) self.assertAlmostEqual(per['design_ALPN_pre_first_h_sum_mV'][0], first_two) np.testing.assert_array_equal(per['design_ALPN_pre_first_h_sum_mV'][1:], [0., 0., 0.]) self.assertEqual(per['design_ALPN_never_fired_h_sum_mV'][0], 0.) np.testing.assert_array_equal(per['design_ALPN_never_fired_h_sum_mV'][1:], per['design_ALPN_h_sum_mV'][1:]) self.assertEqual(per['KC_pre_first_h_sum_mV'][0], 0.) self.assertTrue(out['summary']['fixture_absent_KC_output_is_not_claimed_applied']) def test_supplied_first_times_must_match_actual_logs(self): for first in (np.array([3, 4, -1, -1], np.int32), np.array([2, 4, -1, -1], np.int64)): with self.subTest(first=first), self.assertRaises(ValueError): decompose_apl_sources(**fixture(), first_kc_tick=first) def test_corrupted_positive_trace_violates_unfitted_bound(self): for field in ('drive32', 'state32', 'h32'): with self.subTest(field=field): f = fixture() f[field][1, 0] += np.float32(100.) out = decompose_apl_sources(**f) self.assertFalse(out['summary']['pass_all']) target = {'drive32': 'drive', 'state32': 'state', 'h32': 'inhibition'}[field] self.assertGreater(out['summary']['comparison'][target]['bound_violations'], 0) def test_zero_reference_cannot_accept_nonzero_actual(self): f = fixture(events=()) f['h32'][0, 0] = np.finfo(np.float32).tiny out = decompose_apl_sources(**f) self.assertFalse(out['summary']['pass_all']) self.assertEqual(out['summary']['comparison']['inhibition']['zero_reference_nonzero_actual'], 1) def test_subnormal_disables_theoretical_pass(self): f = fixture(events=()) f['I32'][0, 1] = np.float32(np.finfo(np.float32).tiny / 2) out = decompose_apl_sources(**f) self.assertFalse(out['summary']['pass_all']) self.assertFalse(out['summary']['theoretical_bound']['applicable']) class ValidationTests(unittest.TestCase): def test_nonzero_unknown_source_is_rejected(self): f = fixture() f['I32'][0, 100] = 1. with self.assertRaisesRegex(ValueError, 'outside design/non-design'): decompose_apl_sources(**f) def test_bad_identity_and_nonpartition_are_rejected(self): for key, value in [('design_pn_globals', np.array([10], np.int32)), ('design_pn_globals', np.array([1, 1], np.int32)), ('pn_globals', np.array([1, 4, 10], np.int32)), ('kc_globals', np.array([10, 20, 40, 30], np.int32)), ('pn_globals', np.array([1, 4, 6], np.int64))]: with self.subTest(key=key): f = fixture() f[key] = value with self.assertRaises(ValueError): decompose_apl_sources(**f) def test_invalid_events_dtype_order_duplicate_range(self): changes = [('event_step', np.array([0, 1, 2, 4], np.int64)), ('event_step', np.array([1, 0, 2, 4], np.int32)), ('event_step', np.array([0, 1, 2, 6], np.int32)), ('event_node', np.array([1, 4, 10, N], np.int32))] for key, value in changes: with self.subTest(key=key): f = fixture() f[key] = value with self.assertRaises(ValueError): decompose_apl_sources(**f) f = fixture(events=((0, 1), (0, 1))) with self.assertRaises(ValueError): decompose_apl_sources(**f) def test_invalid_trace_matrix_precision_constants_or_environment(self): variants = [dict(standard_fp32=False), dict(standard_fp32=1), dict(chunk_size=0), dict(decay32=float(DECAY32)), dict(gain32=np.nextafter(GAIN32, np.float32(np.inf))), dict(callback=3)] for change in variants: with self.subTest(change=change), self.assertRaises(ValueError): decompose_apl_sources(**(fixture() | change)) for name in ('I32', 'O32', 'drive32', 'state32', 'h32'): for bad in ('dtype', 'negative', 'nonfinite'): with self.subTest(name=name, bad=bad): f = fixture() if bad == 'dtype': f[name] = f[name].astype(np.float64) else: f[name].flat[0] = -1. if bad == 'negative' else np.nan with self.assertRaises(ValueError): decompose_apl_sources(**f) def test_no_dense_source_history_or_scientific_runtime(self): out = decompose_apl_sources(**fixture()) self.assertEqual(set(out), {'summary', 'tick_rows', 'first_kc_tick', 'per_kc'}) self.assertNotIn('torch', sys.modules) self.assertNotIn('brian2', sys.modules) if __name__ == '__main__': unittest.main(verbosity=2)