Digital Drosophila Archive
124 lines
6.7 KiB
Python
124 lines
6.7 KiB
Python
"""Stateless independent checks for accepted pre-first anatomical event rows.
|
|
|
|
Pure NumPy: no files or brain execution. Passing establishes row consistency
|
|
with the supplied graph and recorded source events, not necessity/unique cause.
|
|
The caller owns provenance, design labels, trial bounds, cross-batch duplicate
|
|
detection, expected totals, and completeness of the full record collection.
|
|
"""
|
|
from collections.abc import Mapping
|
|
|
|
import numpy as np
|
|
|
|
|
|
N_GLOBAL = 166700
|
|
DELAY_STEPS = 18
|
|
COLUMN_DTYPES = {
|
|
'edge_id': np.int64,
|
|
'source_neuron_idx': np.int32,
|
|
'target_neuron_idx': np.int32,
|
|
'source_spike_tick': np.int32,
|
|
'arrival_tick': np.int32,
|
|
'weight_sign': np.int32,
|
|
'ref_at_arrival': np.int32,
|
|
'weight_SI': np.float64,
|
|
'accepted': np.bool_,
|
|
}
|
|
|
|
|
|
def _require(condition, message):
|
|
if not condition:
|
|
raise ValueError(message)
|
|
|
|
|
|
def _vector(value, dtype, name, length=None):
|
|
_require(isinstance(value, np.ndarray), name + ' must be a NumPy array')
|
|
_require(value.dtype == np.dtype(dtype) and value.ndim == 1, name + ' has the wrong dtype/shape')
|
|
_require(length is None or len(value) == length, name + ' has the wrong length')
|
|
return value
|
|
|
|
|
|
def _globals(value, name, length=None, sorted_unique=False):
|
|
value = _vector(value, np.int32, name, length)
|
|
_require(np.all((value >= 0) & (value < N_GLOBAL)), name + ' contains an out-of-range global ID')
|
|
if sorted_unique:
|
|
_require(np.all(value[1:] > value[:-1]), name + ' must be sorted and unique')
|
|
return value
|
|
|
|
|
|
def _lookup(sorted_reference, wanted, name):
|
|
positions = np.searchsorted(sorted_reference, wanted)
|
|
_require(np.all(positions < len(sorted_reference)), name + ' is absent from the reference')
|
|
_require(np.array_equal(sorted_reference[positions], wanted), name + ' is absent from the reference')
|
|
return positions
|
|
|
|
|
|
def validate_batch(columns, *, edge_ids, srcglobals, dstglobals, weights,
|
|
target_globals, target_first, source_event_keys):
|
|
"""Validate arbitrary-order rows; return count and target-indexed int64 counts.
|
|
|
|
Required columns and exact dtypes are COLUMN_DTYPES. Additional columns are
|
|
intentionally not interpreted: in particular incoming_edge_row has a
|
|
caller-specific coordinate system, and design/source labels are separate.
|
|
References: sorted unique nonnegative edge_ids:int64[E]; source/destination
|
|
globals:int32[E]; finite effective SI weights:float64[E]; sorted unique
|
|
target_globals:int32[K]; target_first:int32[K] (-1 means never fired);
|
|
source_event_keys:int64 sorted unique, encoding tick*166700+global ID.
|
|
|
|
Every row must reference a real nonzero edge, its exact weight bits/sign,
|
|
accepted=True/ref_at_arrival=0, a recorded source event exactly 18 ticks
|
|
earlier, and arrival strictly before the target's first spike, if any.
|
|
Duplicate (arrival_tick,edge_id) rows within this batch are rejected without
|
|
assuming any input row ordering. Cross-batch duplicates require caller state.
|
|
Empty batches still require valid complete schema/references and return zero.
|
|
No supplied array is modified or sorted in place.
|
|
"""
|
|
_require(isinstance(columns, Mapping), 'columns must be a mapping')
|
|
_require(set(COLUMN_DTYPES).issubset(columns), 'Missing required event columns')
|
|
ids = _vector(edge_ids, np.int64, 'edge_ids')
|
|
e = len(ids)
|
|
_require(np.all(ids >= 0) and np.all(ids[1:] > ids[:-1]), 'edge_ids must be sorted unique nonnegative IDs')
|
|
sources = _globals(srcglobals, 'srcglobals', e)
|
|
destinations = _globals(dstglobals, 'dstglobals', e)
|
|
weight = _vector(weights, np.float64, 'weights', e)
|
|
_require(np.isfinite(weight).all(), 'Nonfinite reference weight')
|
|
targets = _globals(target_globals, 'target_globals', sorted_unique=True)
|
|
first = _vector(target_first, np.int32, 'target_first', len(targets))
|
|
_require(np.all(first >= -1), 'target_first must be -1 or a nonnegative tick')
|
|
keys = _vector(source_event_keys, np.int64, 'source_event_keys')
|
|
largest_key = int(np.iinfo(np.int32).max) * N_GLOBAL + N_GLOBAL - 1
|
|
_require(np.all((keys >= 0) & (keys <= largest_key)) and np.all(keys[1:] > keys[:-1]),
|
|
'source_event_keys must be sorted unique valid int32-tick/global encodings')
|
|
count = len(_vector(columns['edge_id'], np.int64, 'columns.edge_id'))
|
|
data = {name: _vector(columns[name], dtype, 'columns.' + name, count)
|
|
for name, dtype in COLUMN_DTYPES.items()}
|
|
_require(np.isfinite(data['weight_SI']).all(), 'Nonfinite recorded weight')
|
|
_require(np.all(data['accepted']) and np.all(data['ref_at_arrival'] == 0),
|
|
'Recorded arrival was not accepted at ref=0')
|
|
row_edge = _lookup(ids, data['edge_id'], 'Recorded edge ID')
|
|
_require(np.array_equal(data['source_neuron_idx'], sources[row_edge]), 'Recorded edge source mismatch')
|
|
_require(np.array_equal(data['target_neuron_idx'], destinations[row_edge]), 'Recorded edge target mismatch')
|
|
actual_weight = weight[row_edge]
|
|
_require(np.array_equal(np.ascontiguousarray(data['weight_SI']).view(np.uint64),
|
|
np.ascontiguousarray(actual_weight).view(np.uint64)), 'Recorded weight bits mismatch')
|
|
_require(np.all(actual_weight != 0), 'Pre-first nonzero-event table contains a zero-weight edge')
|
|
_require(np.array_equal(data['weight_sign'], np.sign(actual_weight).astype(np.int32)), 'Recorded weight sign mismatch')
|
|
local_targets = _lookup(targets, data['target_neuron_idx'], 'Recorded target global')
|
|
arrival = data['arrival_tick'].astype(np.int64)
|
|
source_tick = data['source_spike_tick'].astype(np.int64)
|
|
_require(np.all(source_tick >= 0) and np.array_equal(source_tick, arrival - DELAY_STEPS),
|
|
'Source/arrival delay is not exactly 18 nonnegative ticks')
|
|
wanted_keys = source_tick * np.int64(N_GLOBAL) + data['source_neuron_idx'].astype(np.int64)
|
|
_lookup(keys, wanted_keys, 'Recorded source spike')
|
|
target_first_rows = first[local_targets].astype(np.int64)
|
|
_require(np.all((target_first_rows == -1) | (arrival < target_first_rows)),
|
|
'Arrival is not strictly before the observed first target spike')
|
|
if count > 1:
|
|
order = np.lexsort((data['edge_id'], arrival))
|
|
sorted_arrival, sorted_edge = arrival[order], data['edge_id'][order]
|
|
_require(not np.any((sorted_arrival[1:] == sorted_arrival[:-1]) &
|
|
(sorted_edge[1:] == sorted_edge[:-1])), 'Duplicate edge event within batch')
|
|
per_target = np.bincount(local_targets, minlength=len(targets)).astype(np.int64, copy=False)
|
|
_require(int(per_target.sum()) == count, 'Target counts do not cover the batch')
|
|
return dict(pass_all=True, count=count, per_target_count=per_target,
|
|
unique_cause_claim=False, cross_batch_completeness_checked=False)
|