Digital Drosophila Archive
247 lines
13 KiB
Python
247 lines
13 KiB
Python
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()
|