Digital Drosophila Archive
180 lines
10 KiB
Python
180 lines
10 KiB
Python
"""Fixed descriptive ALPN-count comparisons; pure NumPy, no files or tuning.
|
|
|
|
Column positions refer to the caller's fixed canonical 686-ALPN order. This
|
|
module cannot assign body IDs or infer anatomical identities. Repeated runs
|
|
are integrity evidence, never additional independent seeds in these metrics.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import itertools
|
|
|
|
import numpy as np
|
|
|
|
|
|
MODES = ('physiological', 'matched_energy')
|
|
ODORS = tuple(range(6))
|
|
SEEDS = tuple(range(51101, 51106))
|
|
N_ALPN = 686
|
|
FIELDS = frozenset(('mode', 'odor_index', 'seed', 'counts', 'design_mask'))
|
|
METRICS = ('cosine_distance', 'pearson_correlation', 'jaccard', 'euclidean_distance')
|
|
|
|
|
|
def _require(condition, message):
|
|
if not condition:
|
|
raise ValueError(message)
|
|
|
|
|
|
def _integer(value, name):
|
|
_require(isinstance(value, (int, np.integer)) and not isinstance(value, (bool, np.bool_)),
|
|
name + ' must be an integer, not a Boolean')
|
|
return int(value)
|
|
|
|
|
|
def _array(value, dtype, name):
|
|
_require(isinstance(value, np.ndarray) and not np.ma.isMaskedArray(value),
|
|
name + ' must be an unmasked NumPy array')
|
|
_require(value.dtype == np.dtype(dtype) and value.shape == (N_ALPN,),
|
|
name + ' must have dtype ' + np.dtype(dtype).name + ' and shape (686,)')
|
|
return np.array(value, copy=True, order='C')
|
|
|
|
|
|
def _pair_metrics(first, second):
|
|
# Cast BEFORE subtraction/norm so int32 extrema do not overflow.
|
|
x, y = first.astype(np.float64), second.astype(np.float64)
|
|
active_x, active_y = first > 0, second > 0
|
|
intersection = int(np.count_nonzero(active_x & active_y))
|
|
union = int(np.count_nonzero(active_x | active_y))
|
|
result = dict(active_a=int(active_x.sum()), active_b=int(active_y.sum()),
|
|
active_intersection=intersection, active_union=union)
|
|
if not len(first):
|
|
result.update({name: None for name in METRICS})
|
|
return result
|
|
norm_x, norm_y = float(np.linalg.norm(x)), float(np.linalg.norm(y))
|
|
cosine = None if norm_x == 0 or norm_y == 0 else float(np.clip(
|
|
np.dot(x, y) / (norm_x * norm_y), -1., 1.))
|
|
# A constant or one-channel vector has undefined Pearson correlation;
|
|
# equality does not turn it into evidence of shared structure.
|
|
correlation = None if bool(np.all(x == x[0]) or np.all(y == y[0])) else float(
|
|
np.clip(np.corrcoef(x, y)[0, 1], -1., 1.))
|
|
result.update(cosine_distance=None if cosine is None else float(1. - cosine),
|
|
pearson_correlation=correlation,
|
|
jaccard=None if union == 0 else float(intersection / union),
|
|
euclidean_distance=float(np.linalg.norm(x - y)))
|
|
_require(all(value is None or np.isfinite(value) for value in result.values()),
|
|
'A structure metric became nonfinite')
|
|
return result
|
|
|
|
|
|
def _summary(rows):
|
|
result = {}
|
|
for name in METRICS:
|
|
values = np.asarray([row[name] for row in rows if row[name] is not None], np.float64)
|
|
result[name] = dict(mean=None if not len(values) else float(values.mean()),
|
|
variance_ddof0=None if not len(values) else float(values.var(ddof=0)),
|
|
valid_comparisons=int(len(values)), undefined_comparisons=int(len(rows) - len(values)))
|
|
return result
|
|
|
|
|
|
def _view(mode, records, positions):
|
|
arrays = {(odor, seed): records[(mode, odor, seed)]['counts'][positions]
|
|
for odor in ODORS for seed in SEEDS}
|
|
within, between, pairs = [], [], []
|
|
for odor in ODORS:
|
|
for first_seed, second_seed in itertools.combinations(SEEDS, 2):
|
|
within.append(dict(odor_index=odor, seed_a=first_seed, seed_b=second_seed,
|
|
**_pair_metrics(arrays[(odor, first_seed)], arrays[(odor, second_seed)])))
|
|
for first_odor, second_odor in itertools.combinations(ODORS, 2):
|
|
pair_between = []
|
|
for seed in SEEDS:
|
|
row = dict(odor_index_a=first_odor, odor_index_b=second_odor, seed=seed,
|
|
**_pair_metrics(arrays[(first_odor, seed)], arrays[(second_odor, seed)]))
|
|
pair_between.append(row)
|
|
between.append(row)
|
|
# Original pair-specific comparison: 10 within-seed pairs for each
|
|
# member odor, not a baseline pooled across all six odors.
|
|
pair_within = [row for row in within if row['odor_index'] in (first_odor, second_odor)]
|
|
within_summary, between_summary = _summary(pair_within), _summary(pair_between)
|
|
differences = {name: None if (within_summary[name]['mean'] is None or
|
|
between_summary[name]['mean'] is None) else
|
|
float(between_summary[name]['mean'] - within_summary[name]['mean'])
|
|
for name in METRICS}
|
|
pairs.append(dict(odor_index_a=first_odor, odor_index_b=second_odor,
|
|
within_comparison_count=len(pair_within), between_comparison_count=len(pair_between),
|
|
within=within_summary, between=between_summary,
|
|
mean_between_minus_within=differences,
|
|
within_comparisons=pair_within, between_comparisons=pair_between))
|
|
_require(len(within) == 60 and len(between) == 75 and len(pairs) == 15 and
|
|
all(row['within_comparison_count'] == 20 and row['between_comparison_count'] == 5
|
|
for row in pairs), 'Internal fixed comparison inventory changed')
|
|
return dict(dimensions=int(len(positions)), applicable=bool(len(positions)),
|
|
column_positions=[int(value) for value in positions],
|
|
within_comparison_count=len(within), between_comparison_count=len(between), pair_count=len(pairs),
|
|
within=_summary(within), between=_summary(between),
|
|
within_comparisons=within, between_comparisons=between, pairs=pairs)
|
|
|
|
|
|
def aggregate_structure(records):
|
|
"""Describe precisely 60 fixed CAL records, without selecting results.
|
|
|
|
Each record has exactly mode, odor_index, seed, counts:int32[686] and
|
|
design_mask:bool[686]. There are two named modes, six odors (0..5), five
|
|
seeds (51101..51105), and no repeat records. Every odor's design mask must
|
|
be identical across seeds AND modes. No caller-owned array is modified.
|
|
|
|
Undefined cosine (zero norm), Pearson (constant vector), or Jaccard (empty
|
|
active union) is null with explicit valid/undefined counts. An empty
|
|
common-nondesign population is nonapplicable, with every metric null.
|
|
"""
|
|
_require(isinstance(records, list) and len(records) == 60,
|
|
'Exactly 60 base CAL records are required; technical repeats must be excluded')
|
|
indexed, masks = {}, {}
|
|
for position, record in enumerate(records):
|
|
_require(isinstance(record, dict) and set(record) == FIELDS,
|
|
'Record schema must contain exactly ' + ', '.join(sorted(FIELDS)))
|
|
mode = record['mode']
|
|
_require(isinstance(mode, str) and mode in MODES, 'Unexpected CAL mode')
|
|
odor = _integer(record['odor_index'], 'odor_index')
|
|
seed = _integer(record['seed'], 'seed')
|
|
_require(odor in ODORS and seed in SEEDS, 'Odor index or seed outside the fixed CAL inventory')
|
|
key = mode, odor, seed
|
|
_require(key not in indexed, 'Duplicate CAL identity: ' + str(key))
|
|
counts = _array(record['counts'], np.int32, 'counts')
|
|
design = _array(record['design_mask'], np.bool_, 'design_mask')
|
|
_require(bool(np.all(counts >= 0)), 'Spike counts must be nonnegative')
|
|
if odor in masks:
|
|
_require(np.array_equal(design, masks[odor]),
|
|
'An odor design mask changed across seed or mode: ' + str(odor))
|
|
else:
|
|
masks[odor] = design.copy()
|
|
indexed[key] = dict(counts=counts, design_mask=design)
|
|
expected = set(itertools.product(MODES, ODORS, SEEDS))
|
|
_require(set(indexed) == expected, 'Incomplete fixed CAL identity set')
|
|
union = np.logical_or.reduce([masks[odor] for odor in ODORS])
|
|
primary = np.arange(N_ALPN, dtype=np.int32)
|
|
common = np.flatnonzero(~union).astype(np.int32)
|
|
return dict(schema='phase4ai-alpn-structure-v1', record_count=60, ALPN_dimensions=N_ALPN,
|
|
design=dict(identical_across_seeds_and_modes=True,
|
|
design_neuron_count_by_odor={str(odor): int(masks[odor].sum()) for odor in ODORS},
|
|
design_union_count=int(union.sum()), common_non_design_count=int(len(common)),
|
|
common_non_design_column_positions=[int(value) for value in common],
|
|
common_non_design_definition='Fixed complement of the union of all six CAL odor design masks; same columns in all modes/seeds'),
|
|
modes={mode: dict(trials=30, fixed_686=_view(mode, indexed, primary),
|
|
common_non_design=_view(mode, indexed, common)) for mode in MODES},
|
|
protocol=dict(odor_indices=list(ODORS), seeds=list(SEEDS), variance_ddof=0,
|
|
within='All 10 distinct seed pairs within each of six odors: 60 comparisons per mode',
|
|
between='Same-seed pairs for each of 15 odor pairs: 75 comparisons per mode',
|
|
pair_baseline='Only each odor pair: 10+10 within comparisons and five same-seed between comparisons',
|
|
counts='Original nonnegative int32 spike counts; no normalization, fitting or resampling',
|
|
undefined_cosine='null if either vector has zero norm',
|
|
undefined_Pearson='null if either vector is constant, including dimension one',
|
|
undefined_Jaccard='null if active union is empty',
|
|
empty_population='All metrics null, applicable=false; no fabricated zero distance',
|
|
clipping='Only cosine/correlation numerical roundoff is clipped to their mathematical [-1,1] range',
|
|
technical_repeats_included=False, population_columns='Caller-supplied fixed canonical ALPN order; positions are not neuron_idx'),
|
|
decision_gate_applied=False, learning_claimed=False, absence_of_encoding_claimed=False,
|
|
limitations=['Descriptive CAL structure only, not a held-out or learning result',
|
|
'Shared activity can coexist with odor-specific information',
|
|
'Active-set Jaccard can saturate even when spike-count patterns differ',
|
|
'Valid metric counts vary when a metric is undefined; null is not zero similarity',
|
|
'No KC sparsity gate is transferred to the ALPN population'])
|