Digital Drosophila Archive
215 lines
13 KiB
Python
215 lines
13 KiB
Python
"""Pure synthetic records, deliberately not a neural simulation or file test."""
|
|
import hashlib
|
|
import json
|
|
import unittest
|
|
|
|
import numpy as np
|
|
|
|
import engineering_scopes as s
|
|
|
|
|
|
def sha(array):
|
|
return hashlib.sha256(np.ascontiguousarray(array).tobytes()).hexdigest()
|
|
|
|
|
|
def record(start, stop, *, reference=True, blank_requests=False):
|
|
n, k, length = 3, 2, stop - start
|
|
ticks = np.array([0, 7, 19, 21, 41, 61], np.int32)
|
|
nodes = np.array([0, 1, 2, 0, 2, 1], np.int32)
|
|
selected = (ticks >= start) & (ticks < stop)
|
|
event_ticks, event_nodes = ticks[selected] - np.int32(start), nodes[selected].copy()
|
|
v = np.array([-.05, -.04, -.03], np.float64)
|
|
g = np.array([-0., float(stop - 1), -.001], np.float64)
|
|
ref = np.zeros(n, np.int32)
|
|
state = dict(step=event_ticks, neuron_idx=event_nodes, counts=np.bincount(event_nodes, minlength=n).astype(np.int32),
|
|
v=v, g=g, ref=ref)
|
|
raw_global = np.zeros((19, n), np.uint8)
|
|
for tick, node in zip(ticks, nodes):
|
|
if max(0, stop - 19) <= tick < stop:
|
|
raw_global[tick % 19, node] = 1
|
|
normalized = raw_global[(np.arange(19) + stop % 19) % 19]
|
|
raw_local = normalized[(np.arange(19) - length % 19) % 19].copy()
|
|
generated_all = np.repeat(np.arange(1, stop + 1, dtype=np.float32)[:, None], k, axis=1)
|
|
applied_all = np.concatenate((np.zeros((18, k), np.float32), generated_all))[:stop]
|
|
pending = np.zeros((18, k), np.float32)
|
|
kept = min(stop, 18); pending[-kept:] = generated_all[-kept:]
|
|
apl_state = np.full(2, np.float32(stop + 1), np.float32)
|
|
terminal = dict(spike_ring=raw_local, apl_state=apl_state, apl_pending=pending, next_tick=np.asarray(length, np.int64))
|
|
trace = dict(generated=generated_all[start:stop].copy(), applied=applied_all[start:stop].copy(),
|
|
drive=np.repeat(np.arange(start, stop, dtype=np.float32)[:, None], 2, axis=1),
|
|
state=np.repeat(np.arange(start + 2, stop + 2, dtype=np.float32)[:, None], 2, axis=1))
|
|
hashes, queues = [], []
|
|
for local, global_tick in enumerate(range(start, stop)):
|
|
hashes.append(dict(tick=local, v_sha256=sha(v), g_sha256=sha(np.array([-0., float(global_tick), -.001], np.float64)),
|
|
ref_sha256=sha(ref)))
|
|
edges, accepted = np.array([global_tick], np.int64), np.array([global_tick % 2], np.uint8)
|
|
queues.append(dict(tick=local, count=1, accepted_count=int(accepted.sum()), ordered_edge_sha256=sha(edges),
|
|
accepted_mask_sha256=sha(accepted), joint_sha256=hashlib.sha256(
|
|
str(local).encode() + edges.tobytes() + accepted.tobytes()).hexdigest()))
|
|
requests = np.array([[int(t), int(i), 1, 1, 0] for t, i in zip(event_ticks, event_nodes)], np.int64).reshape(-1, 5)
|
|
if blank_requests:
|
|
requests = np.empty((0, 5), np.int64)
|
|
summary = dict(complete=True, pass_all=True, all_finite=True, steps=length, completed_ticks=length, nodes=n,
|
|
precision='float64', apl_precision='float32', apl_delay_steps=18,
|
|
checkpoint_continuation_used=start > 0, global_tick_offset=start, global_next_tick=stop,
|
|
effective_weights_raw_sha256='e' * 64, diagnostic_mode='reference' if reference else 'light',
|
|
full_APL_history_recorded=reference, full_tick_state_hashes_recorded=reference,
|
|
full_queue_hashes_recorded=reference)
|
|
return dict(source='synthetic', state=state, terminal=terminal, requests=requests, steps=length,
|
|
global_tick_offset=start, summary=summary, traces=trace if reference else {},
|
|
state_hashes=hashes if reference else None, queues=queues if reference else None,
|
|
terminal_origin='synthetic_recorded')
|
|
|
|
|
|
class SliceTests(unittest.TestCase):
|
|
def test_reference_slice_all_observations_exact_and_inputs_unchanged(self):
|
|
candidate, whole = record(20, 44), record(0, 64)
|
|
before = candidate['terminal']['spike_ring'].tobytes(), whole['requests'].tobytes()
|
|
report = s.compare_slice(candidate, whole)
|
|
self.assertTrue(report['numerical_exact'], report['failed_checks'])
|
|
self.assertTrue(report['observed_scope']['candidate_per_tick_state_hashes'])
|
|
self.assertTrue(report['observed_scope']['candidate_per_tick_queue_fields'])
|
|
self.assertFalse(report['observed_scope']['queue_joint_hashes_compared'])
|
|
self.assertNotEqual(candidate['queues'][0]['joint_sha256'], whole['queues'][20]['joint_sha256'])
|
|
self.assertEqual(before, (candidate['terminal']['spike_ring'].tobytes(), whole['requests'].tobytes()))
|
|
json.dumps(report, allow_nan=False)
|
|
|
|
def test_light_uses_final_raw_hash_but_does_not_fabricate_optional_history(self):
|
|
report = s.compare_slice(record(20, 44, reference=False), record(0, 64))
|
|
self.assertTrue(report['numerical_exact'])
|
|
self.assertTrue(report['observed_scope']['final_state_against_reference_tick_hash'])
|
|
for name in ('candidate_APL_histories', 'candidate_per_tick_state_hashes', 'candidate_per_tick_queue_fields'):
|
|
self.assertFalse(report['observed_scope'][name])
|
|
self.assertNotIn('slice.per_tick_state_hashes', report['checks'])
|
|
self.assertTrue(any('unobserved' in item for item in report['limitations']))
|
|
|
|
def test_short_fresh_positive_zero_padding_and_prefix_events(self):
|
|
self.assertTrue(s.compare_slice(record(0, 5), record(0, 64))['numerical_exact'])
|
|
candidate = record(0, 5, reference=False)
|
|
candidate['terminal']['apl_pending'][0, 0] = np.float32(-0.)
|
|
report = s.compare_slice(candidate, record(0, 64))
|
|
self.assertFalse(report['checks']['slice.APL_pending'])
|
|
|
|
def test_blank_requests_are_valid_but_wrong_acceptance_is_not_equal(self):
|
|
self.assertTrue(s.compare_slice(record(20, 44, blank_requests=True), record(0, 64, blank_requests=True))['numerical_exact'])
|
|
candidate = record(20, 44)
|
|
candidate['requests'][0, 3:] = [0, 1]
|
|
report = s.compare_slice(candidate, record(0, 64))
|
|
self.assertFalse(report['checks']['slice.requests_five_fields'])
|
|
|
|
def test_wrong_count_or_final_hash_binding_is_invalid_not_a_pass(self):
|
|
for mutation in ('count', 'terminal', 'last_hash'):
|
|
candidate = record(20, 44)
|
|
if mutation == 'count': candidate['state']['counts'][0] += 1
|
|
elif mutation == 'terminal': candidate['state']['g'][1] += 1.
|
|
else: candidate['state_hashes'][-1]['g_sha256'] = '0' * 64
|
|
with self.subTest(mutation=mutation), self.assertRaises(ValueError):
|
|
s.compare_slice(candidate, record(0, 64))
|
|
|
|
def test_signed_zero_light_terminal_vs_reference_hash_is_detected(self):
|
|
candidate = record(20, 44, reference=False)
|
|
candidate['state']['g'][0] = 0.
|
|
report = s.compare_slice(candidate, record(0, 64))
|
|
self.assertFalse(report['checks']['slice.final_g_raw_hash'])
|
|
|
|
def test_interior_trace_state_hash_and_queue_mismatch_are_not_hidden(self):
|
|
candidate = record(20, 44)
|
|
candidate['traces']['drive'][1, 0] += np.float32(1.)
|
|
candidate['state_hashes'][1]['v_sha256'] = '0' * 64
|
|
candidate['queues'][1]['accepted_mask_sha256'] = '1' * 64
|
|
report = s.compare_slice(candidate, record(0, 64))
|
|
for key in ('slice.APL_trace.drive', 'slice.per_tick_state_hashes', 'slice.queue_order_and_acceptance'):
|
|
self.assertFalse(report['checks'][key])
|
|
|
|
def test_missing_reference_observations_and_invalid_bounds_fail_closed(self):
|
|
for mutation in ('light_mono', 'light_labeled_full_mono', 'mono_hashes', 'candidate_traces', 'partial_trace', 'out_of_bounds', 'nonzero_mono'):
|
|
candidate, mono = record(20, 44), record(0, 64)
|
|
if mutation == 'light_mono': mono = record(0, 64, reference=False)
|
|
elif mutation == 'light_labeled_full_mono': mono['summary']['diagnostic_mode'] = 'light'
|
|
elif mutation == 'mono_hashes': mono['state_hashes'] = None
|
|
elif mutation == 'candidate_traces': candidate['traces'] = {}
|
|
elif mutation == 'partial_trace': candidate['traces'].pop('drive')
|
|
elif mutation == 'out_of_bounds': candidate = record(60, 65)
|
|
else: mono = record(1, 65)
|
|
with self.subTest(mutation=mutation), self.assertRaises(ValueError):
|
|
s.compare_slice(candidate, mono)
|
|
|
|
def test_summary_type_scope_nonfinite_and_paths_rejected(self):
|
|
for mutation in ('bool', 'failed', 'offset', 'mode', 'weight', 'nan'):
|
|
candidate = record(20, 44)
|
|
if mutation == 'bool': candidate['summary']['steps'] = True
|
|
elif mutation == 'failed': candidate['summary']['pass_all'] = False
|
|
elif mutation == 'offset': candidate['summary']['global_tick_offset'] = 0
|
|
elif mutation == 'mode': candidate['summary']['diagnostic_mode'] = 'unknown'
|
|
elif mutation == 'weight': candidate['summary'].pop('effective_weights_raw_sha256')
|
|
else: candidate['state']['g'][0] = np.nan
|
|
with self.subTest(mutation=mutation), self.assertRaises(ValueError):
|
|
s.compare_slice(candidate, record(0, 64))
|
|
with self.assertRaises(ValueError): s.compare_slice('/never/open/this', record(0, 64))
|
|
|
|
|
|
class CheckpointTests(unittest.TestCase):
|
|
def test_local_phase_normalization_without_segment_cumulative_counts_confusion(self):
|
|
candidate, prefix = record(20, 64), record(0, 64)
|
|
self.assertFalse(s.ec.bit_equal(candidate['terminal']['spike_ring'], prefix['terminal']['spike_ring']))
|
|
self.assertFalse(s.ec.bit_equal(candidate['state']['counts'], prefix['state']['counts']))
|
|
report = s.compare_checkpoint(candidate, prefix)
|
|
self.assertTrue(report['numerical_exact'], report['failed_checks'])
|
|
self.assertFalse(report['observed_scope']['counts_compared'])
|
|
self.assertFalse(report['observed_scope']['per_tick_state_histories_compared'])
|
|
|
|
def test_signed_zero_different_stop_or_effective_weights_cannot_pass(self):
|
|
for mutation in ('zero', 'stop', 'weight'):
|
|
candidate, prefix = record(20, 64, reference=False), record(0, 64, reference=False)
|
|
if mutation == 'zero': candidate['state']['g'][0] = 0.
|
|
elif mutation == 'stop': prefix = record(0, 63, reference=False)
|
|
else: candidate['summary']['effective_weights_raw_sha256'] = 'f' * 64
|
|
with self.subTest(mutation=mutation):
|
|
self.assertFalse(s.compare_checkpoint(candidate, prefix)['numerical_exact'])
|
|
|
|
|
|
class ParentCarryTests(unittest.TestCase):
|
|
def test_short_parent_carry_observed_and_json_safe(self):
|
|
report = s.compare_parent_carry(record(20, 25), record(0, 20))
|
|
self.assertTrue(report['numerical_exact'], report['failed_checks'])
|
|
self.assertEqual(report['observed_scope']['unwritten_spike_slot_count'], 14)
|
|
self.assertEqual(report['observed_scope']['unconsumed_APL_row_count'], 13)
|
|
self.assertEqual(report['observed_scope']['applied_parent_APL_rows_compared'], 5)
|
|
json.dumps(report, allow_nan=False)
|
|
|
|
def test_unwritten_or_unconsumed_corruption_detected_even_if_terminal_is_self_consistent(self):
|
|
for mutation in ('ring', 'pending', 'applied'):
|
|
candidate = record(20, 25)
|
|
if mutation == 'ring': candidate['terminal']['spike_ring'][10, 0] ^= np.uint8(1)
|
|
elif mutation == 'pending': candidate['terminal']['apl_pending'][0, 0] += np.float32(1.)
|
|
else: candidate['traces']['applied'][0, 0] += np.float32(1.)
|
|
with self.subTest(mutation=mutation):
|
|
self.assertFalse(s.compare_parent_carry(candidate, record(0, 20))['numerical_exact'])
|
|
|
|
def test_light_missing_optional_applied_trace_is_not_fabricated(self):
|
|
report = s.compare_parent_carry(record(20, 25, reference=False), record(0, 20))
|
|
self.assertTrue(report['numerical_exact'])
|
|
self.assertFalse(report['observed_scope']['applied_parent_APL_prefix_checked'])
|
|
self.assertIsNone(report['observed_scope']['applied_parent_APL_rows_compared'])
|
|
self.assertNotIn('parent.applied_APL_prefix', report['checks'])
|
|
|
|
def test_long_segment_discloses_no_unwritten_slots_and_wrong_parent_clock_fails(self):
|
|
report = s.compare_parent_carry(record(20, 44), record(0, 20))
|
|
self.assertTrue(report['numerical_exact'])
|
|
self.assertFalse(report['observed_scope']['unwritten_spike_slots_checked'])
|
|
self.assertFalse(report['observed_scope']['unconsumed_APL_prefix_checked'])
|
|
self.assertEqual(report['observed_scope']['applied_parent_APL_rows_compared'], 18)
|
|
self.assertFalse(s.compare_parent_carry(record(20, 44), record(0, 19))['numerical_exact'])
|
|
with self.assertRaises(ValueError): s.compare_parent_carry(record(0, 5), record(0, 20))
|
|
|
|
def test_long_light_carry_scope_explicitly_has_no_observed_carry_array(self):
|
|
report = s.compare_parent_carry(record(20, 44, reference=False), record(0, 20))
|
|
self.assertTrue(report['numerical_exact'])
|
|
self.assertFalse(report['observed_scope']['any_carry_values_directly_compared'])
|
|
self.assertTrue(any('clock/shape/weight identity only' in line for line in report['limitations']))
|
|
self.assertFalse(report['complete_dynamical_equivalence_proven'])
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|