Digital Drosophila Archive
135 lines
6.3 KiB
Python
135 lines
6.3 KiB
Python
"""Pure synthetic checks for stateless pre-first event validation."""
|
|
import unittest
|
|
|
|
import numpy as np
|
|
|
|
from validate_events import COLUMN_DTYPES, N_GLOBAL, validate_batch
|
|
|
|
|
|
def fixture():
|
|
reference = dict(edge_ids=np.array([7, 20, 99], np.int64),
|
|
srcglobals=np.array([3, 4, 3], np.int32), dstglobals=np.array([10, 10, 20], np.int32),
|
|
weights=np.array([.001, -.002, .003], np.float64),
|
|
target_globals=np.array([10, 20, 30], np.int32), target_first=np.array([30, -1, 0], np.int32),
|
|
source_event_keys=np.array([3, 4, 4 * N_GLOBAL + 3], np.int64))
|
|
# Arbitrary order, repeated edge at a later tick, negative and positive.
|
|
values = dict(edge_id=[99, 20, 7, 99], source_neuron_idx=[3, 4, 3, 3],
|
|
target_neuron_idx=[20, 10, 10, 20], source_spike_tick=[4, 0, 0, 0],
|
|
arrival_tick=[22, 18, 18, 18], weight_sign=[1, -1, 1, 1],
|
|
ref_at_arrival=[0] * 4, weight_SI=[.003, -.002, .001, .003], accepted=[True] * 4)
|
|
columns = {name: np.array(values[name], dtype) for name, dtype in COLUMN_DTYPES.items()}
|
|
return columns, reference
|
|
|
|
|
|
class EventValidationTests(unittest.TestCase):
|
|
def test_unordered_rows_real_edges_signed_weights_and_never_first(self):
|
|
columns, ref = fixture()
|
|
original = {k: v.copy() for k, v in columns.items()}
|
|
result = validate_batch(columns, **ref)
|
|
self.assertEqual(result['count'], 4)
|
|
np.testing.assert_array_equal(result['per_target_count'], [2, 2, 0])
|
|
self.assertEqual(result['per_target_count'].dtype, np.int64)
|
|
self.assertFalse(result['unique_cause_claim'])
|
|
self.assertFalse(result['cross_batch_completeness_checked'])
|
|
for key in columns:
|
|
np.testing.assert_array_equal(columns[key], original[key])
|
|
|
|
def test_empty_valid_but_missing_wrong_dtype_or_shape_still_rejected(self):
|
|
columns, ref = fixture()
|
|
empty = {name: value[:0] for name, value in columns.items()}
|
|
result = validate_batch(empty, **ref)
|
|
self.assertEqual(result['count'], 0)
|
|
np.testing.assert_array_equal(result['per_target_count'], [0, 0, 0])
|
|
bad = empty.copy()
|
|
bad['arrival_tick'] = np.empty(0, np.int64)
|
|
with self.assertRaises(ValueError):
|
|
validate_batch(bad, **ref)
|
|
bad = empty.copy()
|
|
del bad['accepted']
|
|
with self.assertRaises(ValueError):
|
|
validate_batch(bad, **ref)
|
|
bad = empty.copy()
|
|
bad['weight_sign'] = np.empty((0, 1), np.int32)
|
|
with self.assertRaises(ValueError):
|
|
validate_batch(bad, **ref)
|
|
|
|
def test_every_edge_field_and_weight_bit_is_checked(self):
|
|
for field, value in [('edge_id', 8), ('source_neuron_idx', 4), ('target_neuron_idx', 10),
|
|
('weight_sign', -1), ('weight_SI', np.nextafter(.003, np.inf))]:
|
|
with self.subTest(field=field):
|
|
columns, ref = fixture()
|
|
columns[field][0] = value
|
|
with self.assertRaises(ValueError):
|
|
validate_batch(columns, **ref)
|
|
columns, ref = fixture()
|
|
ref['weights'][2] = 0.
|
|
columns['weight_SI'][[0, 3]] = -0.
|
|
with self.assertRaisesRegex(ValueError, 'bits mismatch'):
|
|
validate_batch(columns, **ref)
|
|
columns['weight_SI'][[0, 3]] = 0.
|
|
with self.assertRaisesRegex(ValueError, 'zero-weight'):
|
|
validate_batch(columns, **ref)
|
|
|
|
def test_guard_strict_prefirst_and_source_delay(self):
|
|
changes = [('accepted', False), ('ref_at_arrival', 1), ('source_spike_tick', -1),
|
|
('source_spike_tick', 3), ('arrival_tick', 21)]
|
|
for field, value in changes:
|
|
with self.subTest(field=field):
|
|
columns, ref = fixture()
|
|
columns[field][0] = value
|
|
with self.assertRaises(ValueError):
|
|
validate_batch(columns, **ref)
|
|
for first in (22, 21):
|
|
columns, ref = fixture()
|
|
ref['target_first'][1] = first
|
|
with self.assertRaisesRegex(ValueError, 'strictly before'):
|
|
validate_batch(columns, **ref)
|
|
|
|
def test_source_event_membership_and_target_population(self):
|
|
columns, ref = fixture()
|
|
ref['source_event_keys'] = ref['source_event_keys'][:2]
|
|
with self.assertRaisesRegex(ValueError, 'source spike'):
|
|
validate_batch(columns, **ref)
|
|
columns, ref = fixture()
|
|
ref['target_globals'] = np.array([10, 30], np.int32)
|
|
ref['target_first'] = np.array([30, 0], np.int32)
|
|
with self.assertRaisesRegex(ValueError, 'target global'):
|
|
validate_batch(columns, **ref)
|
|
|
|
def test_duplicate_rows_rejected_without_sort_assumption(self):
|
|
columns, ref = fixture()
|
|
for value in columns.values():
|
|
value[0] = value[3]
|
|
with self.assertRaisesRegex(ValueError, 'Duplicate edge event'):
|
|
validate_batch(columns, **ref)
|
|
|
|
def test_invalid_reference_schema_order_duplicate_range_nonfinite(self):
|
|
variants = [dict(edge_ids=np.array([7, 20, 99], np.int32)),
|
|
dict(edge_ids=np.array([7, 99, 20], np.int64)),
|
|
dict(edge_ids=np.array([7, 20, 20], np.int64)),
|
|
dict(srcglobals=np.array([3, 4, N_GLOBAL], np.int32)),
|
|
dict(weights=np.array([.001, -.002, np.inf])),
|
|
dict(target_first=np.array([30, -2, 0], np.int32)),
|
|
dict(source_event_keys=np.array([3, 3, 4], np.int64)),
|
|
dict(source_event_keys=np.array([4, 3], np.int64)),
|
|
dict(source_event_keys=np.array([-1, 3], np.int64))]
|
|
for change in variants:
|
|
with self.subTest(change=change):
|
|
columns, ref = fixture()
|
|
with self.assertRaises(ValueError):
|
|
validate_batch(columns, **(ref | change))
|
|
|
|
def test_optional_design_and_local_row_fields_are_not_assumed(self):
|
|
columns, ref = fixture()
|
|
columns['incoming_edge_row'] = np.array([1000, 2, 1, 1000], np.int64)
|
|
columns['target_is_design'] = np.array([True, False, True, False])
|
|
self.assertEqual(validate_batch(columns, **ref)['count'], 4)
|
|
bad = columns.copy()
|
|
bad['accepted'] = columns['accepted'].astype(np.uint8)
|
|
with self.assertRaises(ValueError):
|
|
validate_batch(bad, **ref)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main(verbosity=2)
|