"""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)