Digital Drosophila Archive
243 lines
12 KiB
Python
243 lines
12 KiB
Python
"""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)
|